MapReduce调度与执行原理之作业初始化

前言 ：本文旨在理清在Hadoop中一个MapReduce作业（Job）在提交到框架后的整个生命周期过程，权作总结和日后参考，如有问题，请不吝赐教。本文不涉及Hadoop的架构设计，如有兴趣请参考相关书籍和文献。在梳 理过程中，我对一些感兴趣的源码也会逐行研究学习，以期强化基础。
作者 ：Jaytalent
开始日期 ：2013年9月9日

参考资料：【1】《Hadoop技术内幕--深入解析MapReduce架构设计与实现原理》董西成

                  【2】Hadoop 1.0.0 源码

                            【3】《Hadoop技术内幕--深入解析Hadoop Common和HDFS架构设计与实现原理》蔡斌 陈湘萍

在 上一篇文章中，作业准备提交到JobTracker了。本文关注作业在提交到JobTracker后且在执行前经历了哪些事情。

一个MapReduce作业的生命周期大体分为5个阶段 【1】：

1.  作业提交与初始化

2. 任务调度与监控

3. 任务运行环境准备

4. 任务执行

5. 作业完成

一、作业提交与初始化

在  JobClient.submitJobInternal方法中，最后一步就是将作业提交到JobTracker：

 status = jobSubmitClient.submitJob(jobId, submitJobDir.toString(), jobCopy.getCredentials());

这个方法所属的接口在 上一篇文章中有所提及，其实现有两个：JobTracker和LocalRunner。LocalRunner是用于执行本地作业的，当Hadoop配置为本地模式时采用该类处理作业。我们关注JobTracker.submitJob方法。这里多说一句，在JobClient对象初始化时有如下代码：

 /**
   * Connect to the default {@link JobTracker}.
   * @param conf the job configuration.
   * @throws IOException
   */
  public void init(JobConf conf) throws IOException {
    String tracker = conf.get("mapred.job.tracker", "local");
    tasklogtimeout = conf.getInt(
      TASKLOG_PULL_TIMEOUT_KEY, DEFAULT_TASKLOG_TIMEOUT);
    this.ugi = UserGroupInformation.getCurrentUser();
    if ("local".equals(tracker)) {
      conf.setNumMapTasks(1);
      this.jobSubmitClient = new LocalJobRunner(conf);
    } else {
      this.jobSubmitClient = createRPCProxy(JobTracker.getAddress(conf), conf);
    }        
  }

可以看到，tracker变量是从mapred.job.tracker配置获取值的，默认值为字符串“local”。因此，如果没有在配置文件配置这个值或者JobConf对象中没有添加配置文件（通常为mapred-site.xml）资源，jobSubmitClient就会使用LocalJobRunner进行初始化。
回到正题， JobTracker.submitJob方法会做如下工作：

a. 首先创建JobInProgress对象。这个是个非常重要的对象，它维护了作业的生命周期，可以跟踪作业的运行状态和进度。

    // Create the JobInProgress, do not lock the JobTracker since
    // we are about to copy job.xml from HDFS
    JobInProgress job = null;
    try {
      job = new JobInProgress(this, this.conf, jobInfo, 0, ts);
    } catch (Exception e) {
      throw new IOException(e);
    }

我们可以进入构造函数内部看看JobInProgress都有什么内容。

this.jobtracker = jobtracker;
this.status = new JobStatus(jobId, 0.0f, 0.0f, JobStatus.PREP);

JobStatus对象也比较关键，它维护了作业的一些状态信息如作业优先级，开始时间，map和reduce任务的进度等。
其他信息在后面分析JobTracker实现的时候再详述。

b. 检查用户的作业提交权限。Hadoop以队列为单位管理作业和资源。一个用户可以属于一个或多个队列，管理员可以配置用户在某个队列中的作业提交权限。

      // check if queue is RUNNING
      String queue = job.getProfile().getQueueName();
      if (!queueManager.isRunning(queue)) {
        throw new IOException("Queue "" + queue + "" is not running");
      }
      try {
        aclsManager.checkAccess(job, ugi, Operation.SUBMIT_JOB);
      } catch (IOException ioe) {
        LOG.warn("Access denied for user " + job.getJobConf().getUser()
            + ". Ignoring job " + jobId, ioe);
        job.fail();
        throw ioe;
      }

c. 检查作业的内存使用量。用户在配置文件中可以配置Map任务和Reduce任务的内存使用量，而这些值不能超过管理员所配置的最大使用量，否则作业提交就会失败。

      // Check the job if it cannot run in the cluster because of invalid memory
      // requirements.
      try {
        checkMemoryRequirements(job);
      } catch (IOException ioe) {
        throw ioe;
      }

d. 调用调度器模块，对作业进行初始化。具体过程如下

     // Submit the job
      JobStatus status;
      status = addJob(jobId, job);

