Job的任务执行流程之Map阶段

旧城等待， 2022-06-13 05:55 228阅读 0赞

一个作业的JobSetup任务被一个TaskTracker节点成功执行并报告到JobTracker节点之后，JobTracker节点就会更新对应的作业的状态为RUNNING，之后，这个作业的真正Map任务就可以被JobTracker调度分配给合适的TaskTracker节点来执行了。虽然本文的重点是讲述作业的Map任务在Hadoop集群中的执行过程，但JobTracker节点调度一个作业的Map任务的策略关系到作业的执行效率，所以我也将根据目前Hadoop的默认任务调度器来详细的讨论这个问题。

我们知道，每个TaskTracker节点会通过心跳包向JobTracker节点报告它当前的状态信息，其中这个状态信息就包括该TaskTracker节点当前可同时运行Map任务的最大数量以及正在该节点上运行的Map任务数量。那么JobTracker节点就可以根据这些信息来计算整个集群的当前负载情况和最大承载能力。当一个TaskTracker节点向JobTracker节点发送心跳包之后，JobTracker节点就会给该TaskTracker节点分配任务当做是对心跳包的响应。关于JobTracker节点给一个TaskTracker节点分配Map任务，是有一定的策略的。JobTracker节点会首先计算当前整个Hadoop集群的平均负载情况来评估这个TaskTracker节点是否超载了，如果超载了，就不会给他分配Map任务，否在，就计算应该给它分配几个Map任务已达到集群的平均负载。同时，JobTracker节点还考虑到了一些意外情况和特殊的任务，也就是说JobTracker节点为Hadoop集群预留了一定的计算能力，以便应该突发情况。当整个集群预留的map slot不足时，JobTracker节点是如何处理的呢？比如说一个TaskTracker节点根据当前的集群负载情况因该分配n个Map任务，那么首先，默认的任务调度器总是优先调度一个作业的本地Map任务给当前的TaskTracker节点，没有本地Map任务，就分配一个非本地Map任务；如果预留的map slot不足时，默认任务调度器在分配了一个本地Map任务之后，还会分配该作业的一个非本地Map任务给当前的TaskTracker节点；最后，这个TaskTracker最多有可能分配了2\*n个Map任务。

TaskTracker节点成功接收到JobTracker节点分配的Map任务之后，此Map任务实例状态在TaskTracker节点上仍是UNASSIGNED。当有TaskTracker节点上有了空闲的map slot时，它就会调度这些Map任务，不过在把这些Map任务交给空闲的JVM运行之前，这些Map任务还需本地化，如果本地化失败，这个Map任务实例的状态就会变成FAILED，同时将与这个Map任务实例相关的中间文件及目录被交给TaskTracker节点的一个后台线程来清理；如果本地化成功，它的状态就转变为RUNNING，同时将其更新到JobTracker节点上。在Map任务实例交给一个JVM运行之后，它会向TaskTracker节点报告它的状态和进度，也就是它会每隔3000 ms检查一次该任务的进度是否发生了变化，如果发生了变化就向TaskTracker节点报告，如果没有就发送一个ping包来询问这个Map任务实例是否还需要继续执行。Map任务如果执行失败了，它所在的JVM实例会向TaskTracker节点报告相应的错误原因，同时将该Map任务实例的Phase设置为CLEANUP，并将此时的状态和进度报告给TaskTracker节点，然后执行该作业对应的作业输出提交器OutputCommitter的abortTask()；Map任务如果执行成功，它就会调用OutputCommitter的needsTaskCommit()方法来判断此时是否可以向TaskTracker节点提交该任务，若可以提交，则该MapTask实例状态变为COMMIT\_PENDING，并将此状态随提交请求一起发送给TaskTracker节点，TaskTracker节点在收到这个提交请求之后，会更新这个MapTask实例在TaskTracker上的状态为COMMIT\_PENDING(当前的FileOutputCommitter中，Map任务不会处于该状态)。之后这个JVM实例会向TaskTracker节点请求新的Task实例来执行。其中，在这个MapTask执行的过程中，可能发生一些意外情况，如这个任务或者这个TaskTracker节点被JobTracker节点强制终止，则该MapTask状态变为KILLED\_UNCLEAN；如果这个任务在JVM中执行出现错误，则它的状态变为FAILED\_UNCLEAN。

![1332768308_3390.jpg][]

最后，JobTracker节点会不断地收到TaskTracker节点对该任务执行的进度和状态报告，并根据这个信息来更新对应的TaskInProgress和JobInProgress等信息，源代码如下：

