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

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

一、作业提交与初始化

在
 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调度与执行原理之任务调度