Hadoop二次排序及MapReduce处理流程实例详解

一、概述

MapReduce框架对处理结果的输出会根据key值进行默认的排序，这个默认排序可以满足一部分需求，但是也是十分有限的，在我们实际的需求当中，往往有要对reduce输出结果进行二次排序的需求。对于二次排序的实现，网络上已经有很多人分享过了，但是对二次排序的实现原理及整个MapReduce框架的处理流程的分析还是有非常大的出入，而且部分分析是没有经过验证的。本文将通过一个实际的MapReduce二次排序的例子，讲述二次排序的实现和其MapReduce的整个处理流程，并且通过结果和Map、Reduce端的日志来验证描述的处理流程的正确性。

二、需求描述

1.输入数据

[java] view plain copy

1. sort1   1
2. sort2   3
3. sort2   88
4. sort2   54
5. sort1   2
6. sort6   22
7. sort6   888
8. sort6   58

2.目标输出

[java] view plain copy

1. sort1   1,2
2. sort2   3,54,88
3. sort6   22,58,888

三、解决思路

1.首先，在思考解决问题思路时，我们应该先深刻的理解MapReduce处理数据的整个流程，这是最基础的，不然的话是不可能找到解决问题的思路的。我描述一下MapReduce处理数据的大概流程：首先，MapReduce框架通过getSplits()方法实现对原始文件的切片之后，每一个切片对应着一个MapTask，InputSplit输入到map()函数进行处理，中间结果经过环形缓冲区的排序，然后分区、自定义二次排序(如果有的话)和合并，再通过Shuffle操作将数据传输到reduce Task端，reduce端也存在着缓冲区，数据也会在缓冲区和磁盘中进行合并排序等操作，然后对数据按照key值进行分组，然后每处理完一个分组之后就会去调用一次reduce()函数，最终输出结果。大概流程 我画了一下，如下图：

2.具体解决思路

(1)：Map端处理

根据上面的需求，我们有一个非常明确的目标就是要对第一列相同的记录，并且对合并后的数字进行排序。我们都知道MapReduce框架不管是默认排序或者是自定义排序都只是对key值进行排序，现在的情况是这些数据不是key值，怎么办？其实我们可以将原始数据的key值和其对应的数据组合成一个新的key值，然后新的key值对应的value还是原始数据中的valu。那么我们就可以将原始数据的map输出变成类似下面的数据结构：

[java] view plain copy

1. {[sort1,1],1}
2. {[sort2,3],3}
3. {[sort2,88],88}
4. {[sort2,54],54}
5. {[sort1,2],2}
6. {[sort6,22],22}
7. {[sort6,888],888}
8. {[sort6,58],58}

那么我们只需要对[]里面的心key值进行排序就OK了，然后我们需要自定义一个分区处理器，因为我的目标不是想将新key相同的记录传到一个reduce中，而是想将新key中第一个字段相同的记录放到同一个reduce中进行分组合并，所以我们需要根据新key值的第一个字段来自定义一个分区处理器。通过分区操作后，得到的数据流如下：

[java] view plain copy

1. Partition1：{[sort1,1],1}、{[sort1,2],2}
2. Partition2：{[sort2,3],3}、{[sort2,88],88}、{[sort2,54],54}
3. Partition3：{[sort6,22],22}、{[sort6,888],888}、{[sort6,58],58}

分区操作完成之后，我调用自己的自定义排序器对新的key值进行排序。

[java] view plain copy

1. {[sort1,1],1}
2. {[sort1,2],2}
3. {[sort2,3],3}
4. {[sort2,54],54}
5. {[sort2,88],88}
6. {[sort6,22],22}
7. {[sort6,58],58}
8. {[sort6,888],888}

(2).Reduce端处理

经过Shuffle处理之后，数据传输到Reducer端了。在Reducer端按照组合键的第一个字段进行分组，并且每处理完一次分组之后就会调用一次reduce函数来对这个分组进行处理和输出。最终各个分组的数据结果变成类似下面的数据结构：

[java] view plain copy

1. sort1   1,2
2. sort2   3,54,88
3. sort6   22,58,888

四、具体实现

1.自定义组合键

[java] view plain copy

