MapReduce V1：MapTask执行流程分析

final TaskUmbilicalProtocol umbilical =        taskOwner.doAs(new PrivilegedExceptionAction<TaskUmbilicalProtocol>() {          @Override          public TaskUmbilicalProtocol run() throws Exception { // 建立Child到TaskTracker的RPC连接            return (TaskUmbilicalProtocol)RPC.getProxy(TaskUmbilicalProtocol.class,                TaskUmbilicalProtocol.versionID,                address,                defaultConf);          }      });  ... ...      JvmContext context = new JvmContext(jvmId, pid); // 根据启动Child VM命令行传递的参数构造一个JvmContext对象  ... ...          JvmTask myTask = umbilical.getTask(context); // 基于umbilical获取到一个JvmTask  ... ...          task = myTask.getTask(); // 通过JvmTask获取到MapTask或ReduceTask

上面代码中，JvmTask中就包含了一个Task，也就是task，它可能是MapTask或ReduceTask。看一下在org.apache.hadoop.mapred.Child中运行Task的基本代码，如下所示：

// Create a final reference to the task for the doAs block          final Task taskFinal = task;          childUGI.doAs(new PrivilegedExceptionAction<Object>() {            @Override            public Object run() throws Exception {              try {                // use job-specified working directory                FileSystem.get(job).setWorkingDirectory(job.getWorkingDirectory());                taskFinal.run(job, umbilical);        // 这行是核心代码，运行实际的MapTask或ReduceTask              } finally {                TaskLog.syncLogs                  (logLocation, taskid, isCleanup, logIsSegmented(job));                TaskLogsTruncater trunc = new TaskLogsTruncater(defaultConf);                trunc.truncateLogs(new JVMInfo(                    TaskLog.getAttemptDir(taskFinal.getTaskID(),                      taskFinal.isTaskCleanupTask()), Arrays.asList(taskFinal)));              }                return null;            }          });

我们关注执行MapTask，上面，通过调用MapTask的run方法，来实际启动MapTask的运行。

MapTask整体执行流程

@Override    public void run(final JobConf job, final TaskUmbilicalProtocol umbilical)      throws IOException, ClassNotFoundException, InterruptedException {      this.umbilical = umbilical;        // start thread that will handle communication with parent      TaskReporter reporter = new TaskReporter(getProgress(), umbilical, jvmContext); // 创建TaskReporter对象      reporter.startCommunicationThread();      boolean useNewApi = job.getUseNewMapper();      initialize(job, getJobID(), reporter, useNewApi); // 初始化MapTask        // check if it is a cleanupJobTask      if (jobCleanup) {        runJobCleanupTask(umbilical, reporter); // 运行JobCleanupTask        return;      }      if (jobSetup) {        runJobSetupTask(umbilical, reporter); // 运行JobSetupTask        return;      }      if (taskCleanup) {        runTaskCleanupTask(umbilical, reporter); // 运行TaskCleanupTask        return;      }        if (useNewApi) {        runNewMapper(job, splitMetaInfo, umbilical, reporter); // 运行MapTask      } else {        runOldMapper(job, splitMetaInfo, umbilical, reporter);      }      done(umbilical, reporter); // Task运行完成    }

初始化Task分析

这里，主要分析在MapTask的run方法中，调用initialize方法的初始化逻辑。先看在调用initialize方法之前，首先创建了一个TaskReporter线程对象，该对象又是基于TaskUmbilicalProtocol umbilical来实现将Task运行状态汇报给TaskTracker。在initialize方法中的初始化流程如下所示：

1. 根据JobConf job与JobID id，以及TaskReporter，创建一个JobContext对象
2. 创建一个TaskAttemptContext对象
3. 如果Task状态为TaskStatus.State.UNASSIGNED，修改为TaskStatus.State.RUNNING
4. 根据JobConf创建OutputFormat，以及OutputCommitter
5. 为该Task创建Job的输出目录等内容
6. 初始化ResourceCalculatorPlugin，用来计算Task运行过程中对节点资源的使用情况

运行JobCleanupTask

• 运行JobCleanupTask首先将当前Task运行阶段设置为CLEANUP，并向TaskTracker汇报状态变更
• 如果Job的状态是JobStatus.State.FAILED，则删除在该节点上运行Task所产生的临时数据
• 如果Job的状态是JobStatus.State.KILLED，同样删除临时数据，并在该Job对应的目录下创建_SUCCESS文件，标识Job成功
• 最后，向TaskTracker汇报状态，更新相关TIP数据结构状态，并释放锁占用的资源，以供其他Task运行所使用

运行JobSetupTask

运行TaskCleanupTask

运行MapTask

private <INKEY,INVALUE,OUTKEY,OUTVALUE> void runNewMapper(final JobConf job,                      final TaskSplitIndex splitIndex,                      final TaskUmbilicalProtocol umbilical,                      TaskReporter reporter ) throws IOException, ClassNotFoundException, InterruptedException

其中，JobConf job包含了该Job的配置信息，TaskSplitIndex splitIndex包含了该MapTask所处理的InputSplit的信息（包括splitLocation和startOffset），TaskUmbilicalProtocol umbilical是与TaskTracker通信的代理对象，TaskReporter reporter是一个与TaskTracker通信的线程。上面的序列图，所描述的具体处理流程，如下所示：

1. 创建TaskAttemptContext对象
2. 通过反射，创建Mapper<INKEY,INVALUE,OUTKEY,OUTVALUE>对象
3. 通过反射，创建InputFormat<INKEY,INVALUE>对象
4. 根据TaskSplitIndex splitIndex ，创建InputSplit对象
5. 创建RecordReader<INKEY,INVALUE>对象，用来读取输入的InputSplit对应的文件
6. 创建RecordWriter<K, V>对象，用来将处理后的数据写入文件系统
7. 创建Mapper.Context对象，可以在编写MapReduce程序中，实现Mapper时使用
8. 初始化RecordReader<INKEY,INVALUE>对象
9. 执行Mapper的run方法，调用用户编写的MapReduce程序的Mapper中的处理逻辑，内部循环调用map方法
10. 回收资源，关闭相关的流对象

下面，我们详细看一下，Mapper处理过程中相关的要点：

• Mapper.Context结构

private RecordReader<KEYIN,VALUEIN> reader;    private InputSplit split;

通过使用一个RecordReader，能够读取InputSplit对应的HDFS上的Block文件数据。TaskInputOutputContext中包含如下信息：

private RecordWriter<KEYOUT,VALUEOUT> output;    private StatusReporter reporter;    private OutputCommitter committer;

可见，在该层能够实现将Task运行的统计，通过StatusReporter以Counter的形式收集，并提供进度（Progress）获取接口。同时，使用RecordWriter将Mapper的输出写入到文件系统。OutputCommitter是一个比较重要的对象，该抽象类代码，如下所示：

public abstract class OutputCommitter {      public abstract void setupJob(JobContext jobContext) throws IOException;      public void commitJob(JobContext jobContext) throws IOException {      cleanupJob(jobContext);    }      @Deprecated    public void cleanupJob(JobContext context) throws IOException { }      public void abortJob(JobContext jobContext, JobStatus.State state) throws IOException {      cleanupJob(jobContext);    }      public abstract void setupTask(TaskAttemptContext taskContext) throws IOException;      public abstract boolean needsTaskCommit(TaskAttemptContext taskContext) throws IOException;      public abstract void commitTask(TaskAttemptContext taskContext) throws IOException;      public abstract void abortTask(TaskAttemptContext taskContext) throws IOException;  }

• Mapper处理流程

通过调用mapper.run(mapperContext);进入Mapper的执行流程，我们首先看一下，Mapper提供的接口的声明，代码如下所示：

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {      public class Context extends MapContext<KEYIN,VALUEIN,KEYOUT,VALUEOUT> { // Mapper.Context是Mapper运行时上下文对象，包含了运行MapTask所需要的基本信息      public Context(Configuration conf, TaskAttemptID taskid,                     RecordReader<KEYIN,VALUEIN> reader,                     RecordWriter<KEYOUT,VALUEOUT> writer,                     OutputCommitter committer,                     StatusReporter reporter,                     InputSplit split) throws IOException, InterruptedException {        super(conf, taskid, reader, writer, committer, reporter, split);      }    }      /**     * Called once at the beginning of the task.     */    protected void setup(Context context) throws IOException, InterruptedException { // 执行map方法之前，进行初始化工作      // NOTHING    }      /**     * Called once for each key/value pair in the input split. Most applications     * should override this, but the default is the identity function.     */    @SuppressWarnings("unchecked")    protected void map(KEYIN key, VALUEIN value, Context context) throws IOException, InterruptedException { // 用户实际实现的map处理逻辑      context.write((KEYOUT) key, (VALUEOUT) value);    }      /**     * Called once at the end of the task.     */    protected void cleanup(Context context ) throws IOException, InterruptedException { // 执行map完成之后，进行清理      // NOTHING    }      /**     * Expert users can override this method for more complete control over the     * execution of the Mapper.     * @param context     * @throws IOException     */    public void run(Context context) throws IOException, InterruptedException { // 该方法是程序的驱动入库，循环调用map方法处理每一个键值对数据      setup(context);      try {        while (context.nextKeyValue()) {          map(context.getCurrentKey(), context.getCurrentValue(), context);        }      } finally {        cleanup(context);      }    }  }

上面run方法中，输入的InputSplit对应的每一个记录，循环调用map方法，map方法中是用户实现的MapReduce的Mapper中的代码。

Mapper输出流程分析

在runNewMapper方法中可以看到，如果Job有0个ReduceTask，则Mapper将结果直接写入到HDFS文件；而如果Job有大于0个ReduceTask，则Mapper将结果输出到本地，是中间结果。代码如下所示：

if (job.getNumReduceTasks() == 0) {           output =             new NewDirectOutputCollector(taskContext, job, umbilical, reporter);        } else {          output = new NewOutputCollector(taskContext, job, umbilical, reporter);        }

上面，NewDirectOutputCollector实际上将Mapper的输出直接写入到HDFS文件；而NewOutputCollector将Mapper输出写入到本地文件，我们看一下NewOutputCollector是如何实现Mapper输出落地的，它的构造方法，代码如下所示：

NewOutputCollector(org.apache.hadoop.mapreduce.JobContext jobContext,                         JobConf job,                         TaskUmbilicalProtocol umbilical,                         TaskReporter reporter) throws IOException, ClassNotFoundException {        collector = new MapOutputBuffer<K,V>(umbilical, job, reporter); //         partitions = jobContext.getNumReduceTasks(); // 根据ReduceTask个数计算Mapper输出分区个数        if (partitions > 0) { // 分区个数大于0，创建一个Partitioner对象          partitioner = (org.apache.hadoop.mapreduce.Partitioner<K,V>) ReflectionUtils.newInstance(jobContext.getPartitionerClass(), job);        } else {          partitioner = new org.apache.hadoop.mapreduce.Partitioner<K,V>() {            @Override            public int getPartition(K key, V value, int numPartitions) {              return -1;            }          };        }      }

可以看出，上面的collector里面的逻辑非常核心，它是MapOutputBuffer类实例，该类实现也是相对复杂的。接着，我们从如下几个方面详细分析：

• 使用Buffer存储键值对数据

Mapper输出后，会首先将键值对数据写入到一个Buffer中，通过其它的几个Buffer来跟踪键值对的在Buffer中偏移位置信息，以及键值对所属的分区（Partition）。我们看一下NewOutputCollector调用write方法输出键值对数据的逻辑：

@Override      public void write(K key, V value) throws IOException, InterruptedException {        collector.collect(key, value, partitioner.getPartition(key, value, partitions));      }

通过partitioner计算一个输出键值对所属的分区，然后调用MapOutputBuffer的collect方法实现Mapper中间结果数据输出。下面，我们看MapOutputBuffer涉及到的一些存储结构，其中包含3个buffer，如下代码所示：

private final int[] kvoffsets;     // indices into kvindices      private final int[] kvindices;     // partition, k/v offsets into kvbuffer      private byte[] kvbuffer;

上面的数组结构，kvbuffer用来存储Map输出的键值对数据，kvoffsets用来存储Map输出的键值对数据在kvbuffer中的偏移位置，kvindices用来记录一个键值对对应的分区、键开始、值开始位置的数据结构，具体我们会在后面详细说明。首先，看一下用来定义和限制这些buffer的一些基本默认配置参数：

io.sort.mb             = 100MB  io.sort.record.percent = 0.05  io.sort.spill.percent  = 0.8    sortmb   = 100MB  rEcper   = 0.05  spillper = 0.8    RECSIZE  = 16  ACCTSIZE = 3

基于这些配置的参数，计算得到一些buffer的大小及其限制配置，为了直观我们通过使用MB来表示大小，如下所示：

maxMemUsage = sortmb <<20 = 104857600 = 100MB  recordCapacity = maxMemUsage * recper = 104857600 * 0.05 = 5242880 = 5MB  recordCapacity -= recordCapacity % RECSIZE = 5242880 - 5242880 % 16 = 5242880 = 5MB  maxMemUsage – recordCapacity = 104857600 - 5242880 = 99614720 = 95MB    softBufferLimit = ((maxMemUsage – recordCapacity) * spillper) = 99614720 * 0.8 = 79691776 = 76MB  softRecordLimit = (recordCapacity * spillper) = 5242880 * 0.8 = 4194304 = 4MB

• 将Buffer中键值对数据写入本地磁盘文件

</div>