hadoop中map到reduce的过程详解

对于Hadoop的MapReduce执行机制，主要分为两部分来处理数据，mapper和reducer阶段，这两个阶段中间有一个非常重要的shuffle过程，这个过程其实是mapreduce的核心部分，因为优化过程主要就是从shuffle处下手。系统将map输出作为输入传给reducer的过程（同时会排序）成为shuffle。shuffle是MapReduce的“心脏”，是奇迹发生的地方。

现就map到reduce的过程做一个大致的解释：

1、运行作业的客户端通过调用getSplits()计算分片，然后将他们发送到jobtracker；

2、jobtracker使用其存储位置信息来调度map任务从而在tasktracker上处理这些分片数据；

3、在tasktracker上，map任务把输入分片传给InputFormat的getRecordReader（）方法来获得这个分片的RecordReader。RecordReader就像是记录上的迭代器，map任务通过调用mapper的run（）方法用一个RecordReader来生成记录的键/值对，进而将该键/值对传给mapper的map方法作为输入。

4、根据自定义的mapper方法，将输入为键值对的数据处理为新的键值对数据，该数据为mapper方法的输出。

5、mapper方法的输出刚开始是写入map任务所有的环形内存缓冲区，待缓冲内容达到指定阈值（默认80%）时，会启动一个溢写的后台线程把内容从缓冲区写入磁盘（与此同时mapper的输出仍在写入缓冲区中，但如果在此期间缓冲区被填满，map会被阻塞直到写磁盘过程完成）。

5.1、在map输出写到缓冲区之前，会进行一个partition操作，即分区操作。MapReduce提供Partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。

5.2、在从缓冲区写到磁盘的过程中，会实现一个排序的过程，即完成MapReduce的默认排序（若key为IntWritable，则排序为自然数的从小到大排序，若key为Text，则为字典顺序排序），这里的排序也是对序列化的字节做的排序 。

5.3、在map输出写到磁盘的溢写过程中，可以加入一次combine操作，将此时统一缓冲区内的输出结果的key进行合并，这样可以减少内存写入磁盘的溢写IO操作。Combiner会优化MapReduce的中间结果，所以它在整个模型中会多次使用。那哪些场景才能使用Combiner呢？从这里分析，Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。所以从我的想法来看，Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。Combiner的使用一定得慎重，如果用好，它对job执行效率有帮助，反之会影响reduce的最终结果。 

6、待全部的mapper输出均写到磁盘后，map会把这多个临时文件合并，即做merge操作，注意，这里的merge操作只是简单的合并，如果没有在该处设置Combiner，是不会对相同key进行压缩的，所以可能会有相同的key出现。merge操作就是对于同样的key，其value变为list，把多个value放在list中。这种key/value的形式就是reduce的输入数据格式。

至此，map端的所有工作都已结束，最终生成的这个文件也存放在TaskTracker够得着的某个本地目录内。每个reduce task不断地通过RPC从JobTracker那里获取map task是否完成的信息，如果reduce task得到通知，获知某台TaskTracker上的map task执行完成，Shuffle的后半段过程开始启动。 

7、reducer通过HTTP方式得到输出文件的分区。Reduce进程启动一些数据copy线程(Fetcher)，通过HTTP方式请求map task所在的TaskTracker获取map task的输出文件，并行地获取map输出。因为map task早已结束，这些文件就归TaskTracker管理在本地磁盘中。 

8、合并操作：这里的merge如map端的merge动作，只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活，它基于JVM的heap size设置，因为Shuffle阶段Reducer不运行，所以应该把绝大部分的内存都给Shuffle用。

这里需要强调的是，merge有三种形式：1)内存到内存  2)内存到磁盘  3)磁盘到磁盘。默认情况下第一种形式不启用，让人比较困惑，是吧。当内存中的数据量到达一定阈值，就启动内存到磁盘的merge。与map 端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。

9、reducer会一直进行合并merge操作，直到所有的map的输出结果都被合并完毕为止第二种merge方式一直在运行，直到没有map端的数据时才结束，然后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。 

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

11、合并完之后，reducer会直接把数据输入reduce函数，而不会把最后合并的一个大文件再次写入磁盘。最后的合并可以来自北村和磁盘片段。

12、在reduce阶段中，对已排序输出的每个键调用reduce函数。此阶段的输出直接写到输出文件系统，一般为HDFS。如果采用HDFS,由于tasktracker节点也运行数据节点，所以第一个块副本将被写到本地磁盘。

 在Hadoop这样的集群环境中，大部分map task与reduce task的执行是在不同的节点上。当然很多情况下Reduce执行时需要跨节点去拉取其它节点上的map task结果。如果集群正在运行的job有很多，那么task的正常执行对集群内部的网络资源消耗会很严重。这种网络消耗是正常的，我们不能限制，能做的就是最大化地减少不必要的消耗。还有在节点内，相比于内存，磁盘IO对job完成时间的影响也是可观的。从最基本的要求来说，我们对Shuffle过程的期望可以有： 

• 完整地从map task端拉取数据到reduce 端。
• 在跨节点拉取数据时，尽可能地减少对带宽的不必要消耗。
• 减少磁盘IO对task执行的影响。