Hadoop shuffle机制中针对中间数据的排序过程详解（源代码级）

aaahd

　　在所有公开资料中，很少有对Hadoop 中间数据的sort过程进行详细介绍的。如果想要深入了解hadoop对中间数据的排序机制，只有通过阅读源代码才能达到。而hadoop的这段代码本身具有非常大的迷惑性，如果不注意细节，很容易会发生错误的理解。 本篇文章从原理上详细介绍了hadoop针对中间数据的排序机制，并且对一些重要的源代码段进行了介绍。阅读本文对理解该机制或者深入阅读该部分的hadoop源代码都有较大帮助。
　　本篇文章是建立在对于hadoop0.20.2版本的源代码研究之上。其他更高级版本如果有所变动，希望读者能够给予反馈。
　　如果对hadoop的shuffle机制有所了解的人都知道，map所产生的中间数据在送给reduce进行处理之前是要经过排序的。具体的过程实际上是快速排序，堆排序和归并排序的完美结合。
　　首先，当map函数处理完输入数据之后，会将中间数据存在本机的一个或者几个文件当中，并且针对这些文件内部的记录进行一次快速排序，这里的排序是升序排序。这段代码是在MapTask的内部类MapOutputBuffer中实现的。
　　当map阶段完成后，系统会启动reduce过程。reduce过程会把这些由map输出的中间文件拷贝到本地，然后生成一个或者几个Segment类的实例，以下我们称这些实例为segment。Segment类封装了这些中间数据，并且提供了一些针对这些中间数据的操作，比如读取记录等。在reduce端，这些中间数据可以存在内存中，也可以存在硬盘中。同时，系统还会启动两个merge（归并）线程，一个是针对内存中的segment进行归并，一个是针对硬盘中的segment进行归并。merge过程实际上就是调用了Merge类的merge方法。
　　Merge类的merge方法生成了一个MergeQueue类的实例，并且调用了该类的merge方法。MergeQueue类是PriorityQueue类的一个子类，同时实现了RawKeyValueIterator接口。PriorityQueue类实际上是一个小根堆，而MergeQueue的merge方法实际上就是将segment对象存储进父类的数据结构中，并且建立一个小根堆的过程。因此，hadoop的归并和排序不是两个分开的过程，而是一个过程。在将segment归并的同时进行了排序。
　　需要注意的是，这里针对segment排序的过程是以segment为单位的，而不是以segment中存储的记录（record）为单位的。而这里排序过程中对两个segment对象的比较是对segment中存储的第一个记录的键的比较。也就是说假设有两个segment，一个叫a，一个叫b，a<b仅仅是因为a的第一个记录的键小于b的第一个记录的键。具体的比较方法由用户定义的comparator类定义的。具体的比较过程在MergeQueue类中的lessThan方法中定义。
　　现在，我们已经得到了一个以segment为单位，以segment中第一个记录的键为比较依据的小根堆，至此在系统中所谓的sort阶段就已经结束了
　　接下来，系统会不停的从这个小根堆里取出位于根节点的segment的第一个记录交给reduce函数处理。注意，因为该小根堆是以每一个segment的第一个记录的键为排序依据的，所以根节点的第一个记录的键一定是所有segment中第一个记录的键的最小值。由于segment存储的是map输出的数据，而这些数据在传送给reduce之前已经经过排序（升序），所以，每个segment的第一个记录的键一定是该segment中所有键的最小值。从而根segment的第一个记录的键一定是所有记录的键的最小值。这里实际就是利用了归并排序。在从根segment中取出第一个记录之后，系统还会对该小根堆进行调整，以保证小根堆的性质。
　　以上是shuffle过程中排序的完整过程。虽然在hadoop的shuffle过程中有一个明确的sort阶段，但是实际上可以看出中间数据的排序是贯穿于整个shuffle阶段的。


　　在所有公开资料中，很少有对Hadoop 中间数据的sort过程进行详细介绍的。如果想要深入了解hadoop对中间数据的排序机制，只有通过阅读源代码才能达到。而hadoop的这段代码本身具有非常大的迷惑性，如果不注意细节，很容易会发生错误的理解。 本篇文章从原理上详细介绍了hadoop针对中间数据的排序机制，并且对一些重要的源代码段进行了介绍。阅读本文对理解该机制或者深入阅读该部分的hadoop源代码都有较大帮助。
　　本篇文章是建立在对于hadoop0.20.2版本的源代码研究之上。其他更高级版本如果有所变动，希望读者能够给予反馈。
　　如果对hadoop的shuffle机制有所了解的人都知道，map所产生的中间数据在送给reduce进行处理之前是要经过排序的。具体的过程实际上是快速排序，堆排序和归并排序的完美结合。
　　首先，当map函数处理完输入数据之后，会将中间数据存在本机的一个或者几个文件当中，并且针对这些文件内部的记录进行一次快速排序，这里的排序是升序排序。这段代码是在MapTask的内部类MapOutputBuffer中实现的。
　　当map阶段完成后，系统会启动reduce过程。reduce过程会把这些由map输出的中间文件拷贝到本地，然后生成一个或者几个Segment类的实例，以下我们称这些实例为segment。Segment类封装了这些中间数据，并且提供了一些针对这些中间数据的操作，比如读取记录等。在reduce端，这些中间数据可以存在内存中，也可以存在硬盘中。同时，系统还会启动两个merge（归并）线程，一个是针对内存中的segment进行归并，一个是针对硬盘中的segment进行归并。merge过程实际上就是调用了Merge类的merge方法。
　　Merge类的merge方法生成了一个MergeQueue类的实例，并且调用了该类的merge方法。MergeQueue类是PriorityQueue类的一个子类，同时实现了RawKeyValueIterator接口。PriorityQueue类实际上是一个小根堆，而MergeQueue的merge方法实际上就是将segment对象存储进父类的数据结构中，并且建立一个小根堆的过程。因此，hadoop的归并和排序不是两个分开的过程，而是一个过程。在将segment归并的同时进行了排序。
　　需要注意的是，这里针对segment排序的过程是以segment为单位的，而不是以segment中存储的记录（record）为单位的。而这里排序过程中对两个segment对象的比较是对segment中存储的第一个记录的键的比较。也就是说假设有两个segment，一个叫a，一个叫b，a<b仅仅是因为a的第一个记录的键小于b的第一个记录的键。具体的比较方法由用户定义的comparator类定义的。具体的比较过程在MergeQueue类中的lessThan方法中定义。
　　现在，我们已经得到了一个以segment为单位，以segment中第一个记录的键为比较依据的小根堆，至此在系统中所谓的sort阶段就已经结束了
　　接下来，系统会不停的从这个小根堆里取出位于根节点的segment的第一个记录交给reduce函数处理。注意，因为该小根堆是以每一个segment的第一个记录的键为排序依据的，所以根节点的第一个记录的键一定是所有segment中第一个记录的键的最小值。由于segment存储的是map输出的数据，而这些数据在传送给reduce之前已经经过排序（升序），所以，每个segment的第一个记录的键一定是该segment中所有键的最小值。从而根segment的第一个记录的键一定是所有记录的键的最小值。这里实际就是利用了归并排序。在从根segment中取出第一个记录之后，系统还会对该小根堆进行调整，以保证小根堆的性质。
　　以上是shuffle过程中排序的完整过程。虽然在hadoop的shuffle过程中有一个明确的sort阶段，但是实际上可以看出中间数据的排序是贯穿于整个shuffle阶段的。