MapReduce编程小结

package examples; 

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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
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.partition.HashPartitioner;

public class GroupApp {
	static final String INPUT_PATH = "hdfs://192.168.2.100:9000/hello";
	static final String OUTPUT_PATH = "hdfs://192.168.2.100:9000/out";
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);
		final Path outPath = new Path(OUTPUT_PATH);
		if(fileSystem.exists(outPath)) {
			fileSystem.delete(outPath, true);
		}
		
		final Job job = new Job(conf, GroupApp.class.getSimpleName());
		job.setJarByClass(GroupApp.class);
		FileInputFormat.setInputPaths(job, INPUT_PATH);
		
		job.setInputFormatClass(TextInputFormat.class);
		
		job.setMapperClass(MyMapper.class);
		job.setMapOutputKeyClass(NewK2.class);
		job.setMapOutputValueClass(LongWritable.class);
		
		job.setPartitionerClass(MyPartitoner.class);
		job.setNumReduceTasks(3);
		
		job.setGroupingComparatorClass(MyGroupingComparator.class);
		
		job.setReducerClass(MyReducer.class);
		
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(LongWritable.class);
		
		FileOutputFormat.setOutputPath(job, outPath);
		
		job.waitForCompletion(true);
	}
	
	static class MyPartitoner extends HashPartitioner<NewK2, LongWritable> {
		  public int getPartition(NewK2 key, LongWritable value, int numReduceTasks) {
			  System.out.println("the getPartition() is called...");
			  if(key.first == 1) {
				  return 0 % numReduceTasks;
			  }
			  else if(key.first == 2) {
				  return 1 % numReduceTasks;
			  }
			  else {
				  return 2 % numReduceTasks;
			  }
		  }
	}
	
	static class NewK2 implements WritableComparable<NewK2> {
		Long first = 0L;
		Long second = 0L;
		
		public NewK2(){}
		
		public NewK2(long first, long second) {
			this.first = first;
			this.second = second;
		}

		public void write(DataOutput out) throws IOException {
			out.writeLong(first);
			out.writeLong(second);
		}

		public void readFields(DataInput in) throws IOException {
			first = in.readLong();
			second = in.readLong();
		}

		public int compareTo(NewK2 o) {
			System.out.println("the compareTo() is called...");
			
			final long minus = this.first - o.first;
			if(minus != 0) {
				return (int)minus;
			}
			return (int) (this.second - o.second);
		}
	}
	
	static class MyGroupingComparator implements RawComparator<NewK2> {
		public int compare(NewK2 o1, NewK2 o2) {
	//		System.out.println("the compare() is called...");
			return (int) (o1.first - o2.first);
		}

		public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
			System.out.println("the compare() is called...");
			return WritableComparator.compareBytes(b1, s1, 8, b2, s2, 8);
		}
	}
	
	static class MyMapper extends Mapper<LongWritable, Text, NewK2, LongWritable> {
		protected void map(LongWritable k1, Text v1, Context ctx) throws IOException, InterruptedException {
			final String[] splited = v1.toString().split("\t");
			
			System.out.println("the map() is called...");
			
			NewK2 k2 = new NewK2(Integer.parseInt(splited[0]), Integer.parseInt(splited[1]));
			LongWritable v2 = new LongWritable(Long.parseLong((splited[1])));
			ctx.write(k2, v2);
//			System.out.println("the real map output...");
//			System.out.println("<"+k2.first+","+v2+">");
		}
	}
	static class MyReducer extends Reducer<NewK2, LongWritable, LongWritable, LongWritable> {
		long v3 = 0;
		protected void reduce(NewK2 k2, Iterable<LongWritable> v2s, Context ctx) throws IOException, InterruptedException {
			System.out.println("the reduce() is called...");
			for(LongWritable secend : v2s) {
				v3 = secend.get();
				System.out.println("<"+k2.first+","+k2.second+">, "+v3+"");
			}
			System.out.println("--------------------------------------------");
			System.out.println("the real reduce output...");
			System.out.println("<"+k2.first+","+v3+">");
			ctx.write(new LongWritable(k2.first), new LongWritable(v3));
			System.out.println("--------------------------------------------");
		}
	}
}