JobTracker作业调度分析

转自：http://blog.csdn.net/Androidlushangderen/article/details/41408517

JobTracker的作业调度给我感觉就是比较宏观意义上的操作。倘若你只了解了MapReduce的工作原理是远远不够的，这时去学习一下他在宏观层面的原理实现也是对我们非常有帮助的。首先我们又得从上次分析的任务提交之后的操作说起，Job作业通过RPC通信提交到JobTracker端之后，接下来会触发到下面的方法；

/**
   * 初始化作业操作
   */
  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());
      //初始化Task任务
      job.initTasks();
      ......

接着会执行initTasks的方法，但不是JobTracker，而是JobInProgress类中的方法：

/**
  * Construct the splits, etc.  This is invoked from an async
  * thread so that split-computation doesn't block anyone.
  */
 public synchronized void initTasks()
 throws IOException, KillInterruptedException, UnknownHostException {
   if (tasksInited || isComplete()) {
     return;
   }
   ......

   jobtracker.getInstrumentation().addWaitingMaps(getJobID(), numMapTasks);
   jobtracker.getInstrumentation().addWaitingReduces(getJobID(), numReduceTasks);
   this.queueMetrics.addWaitingMaps(getJobID(), numMapTasks);
   this.queueMetrics.addWaitingReduces(getJobID(), numReduceTasks);

   //根据numMapTasks任务数，创建MapTask的总数
   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);
   }
   ......

   //
   // Create reduce tasks
   //根据numReduceTasks，创建Reduce的Task数量
   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]);
   }

   ......

   // create cleanup two cleanup tips, one map and one reduce.
   //创建2个clean up Task任务，1个是Map Clean-Up Task,一个是Reduce Clean-Up Task
   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();

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

   ......

可以看见，在这里JobInProgress首次被划分为了很多的小的Task任务的形式存在，而这些小的任务是以TaskInProgress的类表示。在这里MapReduce把1个作业做出了如下的分解，numMapTasks个Map Task ，numReduceTasks个Reduce Task，2个CleanUp任务，2个SetUp任务，（Map Reduce，每个各占1个），好，可以大致勾画一下，1个JobInProgress的执行流程了。

     ok，initTask的任务已经完成，也就是说前面初始化的准备工作都已经完成了，后面就等着JobTacker分配作业给TaskTracker了。在这里MapReduce用的是HeartBeat的形式，就是心跳机制，心跳包在这里主要有3个作用：

1.判断TaskTracker是否活着

2.获取各个TaskTracker上的资源使用情况和任务的进度

3.给TaskTracker分配任务

而这里用到的就是第三作用。HeartBeat的调用形式同样是Hadoop自带的RPC实现方式。JobTracker不会直接分配作业给TaskTracker，中间会经过一个叫TaskScheduler掉调度器，这个可以用户自定义实现，满足不同的需求设计，在Hadoop中有默认的实现，所以你会看到大致这样的一个模型流程：

        所以接下来JobTracker首先会收到很多来自TaskTracker的心跳包，判断此TaskTracker是否是无任务状态的，无任务的话，马上让TaskSchedulera分配任务给他:

