MapReduce二次排序

默认情况下，Map 输出的结果会对 Key 进行默认的排序，但是有时候需要对 Key 排序的同时再对 Value 进行排序，这时候就要用到二次排序了。下面让我们来介绍一下什么是二次排序。

二次排序原理

我们把二次排序主要分为以下几个阶段。

Map 起始阶段

在Map阶段，使用 job.setInputFormatClass() 定义的 InputFormat ，将输入的数据集分割成小数据块 split，同时 InputFormat 提供一个 RecordReader的实现。本课程中使用的是 TextInputFormat，它提供的 RecordReader 会将文本的行号作为 Key，这一行的文本作为 Value。这就是自定义 Mapper 的输入是 < LongWritable,Text> 的原因。然后调用自定义 Mapper 的map方法，将一个个< LongWritable,Text>键值对输入给 Mapper 的map方法。

Map 最后阶段

在 Map 阶段的最后，会先调用 job.setPartitionerClass() 对这个 Mapper 的输出结果

进行分区，每个分区映射到一个Reducer。每个分区内又调用

job.setSortComparatorClass() 设置的 Key 比较函数类排序。可以看到，这本身就是一个二次排序。如果没有通过 job.setSortComparatorClass() 设置 Key 比较函数类，则使用 Key 实现的 compareTo() 方法。我们既可以使用 IntPair 实现的 compareTo() 方法，也可以专门定义 Key 比较函数类。

Reduce 阶段

在 Reduce 阶段，reduce() 方法接受所有映射到这个 Reduce 的 map 输出后，也是会

调用 job.setSortComparatorClass()方法设置的 Key 比较函数类，对所有数据进行排序。然

后开始构造一个 Key 对应的 Value 迭代器。这时就要用到分组，使用

job.setGroupingComparatorClass()方法设置分组函数类。只要这个比较器比较的两个 Key

相同，它们就属于同一组，它们的 Value 放在一个 Value 迭代器，而这个迭代器的 Key 使用属于同一个组的所有Key的第一个Key。最后就是进入 Reducer 的 reduce() 方法， reduce() 方法的输入是所有的 Key 和它的 Value 迭代器，同样注意输入与输出的类型必须与自定义的 Reducer 中声明的一致。

接下来我们通过数据示例，可以很直观的了解二次排序的原理。

输入数据

输入文件sort.txt内容为：

输出文件的内容（从小到大排序）如下：

从输出的结果可以看出Key实现了从小到大的排序，同时相同Key的Value也实现了从小到大的排序，这就是二次排序的结果。

二次排序的具体流程

在 MapReduce 中，所有的 Key 是需要被比较和排序的，而且是二次，先根据Partitioner，再根据大小。而本例中也是要比较两次。先按照第一字段排序，然后再对第一字段相同的按照第二字段排序。根据这一点，我们可以构造一个复合类IntPair ，它有两个字段，先利用分区对第一字段排序，再利用分区内的比较对第二字段排序。二次排序的流程分为以下几步。

1、自定义 key

所有自定义的 key 应该实现接口 WritableComparable，因为它是可序列化的并且可

比较的。WritableComparable 的内部方法如下所示。

[java] 

1. //反序列化，从流中的二进制转换成IntPair public void readFields(DataInput in) throws IOException  
2.   
3. //序列化，将IntPair转化成使用流传送的二进制 public void write(DataOutput out)  
4.   
5. //key的比较  
6.   
7. public int compareTo(IntPair o)  
8.   
9. //默认的分区类 HashPartitioner，使用此方法 public int hashCode()  
10.   
11. //默认实现  
12.   
13. public boolean equals(Object right)  

2、自定义分区

自定义分区函数类 FirstPartitioner，是 key 的第一次比较，完成对所有 key 的排序。

[java] 

1. public static class FirstPartitioner extends Partitioner< IntPair,IntWritable>  
2.   
3. 在 job 中使用 setPartitionerClasss()方法设置 Partitioner。  
4.   
5. job.setPartitionerClasss(FirstPartitioner.Class);  

3、Key 的比较类

这是 Key 的第二次比较，对所有的 Key 进行排序，即同时完成IntPair中的first和second排序。该类是一个比较器，可以通过两种方式实现。

1) 继承 WritableComparator。

[java] 

1. public static class KeyComparator extends WritableComparator  
2.   
3. 必须有一个构造函数，并且重载以下方法。  
4.   
5. public int compare(WritableComparable w1, WritableComparable w2)  

