（一）概述

   我们在上一篇blog已经详细的分析了一个作业从用户输入提交命令到到达JobTracker之前的各个过程。在作业到达JobTracker之后初始化之前，JobTracker会通过submitJob方法，为每个作业都创建一个JobInProgress对象(本文以后简称JIP)，用于维护作业的运行时信息以及监控正在运行作业的运行状态和进度。然后检查提交作业的用户是否具有指定队列的作业提交权限，接着检查用户提交作业时候的自己配置的内存量情况是否在管理员的配置范围内，具体点说就是用户提交作业时，配置的mapred.job.map.memory.mb和mapred.job.reduce.memory.mb不能超过管理员配置的

mapred.cluster.max.map.memory.mb和mapred.cluster.max.reduce.memory.mb，一旦提交作业就会直接失败。最后JobTracker通知TaskScheduler初始化作业，作业初始化这一步是我们本文讨论的内容。

（二）作业初始化的时机以及框架调度流程

（1）作业初始化时机

   当JobTracker收到用户提交的作业后，不会马上为其进行初始化而是会交给任务调度器调用EagerTaskInitializationListener对作业进行初始化操作。JobTracker接收到新提交的任务后不对其马上进行初始化是因为考虑到几方面的因素：

（2）框架调度流程

   Hadoop任务调度器的内部实现都是实现了TaskScheduler接口，具有热插拔性。Hadoop的默认任务调度器是JobQueueTaskScheduler（FIFO），管理员可以通过mapred.jobtracker.taskScheduler参数进行配置，其他比较常用的任务调度器还有Fair Scheduler和Capacity Scheduler。

   JobTracker和TaskScheduler之前采用了“观察者模式”，JobTracker作业主题会将“新增作业、更新作业状态、kill掉作业”等命令通知“观察者”TaskScheduler中的EagerTaskInitializationListener和JobQueueJobInProgressListener。其整体调用框架如下图：

用户提交作业到JobTracker之后，通知任务调度器新增作业的代码定位到JobTracker类的addJob(JobID jobId, JobInProgress job)方法，代码如下：