public synchronized void updateTaskStatus(TaskInProgress tip, TaskStatus status) {
    
        double oldProgress = tip.getProgress();   // save old progress
        boolean wasRunning = tip.isRunning();
        boolean wasComplete = tip.isComplete();
        boolean wasPending = tip.isOnlyCommitPending();
        TaskAttemptID taskid = status.getTaskID();
        
        // 如果任务实例对应的任务已经完成或被kill，同时该任务实例被成功执行，则应将其看做已经被killed
        if ((wasComplete || tip.wasKilled(taskid)) && (status.getRunState() == TaskStatus.State.SUCCEEDED)) {
          status.setRunState(TaskStatus.State.KILLED);
        }
        
        // If the job is complete and a task has just reported its 
        // state as FAILED_UNCLEAN/KILLED_UNCLEAN, 
        // make the task's state FAILED/KILLED without launching cleanup attempt.
        // Note that if task is already a cleanup attempt, 
        // we don't change the state to make sure the task gets a killTaskAction
        if ((this.isComplete() || jobFailed || jobKilled) && !tip.isCleanupAttempt(taskid)) {
          if (status.getRunState() == TaskStatus.State.FAILED_UNCLEAN) {
            status.setRunState(TaskStatus.State.FAILED);
          } else if (status.getRunState() == TaskStatus.State.KILLED_UNCLEAN) {
            status.setRunState(TaskStatus.State.KILLED);
          }
        }
        
        //更新任务的状态
        boolean change = tip.updateStatus(status);
        if (change) {
          TaskStatus.State state = status.getRunState();
          // get the TaskTrackerStatus where the task ran 
          TaskTrackerStatus ttStatus = this.jobtracker.getTaskTracker(tip.machineWhereTaskRan(taskid));
          String httpTaskLogLocation = null; 
    
          if (null != ttStatus){
            String host;
            if (NetUtils.getStaticResolution(ttStatus.getHost()) != null) {
              host = NetUtils.getStaticResolution(ttStatus.getHost());
            } else {
              host = ttStatus.getHost();
            }
            httpTaskLogLocation = "http://" + host + ":" + ttStatus.getHttpPort(); 
               //+ "/tasklog?plaintext=true&taskid=" + status.getTaskID();
          }
    
          TaskCompletionEvent taskEvent = null;
          if (state == TaskStatus.State.SUCCEEDED) {
            taskEvent = new TaskCompletionEvent(taskCompletionEventTracker, taskid, tip.idWithinJob(), status.getIsMap() && !tip.isJobCleanupTask() && !tip.isJobSetupTask(), TaskCompletionEvent.Status.SUCCEEDED, httpTaskLogLocation);
            taskEvent.setTaskRunTime((int)(status.getFinishTime() - status.getStartTime()));
            tip.setSuccessEventNumber(taskCompletionEventTracker); 
          } 
          else if (state == TaskStatus.State.COMMIT_PENDING) {
            // If it is the first attempt reporting COMMIT_PENDING
            // ask the task to commit.
            if (!wasComplete && !wasPending) {
              tip.doCommit(taskid);
            }
            return;
          } 
          else if (state == TaskStatus.State.FAILED_UNCLEAN || state == TaskStatus.State.KILLED_UNCLEAN) {
            tip.incompleteSubTask(taskid, this.status);
            // add this task, to be rescheduled as cleanup attempt
            if (tip.isMapTask()) {
              LOG.debug("add MapTask["+taskid+"] for cleaning up.");
              mapCleanupTasks.add(taskid);
            } else {
            	LOG.debug("add ReduceTask["+taskid+"] for cleaning up.");
              reduceCleanupTasks.add(taskid);
            }
            // Remove the task entry from jobtracker
            jobtracker.removeTaskEntry(taskid);
          }
          //For a failed task update the JT datastructures. 
          else if (state == TaskStatus.State.FAILED || state == TaskStatus.State.KILLED) {
            // Get the event number for the (possibly) previously successful
            // task. If there exists one, then set that status to OBSOLETE 
            int eventNumber;
            if ((eventNumber = tip.getSuccessEventNumber()) != -1) {
              TaskCompletionEvent t = this.taskCompletionEvents.get(eventNumber);
              if (t.getTaskAttemptId().equals(taskid))
                t.setTaskStatus(TaskCompletionEvent.Status.OBSOLETE);
            }
            
            // Tell the job to fail the relevant task
            failedTask(tip, taskid, status, ttStatus, wasRunning, wasComplete);
    
            // Did the task failure lead to tip failure?
            TaskCompletionEvent.Status taskCompletionStatus = (state == TaskStatus.State.FAILED ) ? TaskCompletionEvent.Status.FAILED : TaskCompletionEvent.Status.KILLED;
            if (tip.isFailed()) {
              taskCompletionStatus = TaskCompletionEvent.Status.TIPFAILED;
            }
            taskEvent = new TaskCompletionEvent(taskCompletionEventTracker, taskid, tip.idWithinJob(), status.getIsMap() && !tip.isJobCleanupTask() && !tip.isJobSetupTask(), taskCompletionStatus, httpTaskLogLocation);
          }          
    
          // Add the 'complete' task i.e. successful/failed
          // It _is_ safe to add the TaskCompletionEvent.Status.SUCCEEDED
          // *before* calling TIP.completedTask since:
          // a. One and only one task of a TIP is declared as a SUCCESS, the
          //    other (speculative tasks) are marked KILLED by the TaskCommitThread
          // b. TIP.completedTask *does not* throw _any_ exception at all.
          if (taskEvent != null) {
            this.taskCompletionEvents.add(taskEvent);
            taskCompletionEventTracker++;
            if (state == TaskStatus.State.SUCCEEDED) {
              completedTask(tip, status);
            }
          }
          
        }
            
        //
        // Update JobInProgress status
        //
        if(LOG.isDebugEnabled()) {
          LOG.debug("Taking progress for " + tip.getTIPId() + " from " + oldProgress + " to " + tip.getProgress());
        }
        
        //更新作业的进度
        if (!tip.isJobCleanupTask() && !tip.isJobSetupTask()) {
          double progressDelta = tip.getProgress() - oldProgress;
          if (tip.isMapTask()) {
              this.status.setMapProgress((float) (this.status.mapProgress() + progressDelta / maps.length));
          } else {
            this.status.setReduceProgress((float) (this.status.reduceProgress() + (progressDelta / reduces.length)));
          }
        }
        
      }

