MapReduce流程详解

简单来讲MapReduce的流程是这样的：



稍详细点是这样的：



首先对输入文件执行分片操作，默认每个split的大小是64M，每一个split对应一个Map任务；对每个split执行map，输出结果会临时存储在硬盘上；等所有的Map任务都执行完毕（或执行完毕的Map任务达到一个比例，可以修改），每个Reducer会从各个Mappers上拉取属于自己的数据；然后对这些数据执行reduce，最后将结果输出到HDFS上。

这里已经体现出了针对中间结果的shuffle过程，但忽略了很多细节，比如是如何对每个split执行map的，如何对数据执行reduce的，如何决定数据映射到那个reducer上的，中间结果的排序、合并、combine操作。

接下来是MapReduce流程详细的描述，主要参考于《MapReduce:详解Shuffle过程》（dennyy99@gmail.com）。

官方对MapReduce过程的描述是这样的：



map的输入是<key,value>对，默认情况下，key是每行行首字符在文件中的偏移量，value是当前行的内容。map的输出也是<key,value>对，得到map的输出结果后，需要在此时决定数据要交给哪个reducer处理。默认对key取hash后再以reduce task数量取模，这只是为了平均reducer的处理能力，MapReduce提供了Partitioner接口，用户可以自己重新定义。



每个map task都有一个内存缓冲区，默认是100MB，它存储着map的输出结果，结果是三元组<partition, key, value>形式的。

当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的形式存放到磁盘，然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为spill，中文可译为溢写，这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8，也就是当缓冲区的数据已经达到阈值（buffer size * spill percent = 100MB * 0.8 = 80MB）时，溢写线程启动，锁定这80MB的内存，执行溢写过程。Map task的输出结果还可以往剩下的20MB内存中写，互不影响。

当溢写线程启动后，需要对这80MB空间内的数据做排序(sort)，每个partition里，相同key的数据排序后做合并，形成<key,value list>，如果设置了combiner，还会执行combine，combine函数一般与reduce函数相同。排序是MapReduce模型默认的行为，这里的排序也是对序列化的字节做的排序。如图所示，假定有3个reducer，则每个临时文件都分为三块。

每次溢写会在磁盘上生成一个溢写文件，如果map的输出结果很大，就有多次这样的溢写发生，磁盘上相应的就会有多个溢写文件存在。当map task真正完成时，内存缓冲区中的数据也全部溢写到磁盘中形成一个溢写文件。最终磁盘中会至少有一个这样的溢写文件存在(如果map的输出结果很少，当map执行完成时，只会产生一个溢写文件)。然后将这些溢写文件归并到一起，这个过程就叫做merge。这个过程实际是对每个partition分别执行归并排序的过程，排序时也会执行合并操作，形成更大的<key,value list>，如果设置过combiner，也会进行combine。最终生成的文件是分块的，每块对应一个reducer。

所有的这些工作，都是为了减少网络传输数据量，和减少reducer的工作量。

至此，map端的所有工作都已结束，然后等待reduce task来拉取最终生成的这个文件。

当所有的map任务完成后，或者一定比例的map任务完成后（可设置），reduce端开始从map端拉取数据。



简单地说，在执行reduce task之前的工作就是不断地拉取当前job里每个map task的最终结果，然后对从不同地方拉取过来的数据不断地做merge，最终形成一个文件作为reduce task的输入文件。

reduce端的shuffle细节可分为三个阶段：

1. copy过程，简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher)，通过HTTP方式请求map端获取map task的输出文件。因为map task早已结束，这些文件就存储在map端的本地磁盘中。

2. merge阶段。这里的merge如map端的merge类似，只是数组中存放的是从不同map端copy来的数据。Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活，它基于JVM的heap size设置，因为Shuffle阶段Reducer不运行，所以应该把绝大部分的内存都给Shuffle用。这里需要强调的是，merge有三种形式：1)内存到内存 2)内存到磁盘 3)磁盘到磁盘。默认情况下第一种形式不启用。当内存中的数据量到达一定阈值，就启动内存到磁盘的merge。与map端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行，直到没有map端的数据时才结束，然后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。

3. reduce的输入文件。不断地merge后，最后会生成一个“最终文件”。为什么加引号？因为这个文件可能存在于磁盘上，也可能存在于内存中。对我们来说，当然希望它存放于内存中，直接作为Reducer的输入，但默认情况下，这个文件是存放于磁盘中的。

当Reducer的输入文件已定，整个Shuffle才最终结束。然后就是Reducer执行，把结果放到HDFS上。