public synchronized HeartbeatResponse heartbeat(TaskTrackerStatus status,
                                                  boolean restarted,
                                                  boolean initialContact,
                                                  boolean acceptNewTasks,
                                                  short responseId)
    throws IOException {
   ....

    //通过心跳机制发送命令回应
    // Initialize the response to be sent for the heartbeat
    HeartbeatResponse response = new HeartbeatResponse(newResponseId, null);
    List<TaskTrackerAction> actions = new ArrayList<TaskTrackerAction>();
    boolean isBlacklisted = faultyTrackers.isBlacklisted(status.getHost());
    // Check for new tasks to be executed on the tasktracker
    if (recoveryManager.shouldSchedule() && acceptNewTasks && !isBlacklisted) {
      TaskTrackerStatus taskTrackerStatus = getTaskTrackerStatus(trackerName);
      if (taskTrackerStatus == null) {
        LOG.warn("Unknown task tracker polling; ignoring: " + trackerName);
      } else {
        List<Task> tasks = getSetupAndCleanupTasks(taskTrackerStatus);
        //说明此TaskTtracker上无任务了
        if (tasks == null ) {
          //为此TaskTracker分配任务
          tasks = taskScheduler.assignTasks(taskTrackers.get(trackerName));
        }

接下来就是TaskScheduler的方法了，不过得找出他的实现类，TaskScheduler只是一个基类：

public synchronized List<Task> assignTasks(TaskTracker taskTracker)
    throws IOException {
  TaskTrackerStatus taskTrackerStatus = taskTracker.getStatus();
  ClusterStatus clusterStatus = taskTrackerManager.getClusterStatus();
  final int numTaskTrackers = clusterStatus.getTaskTrackers();
  final int clusterMapCapacity = clusterStatus.getMaxMapTasks();
  final int clusterReduceCapacity = clusterStatus.getMaxReduceTasks();

  //获取作业队列
  Collection<JobInProgress> jobQueue =
    jobQueueJobInProgressListener.getJobQueue();
   .....
      for (JobInProgress job : jobQueue) {
        if (job.getStatus().getRunState() != JobStatus.RUNNING ||
            job.numReduceTasks == 0) {
          continue;
        }


        //在这里分配了一个新的Reduce任务
        Task t =
          job.obtainNewReduceTask(taskTrackerStatus, numTaskTrackers,
                                  taskTrackerManager.getNumberOfUniqueHosts()
                                  );
        .....

首先获取一个作业列表，在里面挑出一个作业给，在比如从里面挑出1个Reduce的任务区给整个TaskTracker执行，因为我们刚刚已经知道，所有的Task都是以TaskInProgress形式被包含于JobInProgress中的，所以又来到了JobInProgress中了

/**
   * Return a ReduceTask, if appropriate, to run on the given tasktracker.
   * We don't have cache-sensitivity for reduce tasks, as they
   *  work on temporary MapRed files.
   */
  public synchronized Task obtainNewReduceTask(TaskTrackerStatus tts,
                                               int clusterSize,
                                               int numUniqueHosts
                                              ) throws IOException {
    .....

    int  target = findNewReduceTask(tts, clusterSize, numUniqueHosts,
                                    status.reduceProgress());
    if (target == -1) {
      return null;
    }

    //这里继续调用方法，获取目标任务
    Task result = reduces[target].getTaskToRun(tts.getTrackerName());
    if (result != null) {
      addRunningTaskToTIP(reduces[target], result.getTaskID(), tts, true);
    }

    return result;
  }

此时就执行了一个TIP就是TaskInProgress里面去执行了，此时的转变就是JIP->TIP的转变。继续往里看,这时候来到的是TaskInProgress的类里面了:

public Task getTaskToRun(String taskTracker) throws IOException {
    if (0 == execStartTime){
      // assume task starts running now
      execStartTime = jobtracker.getClock().getTime();
    }

    // Create the 'taskid'; do not count the 'killed' tasks against the job!
    TaskAttemptID taskid = null;
    if (nextTaskId < (MAX_TASK_EXECS + maxTaskAttempts + numKilledTasks)) {
      // Make sure that the attempts are unqiue across restarts
      int attemptId = job.getNumRestarts() * NUM_ATTEMPTS_PER_RESTART + nextTaskId;
      //启动一次TA尝试
      taskid = new TaskAttemptID( id, attemptId);
      ++nextTaskId;
    } else {
      LOG.warn("Exceeded limit of " + (MAX_TASK_EXECS + maxTaskAttempts) +
              " (plus " + numKilledTasks + " killed)"  +
              " attempts for the tip '" + getTIPId() + "'");
      return null;
    }

    //加入到相应的数据结构中
    return addRunningTask(taskid, taskTracker);
  }

在这里明显的执行了所谓的TA尝试，就是说这是一次Task的尝试执行，因为不能保证这次任务就一定能执行成功。把这次尝试的任务ID加入系统变量中，就来到了addRunningTask，也就是说来到了方法执行的最末尾:

/**
   * Adds a previously running task to this tip. This is used in case of
   * jobtracker restarts.
   * 添加任务
   */
  public Task addRunningTask(TaskAttemptID taskid,
                             String taskTracker,
                             boolean taskCleanup) {
    .....
    //添加任务和taskTracker的映射关系
    activeTasks.put(taskid, taskTracker);
    tasks.add(taskid);

    // Ask JobTracker to note that the task exists
    //在JobTracker中增加一对任务记录
    jobtracker.createTaskEntry(taskid, taskTracker, this);

    // check and set the first attempt
    if (firstTaskId == null) {
      firstTaskId = taskid;
    }
    return t;
  }

在这里，就增加了任务和TaskTracker的一些任务运行信息的变量关系。后面就等着TaskTracker自己去把任务挑出来，执行就OK了，上面这个步骤从TIP->TA的转变。我们把这种结构流程叫做“三层多叉树”的方式结构。

整个作业的调度的时序关系图如下：