MapReduce按照兩個字段對數據進行排序

按照k2排序，要求k2必須是可以比較的，即必須實現WritableComparable接口。

但是如果還想讓別的字段(比如v2中的一些字段)參與排序怎么辦?

需要重新定義k2....把需要參與排序的字段都放到k2中.

這塊用代碼實現:

假如數據現在的結構是

3       3

3       2

3       1

2       2

2       1

1       1

看代碼:

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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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 TwoIntSortApp {
    
    public static void main(String[] args) throws Exception {
        Job job = Job.getInstance(new Configuration(), TwoIntSortApp.class.getSimpleName());
        job.setJarByClass(TwoIntSortApp.class);
        FileInputFormat.setInputPaths(job, args[0]);
        
        job.setMapperClass(TwoIntSortMapper.class);
        job.setMapOutputKeyClass(TwoInt.class);
        job.setMapOutputValueClass(NullWritable.class);
        
        job.setReducerClass(TwoIntSortReducer.class);
        job.setOutputKeyClass(TwoInt.class);
        job.setOutputValueClass(NullWritable.class);

        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.waitForCompletion(true);        
    }
    
    public static class TwoIntSortMapper extends Mapper<LongWritable, Text, TwoInt, NullWritable>{
        TwoInt k2 = new TwoInt();
        @Override
        protected void map(LongWritable key, Text value,
                Mapper<LongWritable, Text, TwoInt, NullWritable>.Context context)
                throws IOException, InterruptedException {
            String[] splited = value.toString().split("\t");
            k2.set(splited[0],splited[1]);
            context.write(k2, NullWritable.get());
            System.out.println("Mapper-----第一個數:"+k2.first+" 第二個數:"+k2.second);
        }
    }
    
    public static class TwoIntSortReducer extends Reducer<TwoInt, NullWritable, TwoInt, NullWritable>{
        int i=1;
        @Override
        protected void reduce(TwoInt k2, Iterable<NullWritable> arg1,
                Reducer<TwoInt, NullWritable, TwoInt, NullWritable>.Context context)
                throws IOException, InterruptedException {
            context.write(k2,NullWritable.get());
            System.out.println("調用次數"+(i++));
            System.out.println("Reducer-----第一個數:"+k2.first+" 第二個數:"+k2.second);
        }
    }
    
    public static class TwoInt implements WritableComparable<TwoInt>{
        int first;
        int second;
        public void write(DataOutput out) throws IOException {
            out.writeInt(first);
            out.writeInt(second);
        }
        
        public void set(String s1,String s2){
            this.first = Integer.parseInt(s1);
            this.second = Integer.parseInt(s2);
        }

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

        public int compareTo(TwoInt o) {
            int r1 = this.first - o.first;
            if(r1 < 0){
                return -1;
            }else if(r1 > 0){
                return 1;
            }
            int r2 = this.second - o.second;
            return  (r2 < 0 ? -1 : (r2 > 0 ? 1 : 0)); 
        }
        
        @Override
        public String toString() {
            return this.first+"\t"+this.second;
        }
    }
}

 //==============================================================

在job上設置Combiner類...

        
        job.setCombinerClass(TwoIntSortReducer.class);//設置Combiner類
        job.setGroupingComparatorClass(MyGroupingCompartor.class);//設置自定義的分組類

  

    public static class MyGroupingCompartor extends WritableComparator{
        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            TwoInt aa = (TwoInt)a; 
            TwoInt bb = (TwoInt)b; 
            return aa.first-bb.first<0?-1:(aa.first-bb.first>0?1:0);//只要是第一列相同的就認為是一個分組.
            /*
             * 1    1
             * 2    1
             * 2    2
             * 3    1
             * 3    2
             * 3    3
             * 這樣就分成了三組
             */
        }
    }