MR Shuffle流程入门

什么是shuffle?

从map方法的输出开始,到作为输入数据传给reduce方法的过程叫做shuffle.

shuffle流程是怎样的?

map端

map方法开始产生输出数据时，并不是简单地将它写到磁盘.

每个map任务都会有一个环形内存缓冲区用于存储map的输出数据。在默认情况下，缓冲区的大小为100MB,这个值可以通过mapreduce.task.io.sort.mb属性来调整。一旦缓冲区的内容达到阈值（默认是0.8，或者是80%，属属性是mapreduce.map.sort.spill.percent），一个后台线程便开始把内容溢写(spill)到磁盘里，这个位置由属性mapreduce.cluster.local.dir来指定的。在将数据溢写到磁盘过程中，map的输出数据继续写到缓冲区，但如果在此期间缓冲区被填满，map会被阻塞直到写磁盘过程完成。

在写磁盘之前，线程会根据分区器的逻辑把数据划分为不同的分区(partition)。然后，在每个分区中，后台线程会按键进行内存中排序（QuickSort，默认是字典顺序）。如果指定了一个combiner函数，它就在排序后的输出上运行。运行combiner函数使得map输出结果更紧凑，因此减少写到磁盘的数据和传递给reducer的数据。

每次内存缓冲区达到溢出阈值，就会新建一个溢出文件(spill file),因此在map任务写完其最后一个输出记录之后，可能会有几个溢出文件。在MapTask任务完成之前，多个溢出文件被合并成一个已分区且已排序的输出文件。配置属性mapreduce.task.io.sort.factor控制着一次最多能合并多少个文件，默认值是10。

如果至少存在3个溢出文件(通过mapreduce.map.combine.minspills属性设置)时，则combiner就会在输出文件写到磁盘之前再次运行。combiner可以在输入上反复运行，但并不影响最终结果。如果只有1或2个溢出文件，那么由于map输出规模减少，因而不值得调用combiner产生开销，因此不会为该map输出再次运行combiner。

为了使写磁盘的速度更快，节约磁盘空间，并且减少传给reducer的数据量，在溢写到磁盘的过程中对数据进行压缩往往是个很好的主意。在默认情况下，输出是不压缩的，但只要将mapreduce.map.output.compress设置为true,就可以轻松启用此功能。使用的压缩库由mapreduce.map.output.compress.codec指定。

reduce端

reducer通过HTTP得到输出文件的分区。用于文件分区的工作线程的数量由任务的mapreduce. shuffle.max. threads属性控制，此设置针对的是每一个节点管理器，而不是针对每个map任务。

现在转到处理过程的reduce部分。map输出文件位于运行MapTask的本地磁盘（注意，尽管map输出经常写到MapTask本地磁盘，但reduce输出并不这样）。现在，tasktracker需要为分区文件运行reduce任务。并且，
reduce任务需要集群上若干个map任务的map输出作为其特殊的分区文件。每个map任务的完成时间可能不同，因此在每个任务完成时，reduce任务就开始复制其输出。这就是reduce任务的复制阶段。reduce任务有少量复制线程，因此能够并行取得map输出。默认值是5个线程，但这个默认值可以修改设置mapreduce.reduce.shuffle. parallelcopies 属性即可。

reducer如何知道要从哪台机器取得map输出呢？ map任务成功完成后，它们会使用心跳机制通知它们的application master。因此，对于指定作业，applicationmaster知道map输出和主机位置之间的映射关系。reducer中的一个线程定期询问master以便获取map输出主机的位置，直到获得所有输出位置。

由于第一个reducer可能失败，因此主机并没有在第一个reducer检索到map输出时就立即从磁盘上删除它们。相反，主机会等待，直到application master告知它删除map输出，这是作业完成后执行的。

如果map输出相当小，会被复制到reduce任务JVM的内存（缓冲区大小由mapreduce.reduce.shuffle.input. buffer.percent 属性控制，指定用于此用途的堆空间的百分比）（还是为了减少磁盘IO），否则，map输出被复制到磁盘。一旦内存缓冲区达到阈值大小（由 mapreduce.reduce.shuffle.merge.percent 决定）或达到 map 输出阈值（由 mapreduce. reduce. merge. inmem .threshold 控制），则合并后溢出写到磁盘中。如果指定combiner,则在合并期间运行它以降低写入硬盘的数据量。

随着磁盘上的溢写文件数量增多，后台线程会将它们合并为更大的、排好序的文件。这会为后面的合并节省一些时间。注意，为了合并，压缩的map输出（通过map任务）都必须在内存中被解压缩。

复制完所有map输出后，reduce任务进入排序阶段（更恰当的说法是合并阶段，因为排序是在map端进行的），这个阶段将合并map输岀，维持其顺序排序。这是循环进行的。比如，如果有50个map输出，而合并因子是10（10为默认设置，由mapreduce.task. io.sort.factor ,与的合并类似），合并将进行 5 趟，每趟将10个文件合并成一个文件，因此最后有5个中间文件。

在最后阶段，即reduce阶段，直接把数据输入reduce函数，从而省略了一次磁盘往返行程，并没有将这5个文件合并成一个已排序的文件作为最后一趟。最后的合并可以来自内存和磁盘片段。

总结

shuffle中map端进行了几次排序?为什么.reduce端进行了几次排序?
shuffle中如何做可以减少磁盘IO