MapReduce排序之 二次排序

一：背景

Hadoop中虽然有自动排序和分组，由于自带的排序是按照Key进行排序的，有些时候，我们希望同时对Key和Value进行排序。自带的排序功能就无法满足我们了，还好Hadoop提供了一些组件可以让开发人员进行二次排序。

二：技术实现

我们先来看案例需求

#需求1： 首先按照第一列数字升序排列，当第一列数字相同时，第二列数字也升序排列(列之间用制表符\t隔开)

[java] view plain copy

1. 3   3
2. 3   2
3. 3   1
4. 2   2
5. 2   1
6. 1   1

MapReduce计算之后的结果应该是：

[java] view plain copy

1. 1   1
2. 2   1
3. 2   2
4. 3   1
5. 3   2
6. 3   3

#需求2：第一列不相等时，第一列按降序排列，当第一列相等时，第二列按升序排列

[java] view plain copy

1. 3   3
2. 3   2
3. 3   1
4. 2   2
5. 2   1
6. 1   1

MapReduce计算之后的结果应该是：

[java] view plain copy

1. 3   1
2. 3   2
3. 3   3
4. 2   1
5. 2   2
6. 1   1

下面是实现代码，实现两种需求的关键是compareTo()方法的实现不同：

[java] view plain copy

