MapReduce 的 shuffle 机制 原 荐

由于 MapReduce 确保每个 reducer 的输入都是按键 排序 的，因此在 map 处理完数据之后传给 reducer 的这个过程中需要进行一系列操作，这个操作过程就是 shuffle。在《hadoop权威指南》中指出，shuffle 是 MapReduce 的 “心脏”，了解 shuffle 工作机制有助于我们优化 MapReduce 程序，接下来我们就来看看它的运行机制。

shuffle 流程

先用一张图表示 shuffle 的整个过程。从图中我们可以看到 shuffle 流程主要是对 map 的数据进行排序、分组发送给 reduce 后再进行合并的一个过程，我们将分 map 和 reduce 两个部分来讲解 shuffle 的流程。

map 端 shuffle

map 任务开始产生数据时，会先将这些数据存储在一个 内存缓冲区 中，这个缓冲区大小默认为 100MB，可以通过设置 mapreduce.task.io.sort.mb 来改变其大小。由于 hadoop 处理的是海量数据，100MB 的内存显然是不够用的，因此达到一定 阈值 时（默认为 0.8，可以通过设置 mapreduce.map.sort.spill.percent 来改变其大小），会将内存中的内容溢出（spill）到磁盘当中，溢出的路径是由 mapreduce.cluster.local.dir 属性指定的。在溢出到磁盘的过程中，如果缓冲区中还有空间，map 程序会继续输出数据到缓冲区中，如果没有空间的话，map 输出程序则会阻塞直到数据写入到磁盘后。

在上图中 buffer in memory（输出到缓存中） 和 merge on disk（合并到磁盘） 这两个步骤中间还有一个 分区、排序 的步骤。分区能达到跟分组类似的效果，例如读取一个含有大量电话号码的数据时，把 138 的分为一组，把 135 分为一组。这个效果可以通过自定一个类继承 Partitioner，然后在 Job 中调用 setPartitionerClass 方法设置分区类来完成。在每个分区中，后台线程按照键的值对数据在内存中进行排序，如果有一个 combiner 方法，则在排序完成之后运行它。combiner 方法会使 map 输出更紧凑，减少写到磁盘中的数据和传给 reducer 的数据。

一般情况下，map 的输出结果并不会进行压缩，由于数据量大，对网络资源的耗费很大，为了对 mapreduce 程序进行优化，我们可以将 mapreduce.map.output.compress 属性设置为 true，这样当 map 将数据写到磁盘时就会对数据进行压缩。具体的压缩格式可以通过 mapreduce.map.output.compress.codec 属性来设置。当所有记录都写完之后，map 会合并全部的溢出文件为一个分区且排序的文件传给 reduce。

reduce 端 shuffle

reducer 通过 HTTP 的方式获取 map 的的输出数据，这是复制阶段。reducer 在复制阶段把 Map 输出复制到 Reducer 的内存或磁盘，一个 Map 任务完成后，Reduce 就开始复制输出。复制完所有的 map 输出之后，reducer 对这些数据进行合并，使它们仍然保持有序。合并完成之后，直接将这些数据输入到 reduce 函数中，从而省略一次写入磁盘的时间。至此，整个 shuffle 流程就完成了。

以上便是我对 MR shuffle 机制的理解，如果其中有错，欢迎指出。