前言

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。
本文不打算深入地详细分析TaskTracker某个具体的处理流程，而是概要地分析TaskTracker在MapReduce框架中的主要负责处理那些事情，是我们能够在宏观上了解TaskTracker端都做了哪些工作。我尽量将TaskTracker端的全部要点内容提出来，但是涉及到详细的分析，只是点到为止，后续会对相应模块的处理流程结合代码进行分析。
TaskTracker主要负责MapReduce计算集群中Task运行的管理，所以TaskTracker要管理的事情比较多。一个MapReduce Job由很多的Task组成，而一个Job的所有Task被分成几个相斥的子集，每个子集被分配到某一个TaskTracker上去运行，所以一个TaskTracker管理运行了一个Job的所有Task的一个子集，也就是说TaskTracker不仅要维护每个Job对应的一个Task的子集，还要维护这些Task所属的Job的运行状态，对于Job/Task的状态的管理都是与JobTracker通过RPC通信保持状态的同步。
下面是TaskTracker端的主要组件，如下图所示：

为了了解TaskTracker中各个组件都负责处理哪些工作，我们通过下表来简要地说明各个组件的功能，如下表所示：

通过上表，我们可以了解到TaskTracker端各个组件的基本功能，也稍微了解到组件之间的一些关系。下面，我们从TaskTracker抽象层次的视角，来分析组件之间的关系和交互，概要地描述一些主要的处理流程：

• TaskTracker处理心跳响应
• MapReduce Job恢复运行
• Task隔离运行
• 启动MapTask过程
• 启动ReduceTask过程

下面，我们分别分析上述列举的5个处理流程：

TaskTracker处理心跳响应

TaskTracker周期性地向JobTracker发送心跳报告，将TaskTracker上运行的Task的状态信息、节点资源信息、节点健康状况信息封装到TaskTrackerStatus对象中，通过RPC调用heartbeat将心跳发送到JobTracker端，并返回HeartbeatResponse，其中心跳响应对象中包包含了JobTracker分配的任务，通过TaskAction这种指令（包括：LaunchTaskAction/KillTaskAction/CommitTaskAction）列表的方式进行指派。TaskTracker解析RPC调用返回的心跳响应，根据TaskAction指令列表，执行具体的操作。TaskTracker处理心跳响应的流程，如下序列图所示：

TaskTracker收到心跳响应，首先会检查是否存在需要恢复的Job，如果存在，则会检查要进行恢复的Job的状态，从而将需要进行恢复的Job对应的Task加入到恢复队列中，等待调度运行。接着，TAskTracker会检查TaskAction指令的类型，根据其实际类型，执行对应的处理流程：

• 如果是LaunchTaskAction，则启动Task
• 如果是KillTaskAction，则杀掉Task，修改TaskTracker维护的Job及Task状态，并清理临时数据
• 如果是KillJobAction，则杀掉该Job，说明该Job已经完成（成功/失败），修改TaskTracker维护的Job及Task状态，并清理临时数据
• 如果是CommitTaskAction，则说明对应的该Task已经执行完成，修改TaskTracker维护的Task即Job状态，最后还要清理临时数据

由于某些重要的处理流程，如启动一个Task的详细流程，我们会在后续单独写几篇文章，用更加合适的方式来详细分析。

MapReduce Job恢复运行

这里，我们介绍一下MapReduce计算中是如何实现Job的恢复的，包括JobTracker端和TaskTracker端之间的简单交互流程。JobTracker存在一个系统目录（System Directory），默认值为/tmp/hadoop/mapred/system，也可以根据配置项mapred.system.dir指定该值。当JobClient提交一个Job到JobTracker时，JobTracker会首先将该Job的信息写入到JobTracker的系统目录下，每个Job对应一个以Job ID为名称的子目录，以便JobTracker因为重启，能够恢复这些Job的运行。我们可以看一下JobTracker中submitJob方法中保存Job信息的实现，代码如下所示：

// Store the job-info in a file so that the job can be recovered
// later (if at all)
// Note: jobDir & jobInfo are owned by JT user since we are using
// his fs object
if (!recovered) {
  Path jobDir = getSystemDirectoryForJob(jobId); // Job目录，例如：/tmp/hadoop/mapred/system/job_200912121733_0002
  FileSystem.mkdirs(fs, jobDir, new FsPermission(SYSTEM_DIR_PERMISSION));
  FSDataOutputStream out = fs.create(getSystemFileForJob(jobId)); // Job文件，例如：/tmp/hadoop/mapred/system/job_200912121733_0002/job-info
  jobInfo.write(out); // 将JobInfo结构对应的数据写入到Job文件
  out.close();
}

上面代码中，JobInfo主要包含了JobID信息、用户名称、Job提交目录（例如，/tmp/hadoop/mapred/staging/shirdrn/.staging/job_200912121733_0002/ ，该目录是在JobClient提交Job时在HDFS上创建的，用于将该Job所需要的资源都拷贝到该Job对应的提交目录下面，便于后续JobTracker能够读取这些数据）。如果JobTracker因为某些原因重新启动了，那么在JobTracker重启之后，需要从JobTracker的系统目中读取这些Job的信息，以便能够恢复这些尚未完成的Job的运行，并以HeartbeatResponse的结构，在TaskTracker发送Heartbeat的时候响应给TaskTracker，TaskTracker解析响应数据，然后去恢复这些Job的运行。上面的序列图中，我们可以看到，当TaskTracker发送Heartbeat并收到响应后，从HeartbeatResponse中解析取出需要Recovered的Job，并进行处理，代码如下所示：