1. public class CombinationKey implements WritableComparable<CombinationKey>{
2. private Text firstKey;
3. private IntWritable secondKey;
4. //无参构造函数
5. public CombinationKey() {
6. this.firstKey = new Text();
7. this.secondKey = new IntWritable();
8. }
9. //有参构造函数
10. public CombinationKey(Text firstKey, IntWritable secondKey) {
11. this.firstKey = firstKey;
12. this.secondKey = secondKey;
13. }
14. public Text getFirstKey() {
15. return firstKey;
16. }
17. public void setFirstKey(Text firstKey) {
18. this.firstKey = firstKey;
19. }
20. public IntWritable getSecondKey() {
21. return secondKey;
22. }
23. public void setSecondKey(IntWritable secondKey) {
24. this.secondKey = secondKey;
25. }
26. public void write(DataOutput out) throws IOException {
27. this.firstKey.write(out);
28. this.secondKey.write(out);
29. }
30. public void readFields(DataInput in) throws IOException {
31. this.firstKey.readFields(in);
32. this.secondKey.readFields(in);
33. }
34. /*public int compareTo(CombinationKey combinationKey) {
35. int minus = this.getFirstKey().compareTo(combinationKey.getFirstKey());
36. if (minus != 0){
37. return minus;
38. }
39. return this.getSecondKey().get() - combinationKey.getSecondKey().get();
40. }*/
41. /**
42. * 自定义比较策略
43. * 注意：该比较策略用于MapReduce的第一次默认排序
44. * 也就是发生在Map端的sort阶段
45. * 发生地点为环形缓冲区(可以通过io.sort.mb进行大小调整)
46. */
47. public int compareTo(CombinationKey combinationKey) {
48. System.out.println("------------------------CombineKey flag-------------------");
49. return this.firstKey.compareTo(combinationKey.getFirstKey());
50. }
51. @Override
52. public int hashCode() {
53. final int prime = 31;
54. int result = 1;
55. result = prime * result + ((firstKey == null) ? 0 : firstKey.hashCode());
56. return result;
57. }
58. @Override
59. public boolean equals(Object obj) {
60. if (this == obj)
61. return true;
62. if (obj == null)
63. return false;
64. if (getClass() != obj.getClass())
65. return false;
66. CombinationKey other = (CombinationKey) obj;
67. if (firstKey == null) {
68. if (other.firstKey != null)
69. return false;
70. } else if (!firstKey.equals(other.firstKey))
71. return false;
72. return true;
73. }
74. }

说明：在自定义组合键的时候，我们需要特别注意，一定要实现WritableComparable接口，并且实现compareTo()方法的比较策略。这个用于MapReduce的第一次默认排序，也就是发生在Map阶段的sort小阶段，发生地点为环形缓冲区(可以通过io.sort.mb进行大小调整)，但是其对我们最终的二次排序结果是没有影响的，我们二次排序的最终结果是由我们的自定义比较器决定的。

2.自定义分区器

[java] view plain copy

1. /**
2. * 自定义分区
3. * @author 廖钟*民
4. * time : 2015年1月19日下午12:13:54
5. * @version
6. */
7. public class DefinedPartition extends Partitioner<CombinationKey, IntWritable>{
8. /**
9. * 数据输入来源：map输出 我们这里根据组合键的第一个值作为分区
10. * 如果不自定义分区的话，MapReduce会根据默认的Hash分区方法
11. * 将整个组合键相等的分到一个分区中，这样的话显然不是我们要的效果
12. * @param key map输出键值
13. * @param value map输出value值
14. * @param numPartitions 分区总数，即reduce task个数
15. */
16. public int getPartition(CombinationKey key, IntWritable value, int numPartitions) {
17. System.out.println("---------------------进入自定义分区---------------------");
18. System.out.println("---------------------结束自定义分区---------------------");
19. return (key.getFirstKey().hashCode() & Integer.MAX_VALUE) % numPartitions;
20. }
21. }

3.自定义比较器

[java] view plain copy

1. public class DefinedComparator extends WritableComparator{
2. protected DefinedComparator() {
3. super(CombinationKey.class,true);
4. }
5. /**
6. * 第一列按升序排列，第二列也按升序排列
7. */
8. public int compare(WritableComparable a, WritableComparable b) {
9. System.out.println("------------------进入二次排序-------------------");
10. CombinationKey c1 = (CombinationKey) a;
11. CombinationKey c2 = (CombinationKey) b;
12. int minus = c1.getFirstKey().compareTo(c2.getFirstKey());
13. if (minus != 0){
14. System.out.println("------------------结束二次排序-------------------");
15. return minus;
16. } else {
17. System.out.println("------------------结束二次排序-------------------");
18. return c1.getSecondKey().get() -c2.getSecondKey().get();
19. }
20. }
21. }

