hadoop运行原理之Job运行(四) JobTracker端心跳机制分析

 1 public synchronized HeartbeatResponse heartbeat(TaskTrackerStatus status, 
 2                                                   boolean restarted,
 3                                                   boolean initialContact,
 4                                                   boolean acceptNewTasks, 
 5                                                   short responseId)  
 6 throws IOException {
 7      if (LOG.isDebugEnabled()) {
 8        LOG.debug("Got heartbeat from: " + status.getTrackerName() +  " (restarted: " + restarted + 
 9                  " initialContact: " + initialContact + 
10                  " acceptNewTasks: " + acceptNewTasks + ")" +
11                  " with responseId: " + responseId);
12      }
13      // 检查该主机是否允许与JobTracker通信
14     if (!acceptTaskTracker(status)) {
15       throw new DisallowedTaskTrackerException(status);
16     }

1  String trackerName = status.getTrackerName();
2     long now = clock.getTime();
3     if (restarted) {
4       faultyTrackers.markTrackerHealthy(status.getHost());
5     } else {
6       faultyTrackers.checkTrackerFaultTimeout(status.getHost(), now);
7     }

 1 HeartbeatResponse prevHeartbeatResponse =
 2       trackerToHeartbeatResponseMap.get(trackerName);  //获取上次心跳应答
 3     boolean addRestartInfo = false;  //表示JobTracker是否重启
 4 
 5     if (initialContact != true) { //该TaskTracker不是初次连接JobTracker
 6       // If this isn‘t the ‘initial contact‘ from the tasktracker,
 7       // there is something seriously wrong if the JobTracker has
 8       // no record of the ‘previous heartbeat‘; if so, ask the 
 9       // tasktracker to re-initialize itself.
10       if (prevHeartbeatResponse == null) {　//TaskTracker不是初次连接JobTracker，并且上一次心跳应答对象为空
11         // This is the first heartbeat from the old tracker to the newly 
12         // started JobTracker
13         if (hasRestarted()) {　　//JobTracker重启了
14           addRestartInfo = true;
15           // inform the recovery manager about this tracker joining back
16           recoveryManager.unMarkTracker(trackerName);
17         } else {  //JobTracker没有重启
18           // Jobtracker might have restarted but no recovery is needed
19           // otherwise this code should not be reached
20           LOG.warn("Serious problem, cannot find record of ‘previous‘ " +
21                    "heartbeat for ‘" + trackerName + 
22                    "‘; reinitializing the tasktracker");
23           return new HeartbeatResponse(responseId, 
24               new TaskTrackerAction[] {new ReinitTrackerAction()}); //重新初始化该TaskTracker
25         }
26 
27       } else { //该TaskTracker不是第一次连接JobTracker，并且上次心跳应答不为空
28                 
29         // It is completely safe to not process a ‘duplicate‘ heartbeat from a 
30         // {@link TaskTracker} since it resends the heartbeat when rpcs are 
31         // lost see {@link TaskTracker.transmitHeartbeat()};
32         // acknowledge it by re-sending the previous response to let the 
33         // {@link TaskTracker} go forward. 
34         if (prevHeartbeatResponse.getResponseId() != responseId) {
35           LOG.info("Ignoring ‘duplicate‘ heartbeat from ‘" + 
36               trackerName + "‘; resending the previous ‘lost‘ response");
37           return prevHeartbeatResponse;  //心跳丢失，返回上次心跳应答
38         }
39       }
40     }

 1    // Process this heartbeat 
 2     short newResponseId = (short)(responseId + 1);
 3     status.setLastSeen(now);
 4     if (!processHeartbeat(status, initialContact, now)) {　//心跳处理失败
 5       if (prevHeartbeatResponse != null) {  //上次心跳应答不存在
 6         trackerToHeartbeatResponseMap.remove(trackerName);
 7       }
 8       return new HeartbeatResponse(newResponseId, 
 9                    new TaskTrackerAction[] {new ReinitTrackerAction()});  //重新初始化该TaskTracker
10     }

 1  // Initialize the response to be sent for the heartbeat
 2     HeartbeatResponse response = new HeartbeatResponse(newResponseId, null);
 3     List<TaskTrackerAction> actions = new ArrayList<TaskTrackerAction>();
 4     boolean isBlacklisted = faultyTrackers.isBlacklisted(status.getHost());  //检查该结点是否在黑名单中
 5     // Check for new tasks to be executed on the tasktracker
 6     if (recoveryManager.shouldSchedule() && acceptNewTasks && !isBlacklisted) {
 7       TaskTrackerStatus taskTrackerStatus = getTaskTrackerStatus(trackerName);
 8       if (taskTrackerStatus == null) {
 9         LOG.warn("Unknown task tracker polling; ignoring: " + trackerName);
10       } else {  //首先执行cleanup和setup task
11         List<Task> tasks = getSetupAndCleanupTasks(taskTrackerStatus);
12         if (tasks == null ) {  //如果没有cleanup和setup task，则由作业调度器分配map和reduce task
13           tasks = taskScheduler.assignTasks(taskTrackers.get(trackerName));
14         }
15         if (tasks != null) {
16           for (Task task : tasks) {
17             expireLaunchingTasks.addNewTask(task.getTaskID());
18             if(LOG.isDebugEnabled()) {
19               LOG.debug(trackerName + " -> LaunchTask: " + task.getTaskID());
20             }
21             actions.add(new LaunchTaskAction(task));
22           }
23         }
24       }
25     }

 1    // Check for tasks to be killed
 2     List<TaskTrackerAction> killTasksList = getTasksToKill(trackerName);
 3     if (killTasksList != null) {
 4       actions.addAll(killTasksList);
 5     }
 6      
 7     // Check for jobs to be killed/cleanedup
 8     List<TaskTrackerAction> killJobsList = getJobsForCleanup(trackerName);
 9     if (killJobsList != null) {
10       actions.addAll(killJobsList);
11     }
12 
13     // Check for tasks whose outputs can be saved
14     List<TaskTrackerAction> commitTasksList = getTasksToSave(status);
15     if (commitTasksList != null) {
16       actions.addAll(commitTasksList);
17     }

 1    // calculate next heartbeat interval and put in heartbeat response
 2     int nextInterval = getNextHeartbeatInterval();  //计算下次心跳时间间隔
 3     response.setHeartbeatInterval(nextInterval);
 4     response.setActions(  //将下达给TaskTracker的指令封装到心跳应答中
 5                         actions.toArray(new TaskTrackerAction[actions.size()]));
 6     
 7     // check if the restart info is req
 8     if (addRestartInfo) {  //将需要恢复的Job告知放入心跳应答中
 9       response.setRecoveredJobs(recoveryManager.getJobsToRecover());
10     }
11         
12     // Update the trackerToHeartbeatResponseMap
13     trackerToHeartbeatResponseMap.put(trackerName, response);  //将本次心跳应答更新到trackerToHeartbeatResponseMap，作为该TaskTracker最新的心跳应答
14 
15     // Done processing the hearbeat, now remove ‘marked‘ tasks
16     removeMarkedTasks(trackerName);  //从trackerToMarkedTasksMap移除已被标记为完成的task
17         
18     return response;

 1   /**
 2    * Calculates next heartbeat interval using cluster size.
 3    * Heartbeat interval is incremented by 1 second for every 100 nodes by default. 
 4    * @return next heartbeat interval.
 5    */
 6   public int getNextHeartbeatInterval() {  //默认情况下，处理每个RPC大约要10ms
 7     // get the no of task trackers
 8     int clusterSize = getClusterStatus().getTaskTrackers();  //
 9     int heartbeatInterval =  Math.max(
10                                 (int)(1000 * HEARTBEATS_SCALING_FACTOR *
11                                       ((double)clusterSize / 
12                                                 NUM_HEARTBEATS_IN_SECOND)),
13                                 HEARTBEAT_INTERVAL_MIN) ;
14     return heartbeatInterval;
15   }