1. public class SecondSortTest {
2. // 定义输入路径
3. private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/data";
4. // 定义输出路径
5. private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";
6. public static void main(String[] args) {
7. try {
8. // 创建配置信息
9. Configuration conf = new Configuration();
10. /**********************************************/
11. //对Map端输出进行压缩
12. //conf.setBoolean("mapred.compress.map.output", true);
13. //设置map端输出使用的压缩类
14. //conf.setClass("mapred.map.output.compression.codec", GzipCodec.class, CompressionCodec.class);
15. //对reduce端输出进行压缩
16. //conf.setBoolean("mapred.output.compress", true);
17. //设置reduce端输出使用的压缩类
18. //conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class);
19. // 添加配置文件(我们可以在编程的时候动态配置信息，而不需要手动去改变集群)
20. /*
21. * conf.addResource("classpath://hadoop/core-site.xml");
22. * conf.addResource("classpath://hadoop/hdfs-site.xml");
23. * conf.addResource("classpath://hadoop/hdfs-site.xml");
24. */
25. // 创建文件系统
26. FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);
27. // 如果输出目录存在，我们就删除
28. if (fileSystem.exists(new Path(OUT_PATH))) {
29. fileSystem.delete(new Path(OUT_PATH), true);
30. }
31. // 创建任务
32. Job job = new Job(conf, SecondSortTest.class.getName());
33. //1.1   设置输入目录和设置输入数据格式化的类
34. FileInputFormat.setInputPaths(job, INPUT_PATH);
35. job.setInputFormatClass(TextInputFormat.class);
36. //1.2   设置自定义Mapper类和设置map函数输出数据的key和value的类型
37. job.setMapperClass(MySecondSortMapper.class);
38. job.setMapOutputKeyClass(CombineKey.class);
39. job.setMapOutputValueClass(LongWritable.class);
40. //1.3   设置分区和reduce数量(reduce的数量，和分区的数量对应，因为分区为一个，所以reduce的数量也是一个)
41. job.setPartitionerClass(HashPartitioner.class);
42. job.setNumReduceTasks(1);
43. //1.4   排序、分组
44. //1.5   归约
45. //2.1   Shuffle把数据从Map端拷贝到Reduce端。
46. //2.2   指定Reducer类和输出key和value的类型
47. job.setReducerClass(MySecondSortReducer.class);
48. job.setOutputKeyClass(LongWritable.class);
49. job.setOutputValueClass(LongWritable.class);
50. //2.3   指定输出的路径和设置输出的格式化类
51. FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));
52. job.setOutputFormatClass(TextOutputFormat.class);
53. // 提交作业 退出
54. System.exit(job.waitForCompletion(true) ? 0 : 1);
55. } catch (Exception e) {
56. e.printStackTrace();
57. }
58. }
59. public static class MySecondSortMapper extends Mapper<LongWritable, Text, CombineKey, LongWritable>{
60. //定义联合的key
61. private CombineKey combineKey = new CombineKey();
62. protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, CombineKey, LongWritable>.Context context) throws IOException,
63. InterruptedException {
64. //对输入的value进行切分
65. String[] splits = value.toString().split("\t");
66. //设置联合的key
67. combineKey.setComKey(Long.parseLong(splits[0]));
68. combineKey.setComVal(Long.parseLong(splits[1]));
69. //通过context写出去
70. context.write(combineKey, new LongWritable(Long.parseLong(splits[1])));
71. }
72. }
73. public static class MySecondSortReducer extends Reducer<CombineKey, LongWritable, LongWritable, LongWritable>{
74. @Override
75. protected void reduce(CombineKey combineKey, Iterable<LongWritable> values, Reducer<CombineKey, LongWritable, LongWritable, LongWritable>.Context context)
76. throws IOException, InterruptedException {
77. //因为输入的CombineKey已经排好序了，所有我们只要获取其中的两个成员变量写出去就可以了。values在这个例子中没有什么作用
78. context.write(new LongWritable(combineKey.getComKey()), new LongWritable(combineKey.getComVal()));
79. }
80. }
81. }
82. /**
83. * 重新组合成一个key，实现二次排序
84. * @author 廖*民
85. * time : 2015年1月18日下午7:27:52
86. * @version
87. */
88. class CombineKey implements WritableComparable<CombineKey>{
89. public long comKey;
90. public long comVal;
91. //必须提供无参构造函数，否则hadoop反射机制会出错
92. public CombineKey() {
93. }
94. //有参构造函数
95. public CombineKey(long comKey, long comVal) {
96. this.comKey = comKey;
97. this.comVal = comVal;
98. }
99. public long getComKey() {
100. return comKey;
101. }
102. public void setComKey(long comKey) {
103. this.comKey = comKey;
104. }
105. public long getComVal() {
106. return comVal;
107. }
108. public void setComVal(long comVal) {
109. this.comVal = comVal;
110. }
111. public void write(DataOutput out) throws IOException {
112. out.writeLong(comKey);
113. out.writeLong(comVal);
114. }
115. public void readFields(DataInput in) throws IOException {
116. this.comKey = in.readLong();
117. this.comVal = in.readLong();
118. }
119. /**
120. * 这个方法一定要实现
121. * java里面排序默认是小的放在前面，即返回负数的放在前面，这样就是所谓的升序排列
122. * 我们在下面的方法中直接返回一个差值，也就相当于会升序排列。
123. * 如果我们要实现降序排列，那么我们就可以返回一个正数
124. */
125. /*public int compareTo(CombineKey o) {
126. //第一列不相同时按升序排列，当第一列相同时第二列按升序排列
127. long minus = this.comKey - o.comKey;
128. //如果第一个值不相等时，我们就先对第一列进行排序
129. if (minus != 0){
130. return (int) minus;
131. }
132. //如果第一列相等时，我们就对第二列进行排序
133. return (int) (this.comVal - o.comVal);
134. }*/
135. /**
136. * 为了实现第一列不同时按降序排序，第一列相同时第二列按升序排列
137. * 第一列：降序，当第一列相同时，第二列：升序
138. * 为了实现降序，
139. */
140. public int compareTo(CombineKey o) {
141. //如果a-b<0即,a小于b，按这样 的思路应该是升序排列，我们可以返回一个相反数使其降序
142. long tmp = this.comKey - o.comKey;
143. //如果第一个值不相等时，我们就先对第一列进行排序
144. if (tmp != 0){
145. return (int) (-tmp);
146. }
147. //如果第一列相等时，我们就对第二列进行升序排列
148. return (int) (this.comVal - o.comVal);
149. }
150. @Override
151. public int hashCode() {
152. final int prime = 31;
153. int result = 1;
154. result = prime * result + (int) (comKey ^ (comKey >>> 32));
155. return result;
156. }
157. @Override
158. public boolean equals(Object obj) {
159. if (this == obj)
160. return true;
161. if (obj == null)
162. return false;
163. if (getClass() != obj.getClass())
164. return false;
165. CombineKey other = (CombineKey) obj;
166. if (comKey != other.comKey)
167. return false;
168. return true;
169. }
170. }