关于一个任务可以同时由多个TaskTracker同时独立执行这一点前面已经不断的提到过，但是一个任务不可能同时结果任意多个TaskTracker节点来同时执行，也就是说这个上限是什么呢？其实，一个Map/Reduce任务最多可有**MAX\_TASK\_EXECS** \+ maxTaskAttempts个实例被TaskTracker节点同时运行，其中**MAX\_TASK\_EXECS**的值为1，maxTaskAttempts的值由该任务所属的作业的配置来决定，Map/Reduce任务对应的配置分别为：mapred.map.max.attempts、mapred.reduce.max.attempts。这个最大值也可以看做是一个任务能允许尝试执行的最大次数，当这个任务在尝试执行的次数达到这个阈值的时候，它还没有被执行成功，那么这个任务就不会再交给其它的TaskTracker节点来执行了，因为默认该任务已不可能被成功执行完。若一个任务的某一实例确定被一个TaskTracker节点执行失败，则消耗了一次尝试次数；若该任务的某一实例确定由于某种原因被kill掉，则不消耗一次尝试次数。如果一个任务在完成之前，它失败的实力数量就已经达到maxTaskAttempts的话，则认为这个任务已经失败了，即不可能完成。另外还有一种情况就是，对于一个特殊的作业而言，它的一个正在被某TaskTracker节点执行的Map/Reduce任务实例还可以同时被分配给其它的TaskTracker节点来执行，对应的配置项分配为：mapred.map.tasks.speculative.execution、mapred.reduce.tasks.speculative.execution，对应值类型为true/false。

一个作业可能能够容忍少数几个Map任务的失败，但是当这个作业的Map任务失败的数量超过一定的阈值之后，这个作业再执行下去就没有任何意义了。这个阈值可以通过作业的配置文件来设置，对应的配置项为：mapred.max.map.failures.percent，它的值的范围为：\[0,100\]，表示说如果该作业失败的Map任务占该作业总Map任务的百分比超过这个阈值时，就认为该作业执行失败了，也就不需要再继续执行了。另外，为了提高作业的执行效率，作业的Reduce任务并不会等到所有的Map任务执行完才开始执行，而当这个作业的Map任务成功完成了多少个之后就可以开始了，这个阈值是completedMapsForReduceSlowstart：

private static float DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART = 0.05f;
    completedMapsForReduceSlowstart = (int)Math.ceil((conf.getFloat("mapred.reduce.slowstart.completed.maps", DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART) * numMapTasks));

在作业的Map任务容易失败的Hadoop集群环境中，Reduce任务开始的阈值一般应该大于能容忍Map任务失败的阈值。这里还要讨论的一个问题就是Map任务在一个JVM实例中执行的时候，它必须至少每隔taskTimeout ms向TaskTracker节点报告一次状态和进度信息，如果TaskTracker节点在taskTimeout ms内没有收到这个Map任务实例的报告，就会抛弃该任务实例并默认它已经失败了。这个taskTimeout默认值是10\*60\*1000，不过也可以通过作业的配置文件来配置，对应的配置项为：mapred.task.timeout，当它配置为0是则表示这个taskTimeout值为无限大。任何任务在JVM中执行的时候会出现三类异常错误：

1.org.apache.hadoop.fs.FSError 当出现这一类错误时，JVM实例会马上通知TaskTracker节点并告知它这个任务实例已经执行失败了，而TaskTracker节点也会马上对该任务实例作出错处理，而JVM实例此时也会停止；

2.java.lang.Throwable  当出现这一类异常时，它的处理同FSError ；

3.java.lang.Exception当出现这一类异常时，JVM实例会马上通知TaskTracker节点并告知它出现这个异常的原因，之后JVM实例就停止，但TaskTracker节点不会终止该任务实例，而是捕捉到该JVM停止之后才认为该任务实例执行失败了，然后作出错处理。

[1332768308_3390.jpg]: https://img-my.csdn.net/uploads/201203/26/1332768308_3390.jpg