/*** Adds a job to the jobtracker. Make sure that the checks are inplace before* adding a job. This is the core job submission logic* @param jobId The id for the job submitted which needs to be added*/private synchronized JobStatus addJob(JobID jobId, JobInProgress job)throws IOException {totalSubmissions++;synchronized (jobs) {synchronized (taskScheduler) {jobs.put(job.getProfile().getJobID(), job);for (JobInProgressListener listener : jobInProgressListeners) {listener.jobAdded(job);//主要由EagerTaskInitializationListener和JobQueueJobInProgressListener来进行实现}}}......}

上面代码会逐个向已经在JobTracker上注册的监听器发放“新增作业”通知，其中listener的jobAdded(job)方法主要由EagerTaskInitializationListener和JobQueueJobInProgressListener来进行实现。

• EagerTaskInitializationListener中的实现，其主要功能是对将作业添加到初始化队列以及对作业进行排序，代码如下：

/*** We add the JIP to the jobInitQueue, which is processed* asynchronously to handle split-computation and build up* the right TaskTracker/Block mapping.*/
@Override
public void jobAdded(JobInProgress job) {synchronized (jobInitQueue) {jobInitQueue.add(job);//将作业添加到作业初始化队列中resortInitQueue();//根据作业的优先级和提交时间对作业进行排序jobInitQueue.notifyAll();}
}

• JobQueueJobInProgressListener中的实现，其主要功能是将作业添加到作业列表当中。

public void jobAdded(JobInProgress job) {jobQueue.put(new JobSchedulingInfo(job.getStatus()), job);
}

  其中JobSchedulingInfo主要维护了3个变量信息：

private JobPriority priority;
private long startTime;
private JobID id;

（三）作业初始化详细过程分析

任务调度器中的EagerTaskInitializationListener将作业的初始化封装为线程，并且放到了固定大小的线程池(FixedThreadPool)中，其线程池的数量由mapred.jobinit.threads配置，默认大小为4。其线程结构代码如下：

class InitJob implements Runnable {private JobInProgress job;public InitJob(JobInProgress job) {this.job = job;}public void run() {ttm.initJob(job);}
}

变量ttm对象为TaskTrackerManager接口，其实是由JobTracker来实现的，话句话说就是调用了JobTracker里面的initJob方法对作业进行初始化操作的。JobTracker中的initJob方法主要做了2件事情。第一件：构造和初始化Map Task和Reduce Task任务（接下来主要分析的环节），第二件：发通知给任务调度器去更新作业当前状态。大体代码如下：

public void initJob(JobInProgress job) {if (null == job) {LOG.info("Init on null job is not valid");return;}try {JobStatus prevStatus = (JobStatus)job.getStatus().clone();LOG.info("Initializing " + job.getJobID());job.initTasks();//构造和初始化Map Task和Reduce Task任务// Inform the listeners if the job state has changed// Note : that the job will be in PREP state.JobStatus newStatus = (JobStatus)job.getStatus().clone();if (prevStatus.getRunState() != newStatus.getRunState()) {JobStatusChangeEvent event =new JobStatusChangeEvent(job, EventType.RUN_STATE_CHANGED, prevStatus,newStatus);synchronized (JobTracker.this) {updateJobInProgressListeners(event);//发通知给任务调度器去更新作业当前状态}}}

（1）MapTask和ReduceTask任务的构造和初始

   所有Task的普通信息以及运行状态都封装在TaskInProgress类，并且有其对Task进行状态的更新和管理，因此创建TaskInProgress(简称TIP)类其实相当于创建了Task任务。Hadoop中Task任务可以分为Map Task、Reduce Task、Setup Task、Cleanup Task，下面主要分析这几种Task的初始化过程。

• 第一步，读取InputSplit元数据文件job.splitmetainfo，还原成为描述InputSplit具体信息（上一篇blog已经详细说过）的TaskSplitMetaInfo对象，详细代码可以定位到JobInProgress类的initTasks方法（主要负责任务的初始化工作），还原对象的代码如下：

//// read input splits and create a map per a split//TaskSplitMetaInfo[] splits = createSplits(jobId);//还原TaskSplitMetaInfo对象if (numMapTasks != splits.length) {throw new IOException("Number of maps in JobConf doesn't match number of " +"recieved splits for job " + jobId + "! " +"numMapTasks=" + numMapTasks + ", #splits=" + splits.length);}numMapTasks = splits.length;// Sanity check the locations so we don't create/initialize unnecessary tasksfor (TaskSplitMetaInfo split : splits) {NetUtils.verifyHostnames(split.getLocations());}

• 第二步，根据第一步还原的TaskSplitMetaInfo对象信息创建InputSplit.length个TIP对应Map Task，具体代码如下：

maps = new TaskInProgress[numMapTasks];for(int i=0; i < numMapTasks; ++i) {inputLength += splits[i].getInputDataLength();maps[i] = new TaskInProgress(jobId, jobFile,splits[i],jobtracker, conf, this, i, numSlotsPerMap);}

• 第三步，创建mapred.reduce.tasks（默认配置为1）个TIP对应Reduce Task，另外需要注意的是，Reduce Task虽然已经被初始化了，但是不会立即得到调用的，因为它的输入依赖于Map Task的输出，它要等待Map Task完成一定的比例数目才会得到调用，此完成数目比例由参数mappred.reduce.completed.maps配置，默认为0.05。

......// Create reduce tasks//this.reduces = new TaskInProgress[numReduceTasks];//创建Reduce Taskfor (int i = 0; i < numReduceTasks; i++) {reduces[i] = new TaskInProgress(jobId, jobFile,numMapTasks, i,jobtracker, conf, this, numSlotsPerReduce);nonRunningReduces.add(reduces[i]);//将新建的Reudce Task放到没有运行的Reduce Task列表中}
// Calculate the minimum number of maps to be complete before// we should start scheduling reducescompletedMapsForReduceSlowstart = //设定Reduce Task 启动的时机参数(int)Math.ceil((conf.getFloat("mapred.reduce.slowstart.completed.maps",DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART) *numMapTasks));

• 第四步，创建2个TIP分别对应Map Cleanup Task和Reduce Cleanup Task，其作用是清除任务运行之后产生的临时文件信息，他们运行完之后整个作业的状态会由原先的RUNNIN状态变为SUCCEEDED状态,具体代码如下：
  

// create cleanup two cleanup tips, one map and one reduce.cleanup = new TaskInProgress[2];// cleanup map tip. This map doesn't use any splits. Just assign an empty// split.TaskSplitMetaInfo emptySplit = JobSplit.EMPTY_TASK_SPLIT;cleanup[0] = new TaskInProgress(jobId, jobFile, emptySplit,jobtracker, conf, this, numMapTasks, 1);cleanup[0].setJobCleanupTask();// cleanup reduce tip.cleanup[1] = new TaskInProgress(jobId, jobFile, numMapTasks,numReduceTasks, jobtracker, conf, this, 1);cleanup[1].setJobCleanupTask();

• 第五步，创建2个TIP分别对应Map Setup Task和Reduce Setup Task，其作用就是做一些很简单的初始化工作，比如建临时目录等，其具体实现代码在FileOutputCommitter类的setupJob(JobContext context)方法中，如下：

  public void setupJob(JobContext context) throws IOException {JobConf conf = context.getJobConf();Path outputPath = FileOutputFormat.getOutputPath(conf);if (outputPath != null) {Path tmpDir = new Path(outputPath, FileOutputCommitter.TEMP_DIR_NAME);FileSystem fileSys = tmpDir.getFileSystem(conf);if (!fileSys.mkdirs(tmpDir)) {LOG.error("Mkdirs failed to create " + tmpDir.toString());}}}

  当此任务完成之后整个作业的状态会由PREP变成RUNNING。

（四）作业初始化详细过程分析总结

   这里要特别说明一下，上面的图是我按照JobInProgress类的顺序逐步分析的任务初始化结果，但是任务初始化过程的顺序和运行的顺序是不同的。Task运行的过程是这样的：start->TaskInProgress(Setup Task)->TaskInProgress(Map/Reduce Task)->TaskInProgress(CleanUp Task)。

---------------------------------------hadoop源码分析系列------------------------------------------------------------------------------------------------------------

hadoop作业分片处理以及任务本地性分析（源码分析第一篇）

hadoop作业提交过程分析（源码分析第二篇）

hadoop作业初始化过程详解（源码分析第三篇）

JobTracker之作业恢复与权限管理机制（源码分析第四篇）

JobTracker之辅助线程和对象映射模型分析（源码分析第五篇）

---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

参考文献：

[1]《Hadoop技术内幕：深入解析MapReduce架构设计与实现原理》

[2] http://hadoop.apache.org/