hadoop secondary sort exercise 2 ( 보조 정렬 실습 2 )

업으로 하는게 아니니 가끔 해보고 있다.

그런데, 할때마다 조금씩 달라지는 기묘한 일이..

윈도우환경 이클립스에서 작업한 것이다.

 

FlightAnalysis.java

다운로드

 

package com.dbility.hadoop.execise;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 *
 * Description
 *
 *
 * @author hyperrookie@gmail.com
 *
 * @version 1.0.0
 * @date 2017. 2. 21.
 */
public class FlightAnalysis extends Configured implements Tool {

	public int run(String[] args) throws Exception {

		String[] remainArgs = new GenericOptionsParser(getConf(), args)
				.getRemainingArgs();

		if (remainArgs.length != 2) {
			System.out
					.println("Usage : hadoop jar jarName [mainClass] <input> <output>");
			return -1;
		}

		Job job = new Job(getConf());

		job.setJobName("FlightAnalysis");
		job.setJarByClass(FlightAnalysis.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setMapperClass(AsaMapper.class);
		job.setMapOutputKeyClass(CompositeKey.class);
		job.setMapOutputValueClass(IntWritable.class);

		job.setCombinerClass(AsaReducer.class);

		job.setNumReduceTasks(1);
		job.setPartitionerClass(AsaPartitioner.class);

		job.setGroupingComparatorClass(AsaGroupingComparator.class);
		job.setSortComparatorClass(AsaSortComparator.class);

		job.setReducerClass(AsaReducer.class);
		job.setOutputKeyClass(CompositeKey.class);
		job.setOutputValueClass(IntWritable.class);

		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.addInputPath(job, new Path(remainArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(remainArgs[1]));

		FileSystem hdfs = FileSystem.get(getConf());
		Path path = new Path(remainArgs[1]);

		if ( hdfs.exists(path) ) hdfs.delete(path, true);

		return job.waitForCompletion(true) ? 0 : -2;
	}

	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();

		conf.set("fs.default.name", "file:///");
		conf.set("mapred.job.tracker", "local");
		conf.set("fs.file.impl",
				"com.dbility.hadoop.execise.FlightAnalysis$WindowsLocalFileSystem");
		conf.set(
				"io.serializaitions",
				"org.apache.hadoop.io.serializer.JavaSerialization,org.apache.hadoop.io.serializer.WritableSerialization");

		args = new String[]{"d:/hadoop_test/*.csv","d:/hadoop_test/output"};

		int result = ToolRunner.run(conf, new FlightAnalysis(), args);

		System.exit(result);
	}

	public static class WindowsLocalFileSystem extends LocalFileSystem {

		public WindowsLocalFileSystem() {
			super();
		}

		@Override
		public boolean mkdirs(final Path path, final FsPermission permission)
				throws IOException {

			final boolean result = super.mkdirs(path);
			this.setPermission(path, permission);
			return result;
		}

		@Override
		public void setPermission(final Path path, final FsPermission permission)
				throws IOException {
			try {
				super.setPermission(path, permission);
			} catch (final IOException ioe) {
				System.err.println(ioe.getMessage());
			}
		}

	}

	public static class Parser {

		private int year;
		private int month;
		private int arrivalDelayTime = 0;
		private boolean arrivalDelayAvailable = false;


		public Parser(Text value) {
			String[] columns = value.toString().split(",");

			this.year = Integer.parseInt(columns[0]);
			this.month = Integer.parseInt(columns[1]);

			if ( !"NA".equals(columns[14]) ) {
				this.arrivalDelayAvailable = true;
				this.arrivalDelayTime = Integer.parseInt(columns[14]);
			}
		}

		public int getYear() {
			return year;
		}

		public int getMonth() {
			return month;
		}

		public int getArrivalDelayTime() {
			return arrivalDelayTime;
		}

		public boolean isArrivalDelayAvailable() {
			return arrivalDelayAvailable;
		}
	}

	public static class CompositeKey implements WritableComparable<CompositeKey> {

		private String year;
		private Integer month;

		public CompositeKey() {

		}

		public String getYear() {
			return year;
		}

		public void setYear(String year) {
			this.year = year;
		}

		public int getMonth() {
			return month;
		}

		public void setMonth(Integer month) {
			this.month = month;
		}

		public void readFields(DataInput in) throws IOException {
			this.year = WritableUtils.readString(in);
			this.month = in.readInt();
		}

		public void write(DataOutput out) throws IOException {
			WritableUtils.writeString(out, this.year);
			out.writeInt(this.month);
		}

		public int compareTo(CompositeKey o) {

			int result = this.year.compareTo(o.getYear());

			if ( result == 0 ){
				result = this.month.compareTo(o.getMonth());
			}

			return result;
		}

		@Override
		public String toString() {
			return (new StringBuilder()).append(year).append("\t").append(month).toString();
		}
	}

	public static class AsaMapper extends Mapper<LongWritable, Text, CompositeKey, IntWritable> {

		private CompositeKey outputKey = new CompositeKey();
		private static IntWritable outputValue = new IntWritable(1);

		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {

			Parser parser = new Parser(value);

			if ( parser.isArrivalDelayAvailable() && parser.getArrivalDelayTime() > 0 ) {

				outputKey.setYear(Integer.toString(parser.getYear()));
				outputKey.setMonth(parser.getMonth());

				context.write(outputKey, outputValue);

			}

		}

	}

	public static class AsaPartitioner extends Partitioner<CompositeKey, IntWritable> {

		@Override
		public int getPartition(CompositeKey key, IntWritable value, int numReduceTasks) {
			return key.getYear().hashCode() % numReduceTasks;
		}

	}

	public static class AsaGroupingComparator extends WritableComparator {

		public AsaGroupingComparator() {
			super(CompositeKey.class,true);
		}

		@SuppressWarnings("rawtypes")
		@Override
		public int compare(WritableComparable a, WritableComparable b) {

			CompositeKey k1 = (CompositeKey)a;
			CompositeKey k2 = (CompositeKey)b;

			return k1.getYear().compareTo(k2.getYear());
		}
	}


	public static class AsaSortComparator extends WritableComparator {

		public AsaSortComparator() {
			super(CompositeKey.class,true);
		}

		@SuppressWarnings("rawtypes")
		@Override
		public int compare(WritableComparable a, WritableComparable b) {

			CompositeKey k1 = (CompositeKey)a;
			CompositeKey k2 = (CompositeKey)b;

			int cmp = k1.getYear().compareTo(k2.getYear());

			if ( cmp != 0 ) {
				return cmp;
			}

			return k1.getMonth() == k2.getMonth() ? 0 : (k1.getMonth() < k2.getMonth() ? -1 : 1);

		}

	}

	public static class AsaReducer extends Reducer<CompositeKey, IntWritable, CompositeKey, IntWritable> {


		private CompositeKey outputKey = new CompositeKey();
		private IntWritable outputValue = new IntWritable();
		private int sum;

		@Override
		protected void reduce(CompositeKey key, Iterable<IntWritable> values, Context context)
				throws IOException, InterruptedException {

			sum = 0;
			int preMonth = key.getMonth();

			for (IntWritable value : values) {

				if ( preMonth != key.getMonth() ) {

					outputKey.setYear(key.getYear());
					outputKey.setMonth(preMonth);
					outputValue.set(sum);

					context.write(outputKey, outputValue);

					sum = 0;
				}

				sum+=value.get();
				preMonth = key.getMonth();
			}

			if ( preMonth == key.getMonth() ) {

				outputKey.setYear(key.getYear());
				outputKey.setMonth(preMonth);
				outputValue.set(sum);

				context.write(outputKey, outputValue);
			}

		}
	}

}