4.自定义分组

[java] view plain copy

1. /**
2. * 自定义分组有中方式，一种是继承WritableComparator
3. * 另外一种是实现RawComparator接口
4. * @author 廖*民
5. * time : 2015年1月19日下午3:30:11
6. * @version
7. */
8. public class DefinedGroupSort extends WritableComparator{
9. protected DefinedGroupSort() {
10. super(CombinationKey.class,true);
11. }
12. @Override
13. public int compare(WritableComparable a, WritableComparable b) {
14. System.out.println("---------------------进入自定义分组---------------------");
15. CombinationKey combinationKey1 = (CombinationKey) a;
16. CombinationKey combinationKey2 = (CombinationKey) b;
17. System.out.println("---------------------分组结果：" + combinationKey1.getFirstKey().compareTo(combinationKey2.getFirstKey()));
18. System.out.println("---------------------结束自定义分组---------------------");
19. //自定义按原始数据中第一个key分组
20. return combinationKey1.getFirstKey().compareTo(combinationKey2.getFirstKey());
21. }
22. }

5.主体程序实现

[java] view plain copy

1. public class SecondSortMapReduce {
2. // 定义输入路径
3. private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/sort_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. conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, "\t");
11. // 创建文件系统
12. FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);
13. // 如果输出目录存在，我们就删除
14. if (fileSystem.exists(new Path(OUT_PATH))) {
15. fileSystem.delete(new Path(OUT_PATH), true);
16. }
17. // 创建任务
18. Job job = new Job(conf, SecondSortMapReduce.class.getName());
19. //1.1   设置输入目录和设置输入数据格式化的类
20. FileInputFormat.setInputPaths(job, INPUT_PATH);
21. job.setInputFormatClass(KeyValueTextInputFormat.class);
22. //1.2   设置自定义Mapper类和设置map函数输出数据的key和value的类型
23. job.setMapperClass(SecondSortMapper.class);
24. job.setMapOutputKeyClass(CombinationKey.class);
25. job.setMapOutputValueClass(IntWritable.class);
26. //1.3   设置分区和reduce数量(reduce的数量，和分区的数量对应，因为分区为一个，所以reduce的数量也是一个)
27. job.setPartitionerClass(DefinedPartition.class);
28. job.setNumReduceTasks(1);
29. //设置自定义分组策略
30. job.setGroupingComparatorClass(DefinedGroupSort.class);
31. //设置自定义比较策略(因为我的CombineKey重写了compareTo方法，所以这个可以省略)
32. job.setSortComparatorClass(DefinedComparator.class);
33. //1.4   排序
34. //1.5   归约
35. //2.1   Shuffle把数据从Map端拷贝到Reduce端。
36. //2.2   指定Reducer类和输出key和value的类型
37. job.setReducerClass(SecondSortReducer.class);
38. job.setOutputKeyClass(Text.class);
39. job.setOutputValueClass(Text.class);
40. //2.3   指定输出的路径和设置输出的格式化类
41. FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));
42. job.setOutputFormatClass(TextOutputFormat.class);
43. // 提交作业 退出
44. System.exit(job.waitForCompletion(true) ? 0 : 1);
45. } catch (Exception e) {
46. e.printStackTrace();
47. }
48. }
49. public static class SecondSortMapper extends Mapper<Text, Text, CombinationKey, IntWritable>{
50. /**
51. * 这里要特殊说明一下，为什么要将这些变量写在map函数外边
52. * 对于分布式的程序，我们一定要注意到内存的使用情况，对于MapReduce框架
53. * 每一行的原始记录的处理都要调用一次map()函数，假设，这个map()函数要处理1一亿
54. * 条输入记录，如果将这些变量都定义在map函数里面则会导致这4个变量的对象句柄
55. * 非常的多(极端情况下将产生4*1亿个句柄，当然java也是有自动的GC机制的，一定不会达到这么多)
56. * 导致栈内存被浪费掉，我们将其写在map函数外面，顶多就只有4个对象句柄
57. */
58. private CombinationKey combinationKey = new CombinationKey();
59. Text sortName = new Text();
60. IntWritable score = new IntWritable();
61. String[] splits = null;
62. protected void map(Text key, Text value, Mapper<Text, Text, CombinationKey, IntWritable>.Context context) throws IOException, InterruptedException {
63. System.out.println("---------------------进入map()函数---------------------");
64. //过滤非法记录(这里用计数器比较好)
65. if (key == null || value == null || key.toString().equals("")){
66. return;
67. }
68. //构造相关属性
69. sortName.set(key.toString());
70. score.set(Integer.parseInt(value.toString()));
71. //设置联合key
72. combinationKey.setFirstKey(sortName);
73. combinationKey.setSecondKey(score);
74. //通过context把map处理后的结果输出
75. context.write(combinationKey, score);
76. System.out.println("---------------------结束map()函数---------------------");
77. }
78. }
79. public static class SecondSortReducer extends Reducer<CombinationKey, IntWritable, Text, Text>{
80. StringBuffer sb = new StringBuffer();
81. Text score = new Text();
82. /**
83. * 这里要注意一下reduce的调用时机和次数：
84. * reduce每次处理一个分组的时候会调用一次reduce函数。
85. * 所谓的分组就是将相同的key对应的value放在一个集合中
86. * 例如：<sort1,1> <sort1,2>
87. * 分组后的结果就是
88. * <sort1,{1,2}>这个分组会调用一次reduce函数
89. */
90. protected void reduce(CombinationKey key, Iterable<IntWritable> values, Reducer<CombinationKey, IntWritable, Text, Text>.Context context)
91. throws IOException, InterruptedException {
92. //先清除上一个组的数据
93. sb.delete(0, sb.length());
94. for (IntWritable val : values){
95. sb.append(val.get() + ",");
96. }
97. //取出最后一个逗号
98. if (sb.length() > 0){
99. sb.deleteCharAt(sb.length() - 1);
100. }
101. //设置写出去的value
102. score.set(sb.toString());
103. //将联合Key的第一个元素作为新的key，将score作为value写出去
104. context.write(key.getFirstKey(), score);
105. System.out.println("---------------------进入reduce()函数---------------------");
106. System.out.println("---------------------{[" + key.getFirstKey()+"," + key.getSecondKey() + "],[" +score +"]}");
107. System.out.println("---------------------结束reduce()函数---------------------");
108. }
109. }
110. }