2) 实现接口 RawComparator。

[java] 

1. 上面两种实现方式，在 Job 中，可以通过setSortComparatorClass()方法来设置Key的比较类。  
2.   
3. job.setSortComparatorClass(KeyComparator.Class);  
4.   
5.    

注意：如果没有使用自定义的 SortComparator 类，则默认使用 Key 中compareTo()

方法对 Key 排序分组。

4、定义分组类函数

在 Reduce 阶段，构造一个与 Key 相对应的 Value 迭代器的时候，只要 first 相同就属于同一个组，放在一个 Value 迭代器。定义这个比较器，可以有两种方式。

1) 继承 WritableComparator。

[java] 

1. public static class GroupingComparator extends WritableComparator  

必须有一个构造函数，并且重载以下方法。

[java] 

1. public int compare(WritableComparable w1, WritableComparable w2)  

2) 实现接口 RawComparator。

上面两种实现方式，在 Job 中，可以通过 setGroupingComparatorClass()方法来设置分组类。

[java] 

1. job.setGroupingComparatorClass(GroupingComparator.Class);  

另外注意的是，如果reduce的输入与输出不是同一种类型，则 Combiner和Reducer不能共用 Reducer 类，因为 Combiner 的输出是 reduce 的输入。除非重新定义一个

Combiner。

代码实现

Hadoop 的 example 包中自带了一个 MapReduce 的二次排序算法，下面这个示例对 example 包中的二次排序源码的改进。我们按照以下几步完成二次排序：

第一步：自定义IntPair类，将示例数据中的key/value封装成一个整体作为Key，同时实现 WritableComparable 接口并重写其方法。

[java]

