Hadoop实现Secondary Sort

在Hadoop中每个reduce的输入的key都是有序的，而value则是无序的。而且同一个job运行多次，由于map完成顺序不同，reduce收到的value顺序是不固定的。那如何才能实现reduce收到有序的value呢？这就需要Secondary Sort。

Secondary Sort要解决的问题：reduce收到的value有序。

这里举一个场景，来说明Secondary Sort是如何实现的。假设我们有若干公司若干部门的人数，数据样例如下：

公司名   部门的人数

Taobao 52
Taobao 31
Taobao 67
Alipay 10
Alipay 36
Alipay 29
B2B 120
B2B 72
Aliyun 13
Aliyun 32
Aliyun 3

我们想知道每个公司的最大部门（人数最多）的人数。即希望先按公司名group，然后对group内的人数降序排列，最后取每个group的第一个即可。

由于reduce收到的value是无序的，所以要对value进行排序，首先需要将value封装到key里面。即需要自定义key的类型，代码如下：

1. import java.io.DataInput;  
2. import java.io.DataOutput;  
3. import java.io.IOException;  
4.   
5. import org.apache.hadoop.io.IntWritable;  
6. import org.apache.hadoop.io.Text;  
7. import org.apache.hadoop.io.WritableComparable;  
8.   
9. public class MyKey implements WritableComparable<MyKey> {  
10.   public final Text first;  
11.   public final IntWritable second;  
12.   
13.   public MyKey() {  
14.     first = new Text();  
15.     second = new IntWritable();  
16.   }  
17.   
18.   public MyKey(Text first, IntWritable second) {  
19.     this.first = first;  
20.     this.second = second;  
21.   }  
22.   
23.   @Override  
24.   public void write(DataOutput out) throws IOException {  
25.     first.write(out);  
26.     second.write(out);  
27.   }  
28.   
29.   @Override  
30.   public void readFields(DataInput in) throws IOException {  
31.     first.readFields(in);  
32.     second.readFields(in);  
33.   }  
34.   
35.   @Override  
36.   public String toString() {  
37.     return first + "\t" + second;  
38.   }  
39.   
40.   @Override  
41.   public int compareTo(MyKey tp) {  
42.     int cmp = first.compareTo(tp.first);  
43.     if (cmp != 0) {  
44.       return cmp;  
45.     }  
46.     return -second.compareTo(tp.second);  
47.   }  
48. }

更多Hadoop相关信息见Hadoop 专题页面 http://www.linuxidc.com/topicnews.aspx?tid=13