程序运行的结果：

五、处理流程

看到前面的代码，都知道我在各个组件上已经设置好了相应的标志，用于追踪整个MapReduce处理二次排序的处理流程。现在让我们分别看看Map端和Reduce端的日志情况。

(1)Map端日志分析

[java] view plain copy

1. 15/01/19 15:32:29 INFO input.FileInputFormat: Total input paths to process : 1
2. 15/01/19 15:32:29 WARN snappy.LoadSnappy: Snappy native library not loaded
3. 15/01/19 15:32:30 INFO mapred.JobClient: Running job: job_local_0001
4. 15/01/19 15:32:30 INFO mapred.Task:  Using ResourceCalculatorPlugin : null
5. 15/01/19 15:32:30 INFO mapred.MapTask: io.sort.mb = 100
6. 15/01/19 15:32:30 INFO mapred.MapTask: data buffer = 79691776/99614720
7. 15/01/19 15:32:30 INFO mapred.MapTask: record buffer = 262144/327680
8. ---------------------进入map()函数---------------------
9. ---------------------进入自定义分区---------------------
10. ---------------------结束自定义分区---------------------
11. ---------------------结束map()函数---------------------
12. ---------------------进入map()函数---------------------
13. ---------------------进入自定义分区---------------------
14. ---------------------结束自定义分区---------------------
15. ---------------------结束map()函数---------------------
16. ---------------------进入map()函数---------------------
17. ---------------------进入自定义分区---------------------
18. ---------------------结束自定义分区---------------------
19. ---------------------结束map()函数---------------------
20. ---------------------进入map()函数---------------------
21. ---------------------进入自定义分区---------------------
22. ---------------------结束自定义分区---------------------
23. ---------------------结束map()函数---------------------
24. ---------------------进入map()函数---------------------
25. ---------------------进入自定义分区---------------------
26. ---------------------结束自定义分区---------------------
27. ---------------------结束map()函数---------------------
28. ---------------------进入map()函数---------------------
29. ---------------------进入自定义分区---------------------
30. ---------------------结束自定义分区---------------------
31. ---------------------结束map()函数---------------------
32. ---------------------进入map()函数---------------------
33. ---------------------进入自定义分区---------------------
34. ---------------------结束自定义分区---------------------
35. ---------------------结束map()函数---------------------
36. ---------------------进入map()函数---------------------
37. ---------------------进入自定义分区---------------------
38. ---------------------结束自定义分区---------------------
39. ---------------------结束map()函数---------------------
40. 15/01/19 15:32:30 INFO mapred.MapTask: Starting flush of map output
41. ------------------进入二次排序-------------------
42. ------------------结束二次排序-------------------
43. ------------------进入二次排序-------------------
44. ------------------结束二次排序-------------------
45. ------------------进入二次排序-------------------
46. ------------------结束二次排序-------------------
47. ------------------进入二次排序-------------------
48. ------------------结束二次排序-------------------
49. ------------------进入二次排序-------------------
50. ------------------结束二次排序-------------------
51. ------------------进入二次排序-------------------
52. ------------------结束二次排序-------------------
53. ------------------进入二次排序-------------------
54. ------------------结束二次排序-------------------
55. ------------------进入二次排序-------------------
56. ------------------结束二次排序-------------------
57. ------------------进入二次排序-------------------
58. ------------------结束二次排序-------------------
59. ------------------进入二次排序-------------------
60. ------------------结束二次排序-------------------
61. ------------------进入二次排序-------------------
62. ------------------结束二次排序-------------------
63. ------------------进入二次排序-------------------
64. ------------------结束二次排序-------------------
65. 15/01/19 15:32:30 INFO mapred.MapTask: Finished spill 0
66. 15/01/19 15:32:30 INFO mapred.Task: Task:attempt_local_0001_m_000000_0 is done. And is in the process of commiting
67. 15/01/19 15:32:30 INFO mapred.LocalJobRunner:
68. 15/01/19 15:32:30 INFO mapred.Task: Task ‘attempt_local_0001_m_000000_0‘ done.
69. 15/01/19 15:32:30 INFO mapred.Task:  Using ResourceCalculatorPlugin : null
70. 15/01/19 15:32:30 INFO mapred.LocalJobRunner:

