奇迹发生的地方——MapReduce Shuffle
文/孙鹏晖
说起大数据，就不得不提到Hadoop；提到Hadoop，就不得不言及MapReduce；言及MapReduce，众所周知，MapReduce作为并行分布式编程计算模型，利用map算子和reduce算子将计算过程划分为map阶段和reduce阶段，由此进行海量数据的并行分布式计算。
map和reduce相信大家都已经很熟悉了，然而，map阶段处理的数据究竟是怎样传递给reduce阶段的呢？这里还隐藏着一些不为人知的秘密，它是MapReduce计算过程的核心，被称为“奇迹发生的地方”，它就是——Shuffle。
上图是官方给出的Shuffle过程示意图，由此可知，Shuffle过程横跨了map端和reduce端，对于连接map和reduce阶段起到了至关重要的作用。
Shuffle，从字面意义上理解就是洗牌的意思，在MapReduce中概括来讲，就是通过map端的缓冲、分区、排序、溢写、合并和reduce端的复制、排序&合并步骤，完成从map阶段的数据输出到reduce阶段的数据输入处理。
map函数计算完成后需要输出数据，每个map任务都会有一个缓冲区，用于存储map函数输出的内容，缓冲区的大小默认是100M，这个值可以通过mapreduce.task.io.sort.mb配置项进行配置。
随着输出数据的增多，缓冲区中内容到达一定的阈值（默认是0.8，可以通过mapreduce.map.sort.spill.percent配置项进行配置）时，缓冲区会产生溢出。
当数据缓冲区开始产生溢出时，此时会对缓冲区中的内容进行分区，分区的目的是将map的输出划分成不同的区，从而可以将不同的分区分配给不同的reduce进行处理，后面reduce就会根据划分的分区读取自己要负责处理的数据。
默认的分区方法是，将map输出结果的key值计算hashcode，并对代码中配置的reduce task数量进行模运算，将结果相同的内容划分到同一个区，使得一个reduce任务对应一个分区的数据。
在划分完分区后，在每个分区内还需要进行sort by key的排序操作，如果我们设置了Combiner，此时还会运行Combiner函数，使得输出结果更为紧凑，从而减少数据写入和传输量。
当缓冲区内容量达到设定的阈值时，经过分区和排序后的缓冲区数据就会溢出并写入到磁盘中并形成一个溢写文件。
在进行分区和排序操作的同时，map任务仍然会向缓冲区中并行写入数据，如果缓冲区空间已满，map任务此时会阻塞并等待缓冲区中的内容写入磁盘。
由于每次缓冲区溢写都会产生一个新的溢写文件，因此这个过程可能会产生大量的溢写文件，合并步骤通过不断地排序和Combine操作将这些溢写文件合并为完整的输出文件。
reduce任务需要若干个map任务的输出作为其输入内容，前面提到，map任务在输出过程中会进行分区，因此reduce任务在复制阶段只会复制自己对应分区中的数据。
如果reduce接收到的数据比较小则会直接保存在内存缓冲区中，如果数据量达到缓冲区阈值时同样会将数据溢写到磁盘上。reduce端能够能够并行复制map的输出，默认是5个线程，可以通过mapreduce.reduce.shuffle.parallelcopies配置项进行配置。
复制完map任务输出的分区数据后，reduce端进行排序&合并步骤，维持map顺序排序并合并map的输出，可以通过mapreduce.task.io.sort.factor配置项设置合并流因子的大小，默认值是10。
排序和合并步骤完成后，就会把准备好的数据输入到reduce函数进行我们熟悉的reduce计算了。
Shuffle阶段搭建起了map与reduce的桥梁，也是MapReduce设计的精华所在，深入MapReduce的运行机理，会发现Shuffle机制的设计实际上更加精巧与复杂，更进一步学习，去探索另一个奇迹诞生的地方吧。
顾问：许国平 李湘宜
罗学平 赵晓玲 张刚
总编：孙鹏晖
美编：孙鹏晖
-本文为“数风云”第18期文章；
-欢迎来稿：请按“题目-作者”格式命名发送到[email protected]。