MapReduce中的二次排序

20    21
50    51
50    52
50    53
50    54
60    51
60    53
60    52
60    56
60    57
70    58
60    61
70    54
70    55
70    56
70    57
70    58  

/**
 * 把第一列整数和第二列作为类的属性，并且实现WritableComparable接口
 */
public static class IntPair implements WritableComparable<IntPair> {
  private int first = 0;
  private int second = 0;

  public void set(int left, int right) {
    first = left;
    second = right;
  }
  public int getFirst() {
    return first;
  }
  public int getSecond() {
    return second;
  }

  @Override
  public void readFields(DataInput in) throws IOException {
    first = in.readInt();
    second = in.readInt();
  }
  @Override
  public void write(DataOutput out) throws IOException {
    out.writeInt(first);
    out.writeInt(second);
  }
  @Override
  public int hashCode() {
    return first+"".hashCode() + second+"".hashCode();
  }
  @Override
  public boolean equals(Object right) {
    if (right instanceof IntPair) {
      IntPair r = (IntPair) right;
      return r.first == first && r.second == second;
    } else {
      return false;
    }
  }
  //这里的代码是关键，因为对key排序时，调用的就是这个compareTo方法
  @Override
  public int compareTo(IntPair o) {
    if (first != o.first) {
      return first - o.first;
    } else if (second != o.second) {
      return second - o.second;
    } else {
      return 0;
    }
  }
}  

/**
 * 在分组比较的时候，只比较原来的key，而不是组合key。
 */
public static class GroupingComparator implements RawComparator<IntPair> {
  @Override
  public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
    return WritableComparator.compareBytes(b1, s1, Integer.SIZE/8, b2, s2, Integer.SIZE/8);
  }

  @Override
  public int compare(IntPair o1, IntPair o2) {
    int first1 = o1.getFirst();
    int first2 = o2.getFirst();
    return first1 - first2;
  }
}  

public static class MapClass extends Mapper<LongWritable, Text, IntPair, IntWritable> {

  private final IntPair key = new IntPair();
  private final IntWritable value = new IntWritable();

  @Override
  public void map(LongWritable inKey, Text inValue, 
                  Context context) throws IOException, InterruptedException {
    StringTokenizer itr = new StringTokenizer(inValue.toString());
    int left = 0;
    int right = 0;
    if (itr.hasMoreTokens()) {
      left = Integer.parseInt(itr.nextToken());
      if (itr.hasMoreTokens()) {
        right = Integer.parseInt(itr.nextToken());
      }
      key.set(left, right);
      value.set(right);
      context.write(key, value);
    }
  }
}

public static class Reduce extends Reducer<IntPair, IntWritable, Text, IntWritable> {
  private static final Text SEPARATOR = new Text("------------------------------------------------");
  private final Text first = new Text();

  @Override
  public void reduce(IntPair key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    context.write(SEPARATOR, null);
    first.set(Integer.toString(key.getFirst()));
    for(IntWritable value: values) {
      context.write(first, value);
    }
  }
}  

public static void main(String[] args) throws Exception {
  Configuration conf = new Configuration();

  final FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop2:9000"), conf);
  fileSystem.delete(new Path(OUTPUT_PATH), true);

  Job job = new Job(conf, "secondary sort");
  job.setJarByClass(SecondarySortApp.class);
  job.setMapperClass(MapClass.class);
  job.setReducerClass(Reduce.class);

  job.setGroupingComparatorClass(GroupingComparator.class);

  job.setMapOutputKeyClass(IntPair.class);
  job.setMapOutputValueClass(IntWritable.class);

  job.setOutputKeyClass(Text.class);
  job.setOutputValueClass(IntWritable.class);

  FileInputFormat.addInputPath(job, new Path(INPUT_PATH));
  FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));
  System.exit(job.waitForCompletion(true) ? 0 : 1);
}  

------------------------------------------------
20    21
------------------------------------------------
50    51
50    52
50    53
50    54
------------------------------------------------
60    51
60    52
60    53
60    56
60    57
60    61
------------------------------------------------
70    54
70    55
70    56
70    57
70    58
70    58