从Map端的日志，我们可以很容易的看出来每一条记录开始时进入到map()函数进行处理，处理完了之后立马就自定义分区函数中对其进行分区，当所有输入数据经过map()函数和分区函数处理之后，就调用自定义二次排序函数对其进行排序。

(2)Reduce端日志分析

[java] view plain copy

1. 15/01/19 15:32:30 INFO mapred.Merger: Merging 1 sorted segments
2. 15/01/19 15:32:30 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 130 bytes
3. 15/01/19 15:32:30 INFO mapred.LocalJobRunner:
4. ---------------------进入自定义分组---------------------
5. ---------------------分组结果：0
6. ---------------------结束自定义分组---------------------
7. ---------------------进入自定义分组---------------------
8. ---------------------分组结果：-1
9. ---------------------结束自定义分组---------------------
10. ---------------------进入reduce()函数---------------------
11. ---------------------{[sort1,2],[1,2]}
12. ---------------------结束reduce()函数---------------------
13. ---------------------进入自定义分组---------------------
14. ---------------------分组结果：0
15. ---------------------结束自定义分组---------------------
16. ---------------------进入自定义分组---------------------
17. ---------------------分组结果：0
18. ---------------------结束自定义分组---------------------
19. ---------------------进入自定义分组---------------------
20. ---------------------分组结果：-4
21. ---------------------结束自定义分组---------------------
22. ---------------------进入reduce()函数---------------------
23. ---------------------{[sort2,88],[3,54,88]}
24. ---------------------结束reduce()函数---------------------
25. ---------------------进入自定义分组---------------------
26. ---------------------分组结果：0
27. ---------------------结束自定义分组---------------------
28. ---------------------进入自定义分组---------------------
29. ---------------------分组结果：0
30. ---------------------结束自定义分组---------------------
31. ---------------------进入reduce()函数---------------------
32. ---------------------{[sort6,888],[22,58,888]}
33. ---------------------结束reduce()函数---------------------
34. 15/01/19 15:32:30 INFO mapred.Task: Task:attempt_local_0001_r_000000_0 is done. And is in the process of commiting
35. 15/01/19 15:32:30 INFO mapred.LocalJobRunner:
36. 15/01/19 15:32:30 INFO mapred.Task: Task attempt_local_0001_r_000000_0 is allowed to commit now
37. 15/01/19 15:32:30 INFO output.FileOutputCommitter: Saved output of task ‘attempt_local_0001_r_000000_0‘ to hdfs://liaozhongmin:9000/out
38. 15/01/19 15:32:30 INFO mapred.LocalJobRunner: reduce > reduce
39. 15/01/19 15:32:30 INFO mapred.Task: Task ‘attempt_local_0001_r_000000_0‘ done.
40. 15/01/19 15:32:31 INFO mapred.JobClient:  map 100% reduce 100%
41. 15/01/19 15:32:31 INFO mapred.JobClient: Job complete: job_local_0001
42. 15/01/19 15:32:31 INFO mapred.JobClient: Counters: 19
43. 15/01/19 15:32:31 INFO mapred.JobClient:   File Output Format Counters
44. 15/01/19 15:32:31 INFO mapred.JobClient:     Bytes Written=40
45. 15/01/19 15:32:31 INFO mapred.JobClient:   FileSystemCounters
46. 15/01/19 15:32:31 INFO mapred.JobClient:     FILE_BYTES_READ=446
47. 15/01/19 15:32:31 INFO mapred.JobClient:     HDFS_BYTES_READ=140
48. 15/01/19 15:32:31 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=131394
49. 15/01/19 15:32:31 INFO mapred.JobClient:     HDFS_BYTES_WRITTEN=40
50. 15/01/19 15:32:31 INFO mapred.JobClient:   File Input Format Counters
51. 15/01/19 15:32:31 INFO mapred.JobClient:     Bytes Read=70
52. 15/01/19 15:32:31 INFO mapred.JobClient:   Map-Reduce Framework
53. 15/01/19 15:32:31 INFO mapred.JobClient:     Reduce input groups=3
54. 15/01/19 15:32:31 INFO mapred.JobClient:     Map output materialized bytes=134
55. 15/01/19 15:32:31 INFO mapred.JobClient:     Combine output records=0
56. 15/01/19 15:32:31 INFO mapred.JobClient:     Map input records=8
57. 15/01/19 15:32:31 INFO mapred.JobClient:     Reduce shuffle bytes=0
58. 15/01/19 15:32:31 INFO mapred.JobClient:     Reduce output records=3
59. 15/01/19 15:32:31 INFO mapred.JobClient:     Spilled Records=16
60. 15/01/19 15:32:31 INFO mapred.JobClient:     Map output bytes=112
61. 15/01/19 15:32:31 INFO mapred.JobClient:     Total committed heap usage (bytes)=391118848
62. 15/01/19 15:32:31 INFO mapred.JobClient:     Combine input records=0
63. 15/01/19 15:32:31 INFO mapred.JobClient:     Map output records=8
64. 15/01/19 15:32:31 INFO mapred.JobClient:     SPLIT_RAW_BYTES=99
65. 15/01/19 15:32:31 INFO mapred.JobClient:     Reduce input records=8