1. /** 
2.  
3. * 自己定义的key类应该实现WritableComparable接口 */  
4.   
5. public  class IntPair implements WritableComparable<IntPair>{ int first;//第一个成员变量 int second;//第二个成员变量  
6.   
7. public void set(int left, int right){ first = left;  
8.   
9. second = right;  
10.   
11. }  
12.   
13. public int getFirst(){ return first;  
14. }  
15.   
16. public int getSecond(){ return second;  
17. }  
18.   
19. @Override  
20.   
21. //反序列化，从流中的二进制转换成IntPair public void readFields(DataInput in) throws IOException{  
22.   
23. first = in.readInt(); second = in.readInt();  
24. }  
25.   
26. @Override  
27.   
28. //序列化，将IntPair转化成使用流传送的二进制 public void write(DataOutput out) throws IOException{  
29.   
30. out.writeInt(first);  
31.   
32. out.writeInt(second);  
33.   
34. }  
35.   
36. @Override //key的比较  
37.   
38. public int compareTo(IntPair o)  
39.   
40. {  
41.   
42. // TODO Auto‐generated method stub if (first != o.first){  
43.   
44. return first < o.first ? -1 : 1; }else if (second != o.second){  
45.   
46. return second < o.second ? -1 : 1; }else{  
47.   
48. return 0;  
49.   
50. }  
51.   
52. }  
53.   
54. @Override  
55.   
56. public int hashCode(){  
57.   
58. return first * 157 + second;  
59.   
60. }  
61.   
62. @Override  
63.   
64. public boolean equals(Object right){ if (right == null)  
65.   
66. return false; if (this == right) return true;  
67.   
68. if (right instanceof IntPair){ IntPair r = (IntPair) right;  
69.   
70. return r.first == first && r.second == second; }else{  
71.   
72. return false;  
73.   
74. }  
75.   
76. }  
77.   
78. }  
79.   
80. 第二步：自定义分区函数类FirstPartitioner，根据 IntPair 中的first实现分区。  
81.   
82.   
83. /** 
84.  
85. * 分区函数类。根据first确定Partition。 */  
86.   
87. public static class FirstPartitioner extends Partitioner<IntPair, IntWritable>{ @Override  
88.   
89. public int getPartition(IntPair key, IntWritable value,int numPartitions){ return Math.abs(key.getFirst() * 127) % numPartitions;  
90. }  
91.   
92. }  
93.   
94. 第三步：自定义 SortComparator 实现 IntPair 类中的first和second排序。本课程中没  
95.   
96. 有使用这种方法，而是使用 IntPair 中的compareTo()方法实现的。  
97.   
98. 第四步：自定义 GroupingComparator 类，实现分区内的数据分组。  
99.   
100. /** *继承WritableComparator */  
101.   
102. public static class GroupingComparator extends WritableComparator{ protected GroupingComparator(){  
103. super(IntPair.class, true);  
104.   
105. }  
106.   
107. @Override  
108.   
109. //Compare two WritableComparables.  
110.   
111. public int compare(WritableComparable w1, WritableComparable w2){ IntPair ip1 = (IntPair) w1;  
112. IntPair ip2 = (IntPair) w2; int l = ip1.getFirst(); int r = ip2.getFirst();  
113.   
114. return l == r ? 0 : (l < r ? -1 : 1);  
115.   
116. }  
117.   
118. }  
119.   
120. 第五步：编写 MapReduce 主程序实现二次排序。  
121.   
122.       
123. import  java.io.DataInput;  
124. import  java.io.DataOutput;  
125. import  java.io.IOException;  
126. import  java.util.StringTokenizer;  
127. import  org.apache.hadoop.conf.Configuration;  
128. import  org.apache.hadoop.fs.Path;  
129. import  org.apache.hadoop.io.IntWritable;  
130. import  org.apache.hadoop.io.LongWritable;  
131. import  org.apache.hadoop.io.Text;  
132. import  org.apache.hadoop.io.WritableComparable;  
133. import  org.apache.hadoop.io.WritableComparator;  
134. import  org.apache.hadoop.mapreduce.Job;  
135. import  org.apache.hadoop.mapreduce.Mapper;  
136. import  org.apache.hadoop.mapreduce.Partitioner;  
137. import  org.apache.hadoop.mapreduce.Reducer;  
138. import  org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
139. import  org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
140. import  org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
141. import  org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
142. public  class SecondarySort{  
143. //  自定义map  
144. public static class Map extends Mapper<LongWritable, Text, IntPair, IntWritable>{  
145.     private final IntPair intkey = new IntPair();  
146.     private final IntWritable intvalue = new IntWritable();  
147.     public void map(LongWritable key, Text value, Context context) throws IOException, Interrupted  
148.     String line = value.toString();  
149.     StringTokenizer tokenizer = new StringTokenizer(line);  
150.     int left = 0;  
151.     int right = 0;  
152.     if (tokenizer.hasMoreTokens()){  
153.     left = Integer.parseInt(tokenizer.nextToken());  
154.     if (tokenizer.hasMoreTokens())  
155.     right = Integer.parseInt(tokenizer.nextToken());  
156.     intkey.set(left, right);  
157.     intvalue.set(right);  
158.     context.write(intkey, intvalue);  
159.     }  
160.     }  
161. }     
162. //  自定义reduce  
163.   
164. public static class Reduce extends Reducer< IntPair, IntWritable, Text, IntWritable>{ private final Text left = new Text();  
165.   
166. public void reduce(IntPair key, Iterable< IntWritable> values,Context context) throws IOExcepti left.set(Integer.toString(key.getFirst()));  
167. for (IntWritable val : values){ context.write(left, val);  
168. }  
169.   
170. }  
171.   
172. }  
173.   
174. /** 
175.  
176. * @param args */  
177. public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundExce // TODO Auto‐generated method stub  
178.   
179. Configuration conf = new Configuration();  
180.   
181. Job job = new Job(conf, "secondarysort"); job.setJarByClass(SecondarySort.class);  
182.   
183. FileInputFormat.setInputPaths(job, new Path(args[0]));//输入路径  
184.   
185. FileOutputFormat.setOutputPath(job, new Path(args[1]));//输出路径  
186.   
187. job.setMapperClass(Map.class);// Mapper job.setReducerClass(Reduce.class);// Reducer  
188.   
189. job.setPartitionerClass(FirstPartitioner.class);// 分区函数//job.setSortComparatorClass(KeyComparator.Class);//本课程并没有自定义SortComparator，而是使用In job.setGroupingComparatorClass(GroupingComparator.class);// 分组函数  
190.   
191.   
192.   
193.   
194. job.setMapOutputKeyClass(IntPair.class);  
195.   
196. job.setMapOutputValueClass(IntWritable.class);  
197.   
198. job.setOutputKeyClass(Text.class);  
199.   
200. job.setOutputValueClass(IntWritable.class);  
201.   
202. job.setInputFormatClass(TextInputFormat.class);  
203.   
204. job.setOutputFormatClass(TextOutputFormat.class);  
205.   
206. System.exit(job.waitForCompletion(true) ? 0 : 1);  
207.   
208. }  
209.   
210. }