MapReduce编程实例之自定义排序

任务描述：

给出一组数据，自定义排序的样式，第一列降序，相同时第二列升序

example Data：

2013 1
2013 5
2014 5
2014 8
2015 9
2015 4

Code:

package mrTest; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.io.WritableComparable; 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.output.FileOutputFormat; public class zidingyiSort { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { // TODO Auto-generated method stub Job job = new Job(); job.setJarByClass(zidingyiSort.class); // 1 FileInputFormat.setInputPaths(job, new Path(args[0])); // 2 job.setMapperClass(Map.class); job.setMapOutputKeyClass(MyK2.class); job.setMapOutputValueClass(LongWritable.class); // 3 // 4 // 5 job.setNumReduceTasks(1); // 6 job.setReducerClass(Reduce.class); job.setOutputKeyClass(LongWritable.class); job.setOutputValueClass(LongWritable.class); // 7 FileOutputFormat.setOutputPath(job, new Path(args[1])); // 8 System.exit(job.waitForCompletion(true)? 0 :1 ); } public static class Map extends Mapper<Object, Text, MyK2, LongWritable>{ public void map(Object key, Text value, Context context) throws IOException, InterruptedException{ String line = value.toString(); String[] split = line.split("\t"); MyK2 my = new MyK2(Long.parseLong(split[0]), Long.parseLong(split[1])); context.write(my, new LongWritable(1)); } } public static class Reduce extends Reducer<MyK2, LongWritable, LongWritable, LongWritable>{ public void reduce(MyK2 key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException{ context.write(new LongWritable(key.myk2), new LongWritable(key.myv2)); } } public static class MyK2 implements WritableComparable<MyK2>{ public long myk2; public long myv2; MyK2(){} MyK2(long myk2, long myv2){ this.myk2 = myk2; this.myv2 = myv2; } @Override public void readFields(DataInput in) throws IOException { // TODO Auto-generated method stub this.myk2 = in.readLong(); this.myv2 = in.readLong(); } @Override public void write(DataOutput out) throws IOException { // TODO Auto-generated method stub out.writeLong(myk2); out.writeLong(myv2); } @Override public int compareTo(MyK2 myk2) { // TODO Auto-generated method stub //myk2之差>0 返回-1 <0 返回1 代表 myk2列降序 //myk2之差<0 返回-1 >0 返回1 代表 myk2列升序 long temp = this.myk2 - myk2.myk2; if(temp>0) return -1; else if(temp<0) return 1; //控制myv2升序 return (int)(this.myv2 - myk2.myv2); } } }

2015 4
2015 9
2014 5
2014 8
2013 1
2013 5