// Check if the map-event list needs purging
Set<JobID> jobs = heartbeatResponse.getRecoveredJobs(); // 获取到需要Recovered的Job列表
if (jobs.size() > 0) {
  synchronized (this) {
    // purge the local map events list
    for (JobID job : jobs) {
      RunningJob rjob;
      synchronized (runningJobs) { // TaskTracker维护的当前在其上运行的Job列表
        rjob = runningJobs.get(job);        
        if (rjob != null) {
          synchronized (rjob) {
            FetchStatus f = rjob.getFetchStatus();
            if (f != null) {
              f.reset();
            }
          }
        }
      }
    }
 
    // Mark the reducers in shuffle for rollback
    synchronized (shouldReset) {
      for (Map.Entry<TaskAttemptID, TaskInProgress> entry
           : runningTasks.entrySet()) {
        if (entry.getValue().getStatus().getPhase() == Phase.SHUFFLE) {
          this.shouldReset.add(entry.getKey()); // 将处于SHUFFLE阶段的Task放到shouldReset集合中
        }
      }
    }
  }
}

通过上面的代码我们可以看到，当JobTracker重启的时候，已经在TaskTracker上运行的属于某些Job的Task可能无法立即感知到，对应的Job仍然存在于TaskTracker的runningJobs集合中。在JobTracker重启之后，TaskTracker所发送的第一个Heartbeat返回的响应数据中，应该会存在需要Recovered的Job列表，所以这时在TaskTracker端只需要从runningJobs中取出需要Recovered的Job，并查看其是否存在Fetch状态，如果存在，应该重新设置状态（主要对应于MapEventsFetcherThread 维护的TaskCompletionEvent列表，触发ReduceTask拉取MapTask的输出中间结果），以便该Job的各个Task恢复运行。如果该ReduceTask正在运行于SHUFFLE阶段，需要将对应的Job的MapTask的输出拷贝到该ReduceTask所在的节点上，通过调用FetchStatus的reset方法重置状态，这样就重新恢复了ReduceTask的运行。

Task隔离运行

由于MapReduce用户程序包含用户代码，可能会存在Bug，为了不因为用户代码存在的Bug影响TaskTracker服务，所以MapReduce采用了隔离Task运行的方式来运行MapTask/ReduceTask。在运行Task时，会单独创建一个独立的JVM实例，让Task的代码再该JVM实例中加载运行，TaskTracker需要跟踪该JVM实例中运行的Task的状态。在TaskTracker端，加载一个运行Task的JVM实例，是通过org.apache.hadoop.mapred.Child类来实现。下面，我们看一下Child类是如何实现Task加载运行的，如下面序列图所示：

Child类包含一个入口主方法main，在运行的时候需要传递对应的参数，来运行MapTask和ReduceTask，通过上面序列图我们可以看出，命令行输入如下5个参数：

• host：表示TaskTracker节点的主机名称
• port：表示TaskTracker节点RPc端口号
• taskID：表示启动的Task对应的TaskAttemptID，标识一个Task的一个运行实例
• log location：表示该Task运行实例对应的日志文件的路径
• JVM ID：表示该Task实例对应的JVMId信息，包括JobID、Task类型（MapTask/ReduceTask）、JVM编号（标识该JVM实例对应的id）

有了上述参数，就可以获取到一个Task运行所需要的全部资源，如一个Task处理哪一个Split，一个Task对应的Job配置信息，还可以方便TaskTracker监控该Task实例所在的JVM的状态。该Child创建时，会创建一个到TaskTracker的RPC代理对象，通过该RPC连接向TaskTracker汇报Task执行进度及其状态信息。然后，一切运行Task的基本条件都已经具备，接下来从该Task对应的Job的代码（job.jar）开始加载任务处理类，如果是MapTask则执行MapTask运行的处理流程，如果是ReduceTask则执行ReduceTask的处理流程，最后，断开Task汇报状态的RPc连接，Task运行结束。

启动MapTask过程

在Child类中加载启动Task，如果是MapTask，则执行MapTask对应的处理流程，如下序列图所示：

启动一个MapTask运行，包含4个阶段，我们通过运行各个阶段的方法来表示：

• runJobCleanupTask()：清理Job对应的相关目录和文件
• runJobSetupTask()：创建Job运行所需要的相关目录和文件
• runTaskCleanupTask()：清理一个Task对应的工作目录下与Task相关的目录或文件
• runNewMapper()/runOldMapper()：调用用户编写的MapReduce程序中的Mapper中的处理逻辑

在MapTask运行过程中， 如果阶段或者状态发生变化，要与TaskTracker进行通信，汇报状态，并更新TaskTracker维护的关于Task和Job对应的状态数据。最后，Task运行完成，也要通知TaskTracker。

启动ReduceTask过程

在Child类中加载启动Task，如果是ReduceTask，则执行ReduceTask对应的处理流程，如下序列图所示：

启动一个ReduceTask运行，与MapTask的处理流程有很大的不同，它包含3个阶段，如下所示：

• copy阶段：从运行MapTask的TaskTracker节点，拷贝属于该ReduceTask对应的Job所包含的MapTask的输出中间结果数据，这些数据存储在该Reduce所在TaskTracker的本地文件系统（可能会放在内存中），为后续阶段准备数据
• sort阶段：对从MapTask拉取过来的数据进行合并、排序
• reduce阶段：调用用户编写的MapReduce程序中的Reducer中的处理逻辑

执行reduce阶段，比MapTask复杂的多。在ReduceTask运行过程中，也会周期性地与TaskTracker通信，汇报Task运行进度和状态，以保证与TaskTracker所维护的Task的状态数据同步。当ReduceTask完成后，如果有输出的话，最终的结果数据会输出到HDFS中保存。[原创：时延军（包含链接：http://shiyanjun.cn）]

觉得文章不错的话，可以转发关注小编一手，之后持续更新干货文章！！