在addJob方法中，作业首先被加入到已经提交的作业列表中，然后通知JobTracker所有的监听器对象，当前作业被提交，并采取相应的行动。

    synchronized (jobs) {
      synchronized (taskScheduler) {
        jobs.put(job.getProfile().getJobID(), job);
        for (JobInProgressListener listener : jobInProgressListeners) {
          listener.jobAdded(job);
        }
      }
    }

其中，任务调度器taskScheduler对象是在JobTracker构造时创建出来的。调度器对象和JobTracker对象是互相包含的关系。

    // Create the scheduler
    Class<? extends TaskScheduler> schedulerClass
      = conf.getClass("mapred.jobtracker.taskScheduler",
          JobQueueTaskScheduler.class, TaskScheduler.class);
    taskScheduler = (TaskScheduler) ReflectionUtils.newInstance(schedulerClass, conf);

可以看出，Hadoop任务调度器时可插拔的模块，调度器的类型通过配置文件获得，并使用反射机制实例化。用户可以通过继承TaskScheduler类实现自己的调度器（由于做研究需要，本人实现了一个简单的调度器，实现过程日后分享）。Hadoop默认的调度器为JobQueueTaskScheduler，调度策略为先进先出（FIFO）。

注意：JobTracker采用了典型的观察者模式。TaskScheduler为订阅者，JobTracker为发布者，二者通过作业监听器JobInProgressListener对象发生关系。当用户自定义了一个调度器后，同时要自定义监听器类。一个调度器对象可以包含若干个监听器，调度器在初始化时，会将其所有的监听器对象注册到JobTracker，订阅其发布的消息。以JobQueueTaskScheduler为例：

  public synchronized void start() throws IOException {
    super.start();
    taskTrackerManager.addJobInProgressListener(jobQueueJobInProgressListener);
    eagerTaskInitializationListener.setTaskTrackerManager(taskTrackerManager);
    eagerTaskInitializationListener.start();
    taskTrackerManager.addJobInProgressListener(
        eagerTaskInitializationListener);
  }

其中taskTrackerManager对象就是JobTracker，通过addJobInProgressListener方法注册监听器。这样，当有JobTracker发现有作业被提交、更新或删除时，就会通知订阅者TaskScheduler，并调用相应的回调函数，如上面提到的listener.jobAdded方法。更多调度器的细节请关注后续文章。

作业初始化的工作就是由上述的EagerTaskInitializationListener监听器对象实现的。在该监听器内部有一个作业初始化管理线程在运行，该线程访问一个初始化作业队列，取出一个作业，并新开一个作业初始化线程执行JobTracker.initJob方法。

初始化的过程主要是根据作业信息创建该作业的任务（Task）。作业的任务包括四种：

1. Setup Task。该任务进行一些简单工作，运行状态设置为setup。它在运行时会占用slot。Map和Reduce Setup任务各有一个。

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

2. Map Task。Map阶段处理数据的任务。其创建过程如下：

    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);
    }

TaskInProgrees维护任务运行时信息，与JobInProgress类似。

3. Reduce Task。Reduce阶段处理数据的任务。其创建过程如下：

    this.reduces = new TaskInProgress[numReduceTasks];
    for (int i = 0; i < numReduceTasks; i++) {
      reduces[i] = new TaskInProgress(jobId, jobFile, 
                                      numMapTasks, i, 
                                      jobtracker, conf, this, numSlotsPerReduce);
      nonRunningReduces.add(reduces[i]);
    }

用户可以在配置文件中指定Reduce任务的个数。

4. Cleanup Task。 作业完成后完成一些清理工作的任务。清理包括删除临时目录，设置状态等操作。

    // 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();

至此，作业初始化工作完成。接下来，调度器会根据当前可用的slot资源，从队列中选择一个作业，进而选择该作业的一个任务，放到空闲slot上执行。四种任务执行的顺序为Setup、Map、Reduce和Cleanup。由于Reduce任务的输入依赖于Map任务的输出，因此Reduce任务通常延后开始，否则将闲置reduce slot。可以配置档Map任务进度大于mapred.reduce.slowstart.completed.maps时，Reduce任务才开始，该值默认为5%。Map和Reduce任务这种依赖性在任务调度器设计中时常考虑。例如Facebook在提出FairScheduler的论文中就试图解决这个问题。

另外，关于为什么JobTracker将初始化工作交给调度器处理，文献【1】给出的理由是：

• 作业初始化后会占用内存资源，如果有大量初始化作业在JobTracker等待调度就会占用不必要的资源。在交给调度器后，Hadoop按照一定策略选择性地初始化以节省内存资源。
• 只有经过初始化的作业才能得到调度，因此将初始化工作嵌入调度器中比较合理。

有关任务调度器内容详见下一篇文章： MapReduce调度与执行原理之任务调度