首先，我们看了Reduce端的日志，第一个信息我应该很容易能够很容易看出来，就是分组和reduce()函数处理都是在Shuffle完成之后才进行的。另外一点我们也非常容易看出，就是每次处理完一个分组数据就会去调用一次的reduce()函数对这个分组进行处理和输出。此外，说明一些分组函数的返回值问题，当返回0时才会被分到同一个组中。另外一点我们也可以看出来，一个分组中每合并n个值就会有n-1分组函数返回0值，也就是说进行了n-1次比较。

六、总结

本文主要从MapReduce框架执行的流程，去分析了如何去实现二次排序，通过代码进行了实现，并且对整个流程进行了验证。另外，要吐槽一下，网络上有很多文章都记录了MapReudce处理二次排序问题，但是对MapReduce框架整个处理流程的描述错漏很多，而且他们最终的流程描述也没有证据可以支撑。所以，对于网络上的学习资源不能够完全依赖，要融入自己的思想，并且要重要的观点进行代码或者实践的验证。另外，今天在一个hadoop交流群上听到少部分人在讨论，有了hive我们就不用学习些MapReduce程序？对这这个问题我是这么认为：我不相信写不好MapReduce程序的程序员会写好hive语句，最起码的他们对整个执行流程是一无所知的，更不用说性能问题了，有可能连最常见的数据倾斜问题的弄不清楚。