第八章 Task运行过程分析

所有Task需要周期性地向TaskTracker汇报最新进度和计数器值，而这正是由Reporter组件实现的，其中Reporter汇报的信息中包含两个部分：任务执行进度以及任务计数器值。

任务执行进度

hadoop采用简单的线性模型计算每个阶段的进度值，对于Map Task而言，作为一个大阶段不再分解，一般实用RecordReader中的getProgress()方法划定执行进度；对于Reduce Task而言，可以分成三个阶段：Shuffle，Sort和Reduce，每个阶段占任务总进度的1/3。

对于任务的Reporter而言，并不会总是每隔一段时间汇报进度和计数器值，而是仅当发现以下两种情况之一时才会汇报：

任务执行进度发生变化；
任务的某个计数器值发生变化；

某个时间间隔内，如果任务执行进度和计数器值均未发生变化，则Task只会RPC调用ping函数探测TaskTracker是否alive，如果一直发生这种情况，TaskTracker认为这个任务处于悬挂状态，直接kill掉。因此我怀疑我们的任务被杀，也是因为这个原因：

AttemptID:attempt_1412848624484_2765_m_000032_0 Timed out after 1200 secsContainer killed by the ApplicationMaster. Container killed on request. Exit code is 143

为了避免这个问题，可以采用下面两种方法：

每隔一段时间调用一次TaskReporter.progress()函数，告诉TaskTracker自己仍然活着；
增大任务超时参数mapreduce.task.timeout参数（默认10分钟，用毫秒表示）。

任务计数器

任务计数器（Counter）是hadoop提供的，用于实现跟踪任务运行进度的全局计数功能，用户可以在自己的应用程序中添加计数器，计数器包括两个部分<name, value>，name表示计数器名称，value表示计数器值，hadoop规定一个作业最多包含120个计数器(修改mapreduce.job.counters.limit)，50个计数器组。

Job的计数器分成两类：内置计数器和用户自定义计数器。

用户可以自定义计数器来进行MR程序的一些数据的统计，在老版本中需要自定义一个枚举，新版本中只需要提供对应的字符串变量即可：

reporter.getCounter(groupName, counterName).increment(1);
reporter.incrCounter(groupName, counterName, 1);

Map Task整体流程

Map Task总共分成5个阶段：

Read阶段：通过用户编写的RecordReader，从输入InputSplit中解析出一个key/value;
Map阶段：将解析出来的key/value交给用户编写的map函数处理，并产生一系列新的key/value；
Collect阶段：当Map阶段处理完数据之后，一般会调用OutputCollector.collect()输出结果，该函数内部会将生成的key/value通过Partitioner分片，写入一个环形缓冲区中；
Spill阶段：当缓冲区满之后，MapReduce会将数据写到本地磁盘中，生成临时文件，在将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并，压缩等操作；
Combine阶段：当所有的数据处理完成后，Map Task会对所有临时文件进行一次合并，确保最终只会生成一个数据文件。

Reduce Task整体流程

Reduce Task也分为5个阶段：

Shuffle阶段：也称为Copy阶段，Reduce Task从各个Map Task上远程拷贝一片数据，并针对某一片数据，如果其大小超过一点阈值，写到磁盘上，否则直接在内存中操作；
Merge阶段：在远程拷贝数据的同时，Reduce Task启动了两个后台线程对内存和磁盘上的文件进行合并，防止内存实用过多或磁盘上的文件过多；
Sort阶段：用户编写的Reduce函数输入数据是按照key进行聚集的一组数据，与Map的数据有所不同。为了将key相同的数据聚集在一起，hadoop会先进行排序。由于事先在Map Task中的输出数据已经局部有序，因此Reduce Task只需要再对所有数据进行一次归并排序即可；
Reduce阶段：该阶段中，Reduce Task将每组数据依次交给用户编写的reduce函数处理；
Write阶段：OutputCollector.collect(key, value)会将计算结果写到HDFS上。

MapReduce的排序

排序是MapReduce框架中最重要的操作之一，Map Task和Reduce Task均会对数据按照key进行排序，这种排序操作属于Hadoop的默认行为，任何应用程序中的数据均会被排序。

Map Task会讲处理的结果暂时放到一个缓冲区中，当缓冲区使用率达到一定的阀值之后，对数据进行一次排序（内存中），并将这些数据以IFile的文件形式写到磁盘上。当全部的数据处理完成后，会对磁盘上的所有文件进行一次合并并排序成一个大的有序文件。

Reduce Task从每个Map Task上远程拷贝相应的数据文件，如果文件大小超过一定的阈值，放到磁盘上，否则暂时放到内存中，如果磁盘上的文件数目达到一定阈值，进行一次合并以声称一个更大的文件。当所有的数据都拷贝完成后，Reduce Task会统一对内存和磁盘上的所有数据进行一次合并。

Hadoop中的文件合并采用了多轮递归合并的方法，每轮选取最小的前io.sort.factor个文件进行合并，并将产生的文件重新加入待合并列表中，直到剩下的文件数量小于io.sort.factor。

在合并的过程中，采用的方式为小顶堆，在小顶堆中选取io.sort.factor，在排序完成后，将结果继续放到该小顶堆中，继续直到剩下的数量小于io.sort.factor，排序到只剩下一个文件。

Reduce端的Shuffle和Merge阶段

Reduce Task中Shuffle阶段和Merge阶段是并行进行的，当远程拷贝数据量达到一定的阈值后，便会出发相应的合并线程对数据进行合并，这个阶段也可以进一步划分为三个子阶段：

准备运行完成的Map Task列表，GetMapEventsThread线程周期性地通过RPC从TaskTracker获取已完成的Map Task列表，为防止网络热点，Reduce Task通过对所有TaskTracker Host进行混洗操作以打乱数据拷贝顺序，并将调整后的Map Task输出数据位置保存到scheduleCopies列表中。
远程拷贝数据，Reduce Task同时启动多个MapOutputCopier线程，这些线程从scheduleCopies列表中获取Map Task输出位置，并通过HTTP Get远程拷贝数据，对于获取的数据分片，如果大小超过一定阈值，则存放到磁盘上，否则直接放到内存中。
合并内存文件和磁盘文件，为了防止内存或者磁盘上的文件过多，Reduce Task启动两个线程分别对磁盘和内存中的文件进行合并。

Reduce端的Sort和Reduce阶段

所有的数据拷贝完成后，数据可能放在内存中或者磁盘上，还不能将数据直接交给用户编写的reduce函数处理，根据MapReduce的语义，Reduce Task需将key值相同的数据聚集在一起，并按组将数据交给reduce函数处理。

为此，Hadoop采用了基于排序的数据聚集策略，各个Map Task已经事先对自己的输出分派呢进行了局部排序，因此，Reduce Task只需要进行一次归并排序即可保证数据整体有序。为了提高效率，Hadoop将Sort阶段和Reduce阶段并行化。

在Sort阶段，Reduce Task为内存和磁盘中的文件建立了小顶堆，保存了指向该小顶堆节点的迭代器，且该迭代器保证了磁盘上的文件数目小于io.sort.factor, 当Reduce阶段开始时，内存中数据量小于最大可用内存(JVM MaxHeapSize)的mapred.job.reduce.input.buffer.precent。

在Reduce阶段，Reduce Task不断地移动迭代器，以将key相同的数据顺次地交给reduce函数处理，期间移动迭代器的过程实际上就是不断调整小顶堆的过程，这样Sort，Reduce就可以并行执行。