Shuffle过程详解

Shuffle过程是MapReduce的核心，最近看了很多资料，网上说法大体相同，但有些地方有一点点出入，就是各个阶段的执行顺序

总个shuffle过程可以看做是从map输出到reduce输入的这个中间过程，在这个中间过程中，经过了一系列的步骤

下面看看官方给出的图

Map端

下面是我画的一张图 

1.Input Split分配给Map

2.Map过程进行处理，Mapper任务会接收输入分片，然后不断的调用map函数，对记录进行处理。处理完毕后，转换为新的<key,value>输出。（其中每个分片对应一个map,一个map可以被调用多次来处理该分片）

3.Map的输出结果缓存在内存里

4.内存中进行Partition，默认是HashPartitioner(采用取模hash (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks)， 目的是将map的结果分给不同的reducer，有几个Partition，就有几个reducer，partition的数目可以在job启动时通过参数 “-Dmapreduc.job.reduces”设置(Hadoop 2.2.0), HashPartitioner可以让map的结果均匀的分给不同机器上的reducer，保证负载均衡。

5.内存中在Partition结束后，对于不同分区的数据，会按照key进行排序，这里的key必须实现WritableComparable接口。该接口实现了Comparable接口，因此可以进行比较排序

6.对于排序后的<key,value>，会按照key进行分组。如果key相同，那么相同key的<key,value>就被分到一个组中。最终，每个分组会调用一次reduce函数

7.排序分组结束后，相同的key在一起组成了一个列表，如果设置过combiner，就合并数据，减少写入磁盘的记录数（combiner本质就是一个reducer）

8.当内存中buffer(default 512M)达到阈值(default 80%)，就会把记录spill(即溢写)到磁盘中，优化Map时可以调大buffer的阈值，缓存更多的数据。

9.当磁盘中的spill文件数目比规定的文件数目多时,会多次调用combiner。在不影响结果的前下,Combiner可以被调用多次。（比如磁盘中的spill文件数目规定在个（包括）以上,当达到规定数量时, map的新的output会再次运行combiner，而如果磁盘中spill file文件就1~2个，就没有必要调用combiner）

10.Map结束时会把spill出来的多个文件合并成一个，merge过程最多10（默认）个文件同时merge成一个文件，多余的文件分多次merge，merge过程是merge sort的算法。

11.Map端shuffle完毕，数据都有序的存放在磁盘里，等待reducer来拿取

Reducer端

shuffle and sort的过程不仅仅在map端，别忘了reducer端还没拿数据呢，reduce job当然不能开启。

1. Copy phase: reducer的后台进程(default 5个)到被Application Master (Hadoop 2.2)指定的机器上将map的output拷贝到本地，先拷贝到内存，内存满了就拷贝的磁盘。

2. Sort phase(Merge phase): Reduer采用merge sort，将来自各个map的data进行merge， merge成有序的更大的文件。

3. 如果设置过Combiner，merge过程可能会调用Combiner，调不调用要看在磁盘中产生的文件数目是否超过了设定的阈值。(这一点我还没有确认，但Combiner在Reducer端是可能调用。)

4. Reduce phase: reduce job开始，输入是shuffle sort过程merge产生的文件。

大家有什么指教，欢迎大家提出来，让我更进一步