MapReduce执行流程

核心中的核心：大问题拆分成小问题，然后分布式的并行执行

两个阶段：

1、mapper阶段： 提取数据，赋予特征 映射  value ====>   key,value mapreduce 框架是怎么把相同特征的数据组合到一起来，然后交给reduceTask执行一次聚合操作（这里到底是怎么操作的呢？）

2、reducer阶段: 把相同特征的数据进行聚合操作 key, (value, value, ...)

目录

 任务提交

Map端分析

 Reduce端的shuffle

---

 

 任务提交

1. client提交一个job 向JobTracker请求一个JobID
2. JobTracker接受到请求后，将其放到一个任务队列中然后等待TaskTracker执行这里也是分配给TaskTracker的但是不是随便分配的（运算移动，数据不移动）
3. TaskTracker每隔一段时间会给JobTracker发送一个心跳，告诉JobTracker它依然在运行，同时心跳中还携带者很多信息，比如当前map任务完成的进度等信息。当JobTracker收到作业的最后一个任务完成信息时，便把该作业设置成“成功”。当JobTracker查询状态时，它将得知任务已完成，便显示一条消息给用户。

Map端分析

1. 每个输入分片会让一个map任务来处理map输出的结果会暂且放在一个环形内存缓冲区中（该缓冲区的大小默认为100M，由io.sort.mb属性控制），当该缓冲区快要溢出时（默认为缓冲区大小的80%，由io.sort.spill.percent属性控制），会在本地文件系统中创建一个溢出文件，将该缓冲区中的数据写入这个文件。
2. 在map task执行时，它的输入数据来源于HDFS的block，当然在MapReduce概念中，map task只读取split。split与block对应关系可能是多对一，默认是一对一。在wordcount例子里，假设map的输入数据都是是像“aaa”这样的字符串。在经过mapper的运行后，我们得知mapper的输出是这样一个key/value对：key是“aaa”，value是数值1。因为当前map端只做加1的操作，在reduce task里采取合并结果集。前面我们知道这个job有3个reduce task。那到底当前的“aaa”究竟该丢给哪个reduce去处理呢？是需要现在做决定的MapReduce提供Partitioner接口，作用就是根据key或value及reduce的数量来决定当前的输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数据取模【 'aaa'.hashcode()%reducetask数】。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以定制并设置到job上
3. “aaa”经过Partition后返回0，也就是这对值应当交由第一个reduce来处理。接下来，需要将数据写入内存缓冲区中，缓冲区的作用是批量收集map结果，减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然，写入之前，key与value值都会被序列化成字节数组。
4. 内存缓冲区是有大小限制的，默认是100MB。当map task的输出结果很多时，就可能会撑爆内存，所以需要在一定条件下将缓冲区中的数据临时写入磁盘，然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为spill，中文可理解为溢写。溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。比例默认是0.8，也就是当缓冲区的数据值已经达到阈值（buffer size * spill percent = 100MB * 0.8 = 80MB），溢写线程启动，锁定这80MB的内存，执行溢写过程。map task的输出结果还可以往剩下的20MB内存中写，互不影响。
5. 当溢写线程启动后，需要对这80MB空间内的key做排序（sort）。排序是MapReduce模型默认的行为，这里的排序也是对序列化的字节做的排序。

6.  因为map task的输出是需要发送到不同的reduce端去，而内存缓冲区没有对将发送到相同reduce端的数据做合并，那么这种合并应该是体现在磁盘文件中的。从官方图上也可以看到写到磁盘中的一些文件是对不同的reduce端的数值做过合并。所以溢写过程一个很重要的细节在于，如果有很多个key/value对需要发送到某个reduce端去，那么需要将这些key/value值拼接到一块，减少与partition相关的索引记录。

   　　在针对每个reduce端而合并数据时，有些数据可能像这样：“aaa”/1，“aaa”/1。对于wordcount例子，只是简单地统计单词出现的次数，如果在同一个map task的结果中有很多像“aaa”一样出现多次的key，我们就应该把它们的值合并到一块，这个过程叫reduce也叫combine。但MapReduce的术语中，reduce只值reduce端执行从多个map task取数据做计算的过程。除reduce外，非正式地合并数据只能算作combine了。其实大家知道的，MapReduce中将Combiner等同于Reducer。

   　　如果client设置过Combiner，那么现在就是使用Combiner的时候了。将有相同key的key/value对的value加起来，减少溢写到磁盘的数据量。Combiner会优化MapReduce的中间结果，所以它在整个模型中会多次使用。那哪些场景才能使用Combiner呢？从这里分析，Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。所以从我的想法来看，Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。Combiner的使用一定得慎重，如果用好，它对job执行效率有帮助，反之会影响reduce的最终结果。

7. 每次溢写会在磁盘上生成一个溢写文件，如果map的输出结果真的很大，有多次这样的溢写发生，磁盘上相应的就会有多个溢写文件存在。当map task真正完成时，内存缓冲区中的数据也全部溢写到磁盘中形成一个溢写文件。最终磁盘中会至少有一个这样的溢写文件存在（如果map的输出结果很少，当map执行完成时，只会产生一个溢写文件），因为最终的文件只有一个，所以需要将这些溢写文件归并到一起，这个过程就叫Merge。Merge是怎样的？如前面的例子，“aaa”从某个map task读取过来时值是3，从另外一个map读取时值是4，因为他们有相同的key，所以要merge成group。

   　　什么是group：对于“aaa”就是像真阳的：{“aaa”，[3,4,2,...]}，数组中的值就是从不同的溢写文件中读取出来的，然后再把这些值加起来。请注意，因为merge是将多个溢写文件合并到一个文件，所以可能也有相同的key存在，在这个过程中，如果client设置过Combiner，也会使用Combiner来合并相同的key。

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

 Reduce端的shuffle

1reduce会接收到不同map任务传来的数据，并且每个map传来的数据都是有序的。如果reduce端接收的数据量相当小，则直接存储在内存中（缓冲区大小由mapred.job.shuffle.input.buffer.percent属性控制，表示用作此用途的堆空间百分比），如果数据量超过了该缓冲区大小的一定比例（由mapred.job.shuffle.merg.percent决定），则对数据合并后溢写到磁盘中。

2 随着溢写文件的增多，后台线程会将它们合并成一个更大的有序的文件，这样做是为了给后面的合并节省空间。其实不管在map端还是在reduce端，MapReduce都是反复地执行排序，合并操作，现在终于明白了有些人为什么会说：排序是hadoop的灵魂。

3 合并的过程中会产生许多的中间文件（写入磁盘了），但MapReduce会让写入磁盘的数据尽可能地少，并且最后一次合并的结果并没有写入磁盘，而是直接输入到reduce函数。

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