码迷,mamicode.com
首页 > 其他好文 > 详细

TaskTracker启动过程源码级分析

时间:2014-06-05 17:21:13      阅读:288      评论:0      收藏:0      [点我收藏+]

标签:c   style   class   blog   code   java   

  TaskTracker也是作为一个单独的JVM来运行的,其main函数就是TaskTracker的入口函数,当运行start-all.sh时,脚本就是通过SSH运行该函数来启动TaskTracker的。

  TaskTracker是JobTracker和Task之间的桥梁:一方面,从JobTracker接收并执行各种命令:运行任务、提交任务、杀死任务等;另一方面,将本地节点上各个任务的状态通过心跳周期性汇报给JobTracker。TaskTracker与JobTracker和Task之间采用了RPC协议进行通信。

  TaskTracker的功能:1.汇报心跳:Tracker周期性将所有节点上各种信息通过心跳机制汇报给JobTracker。这些信息包括两部分:机器级别信息:节点健康情况、资源使用情况等;任务级别信息:任务执行进度、任务运行状态等。2.执行命令:JobTracker会给TaskTracker下达各种命令,主要包括:启动任务(LaunchTaskAction)、提交任务(CommitTaskAction)、杀死任务(KillTaskAction)、杀死作业(KillJobAction)和重新初始化(TaskTrackerReinitAction)。
  main()方法中主要工作就是初始化TaskTracker并启动它。
  一、初始化:TaskTracker tt = new TaskTracker(conf),TaskTracker(conf)获取本机的一些配置信息,初始化服务器并启动服务器(StatusHttpServer)。
  获取map和reduce最大的slot数量;获取jobTracker的地址jobTrackAddr;构造TaskTracker的HttpServer,可以从浏览器查看该tasktracker的运行情况及所有task;获取从HDFS下载文件的本地目录localStorage(即"mapred.local.dir");构造一个TaskController(是一个接口)默认是DefaultTaskController,用来控制任务的初始化、终结、清理tasks的,也可以启动和杀死tasks的JVM;调用initialize()方法,该方法是真正的构造函的地方,使用单独的方法可以再次调用;启动HttpServer。
  initialize()方法会清理一些历史文件(第一次启动TaskTracker不会有历史文件),并新建一些目录;初始化一些参数;新建一个JvmManager来管理着mapJvmManager和reduceJvmManager,这俩都是JvmManagerForType,JvmManager.launchJvm(TaskRunner t, JvmEnv env)方法会启动Task JVM运行Task,这个过程还比较复杂后续再说;构造taskReportServer,Task通过RPC向该Server汇报进度;构造一个distributedCacheManager分布式缓存管理并启动CleanupThread线程周期性的检查和清理不再使用的文件;然后实例化InterTrackerProtocol的jobClient,InterTrackerProtocol是用于TaskTracker和JobTracker通信的协议;启动一个MapEventsFetcherThread线程用于获取Map任务的输出数据信息,获取已运行完成的Map Task列表,为Reduce Task远程拷贝数据做准备;初始化内存,做任务内存监控;构造两个TaskLauncher线程:mapLauncher和reduceLauncher并启动,这个线程的run函数就是不断监测taskToLaunch队列中是否有新的 TaskTracker.TaskInProgress对象加入。如果有则从中取出一个对象,然后调用TaskTracker类的 startNewTask(TaskInProgress tip)来启动一个task。
  二、启动:tt.run()启动TaskTracker线程,维护一个和JobTracker通信的链接,其中run函数主要调用了offerService函数。JobTracker从不会主动与TaskTracker通信,总是被动等待TaskTracker汇报信息并领取其对应的命令。  
bubuko.com,布布扣
 1   public void run() {
 2     try {
 3       getUserLogManager().start();
 4       startCleanupThreads();
 5       boolean denied = false;
 6       while (running && !shuttingDown && !denied) {
 7         boolean staleState = false;
 8         try {
 9           // This while-loop attempts reconnects if we get network errors
10           while (running && !staleState && !shuttingDown && !denied) { //如果连接断开,则会循环尝试连接JobTracker
11             try {
12                 //如果连接JobTracker服务成功,TaskTracker就会调用offerService()函数进入主执行循环中。
13                 //这个循环会每隔一段时间与JobTracker通讯一次,调用transmitHeartBeat(),获得HeartbeatResponse信息。
14               State osState = offerService();    
15               if (osState == State.STALE) {
16                 staleState = true;
17               } else if (osState == State.DENIED) {
18                 denied = true;
19               }
20             } catch (Exception ex) {
21               if (!shuttingDown) {
22                 LOG.info("Lost connection to JobTracker [" +
23                          jobTrackAddr + "].  Retrying...", ex);
24                 try {
25                   Thread.sleep(5000);
26                 } catch (InterruptedException ie) {
27                 }
28               }
29             }
30           }
31         } finally {
32           close();
33         }
34         if (shuttingDown) { return; }
35         LOG.warn("Reinitializing local state");
36         initialize();    //初始化所有成员和参数
37       }
38       if (denied) {
39         shutdown();
40       }
41     } catch (IOException iex) {
42       LOG.error("Got fatal exception while reinitializing TaskTracker: " +
43                 StringUtils.stringifyException(iex));
44       return;
45     }
46     catch (InterruptedException i) {
47       LOG.error("Got interrupted while reinitializing TaskTracker: " +
48           i.getMessage());
49       return;
50     }
51   }
bubuko.com,布布扣

  getUserLogManager().start()会启动两个线程:用户日志清理线程和监控日志行为线程;会启动一个taskCleanupThread线程,这个线程会始终监视BlockingQueue<TaskTrackerAction> tasksToCleanup(用来存放要被清理的TaskInProgress),里面存储的是KillJobAction或者是KillTaskAction类型;接下来是两层的while循环,其中内层循环是一旦连接断开就循环尝试连接JobTracker,循环主体是State osState = offerService(),offerService()方法一旦执行如果没有其它情况将会一直执行,如果内层while循环退出则尝试重新initialize(),如果offerService()方法返回的状态是DENIED则退出所有循环并执行shutdown()。

   offerService()方法代码较多,如下:
bubuko.com,布布扣
  1 /**
  2    * Main service loop.  Will stay in this loop forever.
  3    */
  4   State offerService() throws Exception {
  5     long lastHeartbeat = System.currentTimeMillis();//上一次发心跳距现在时间
  6     //此循环主要根据控制完成task个数控制心跳间隔。//TaskTracker进行是一直存在的
  7     while (running && !shuttingDown) {
  8       try {
  9         long now = System.currentTimeMillis();//获得当前时间    
 10         
 11         // accelerate to account for multiple finished tasks up-front
 12       //通过完成的任务数动态控制心跳间隔时间  
 13         long remaining = 
 14           (lastHeartbeat + getHeartbeatInterval(finishedCount.get())) - now;
 15         while (remaining > 0) {
 16           // sleeps for the wait time or 
 17           // until there are *enough* empty slots to schedule tasks
 18           synchronized (finishedCount) {
 19             finishedCount.wait(remaining);
 20             
 21             // Recompute
 22             now = System.currentTimeMillis();
 23             remaining = 
 24               (lastHeartbeat + getHeartbeatInterval(finishedCount.get())) - now;
 25             
 26             if (remaining <= 0) {
 27               // Reset count 
 28               finishedCount.set(0);
 29               break;
 30             }
 31           }
 32         }
 33 
 34         // If the TaskTracker is just starting up:
 35         // 1. Verify the buildVersion
 36         // 2. Get the system directory & filesystem
 37         if(justInited) {
 38           String jobTrackerBV = jobClient.getBuildVersion();
 39           if(!VersionInfo.getBuildVersion().equals(jobTrackerBV)) {
 40             String msg = "Shutting down. Incompatible buildVersion." +
 41             "\nJobTracker‘s: " + jobTrackerBV + 
 42             "\nTaskTracker‘s: "+ VersionInfo.getBuildVersion();
 43             LOG.error(msg);
 44             try {
 45               jobClient.reportTaskTrackerError(taskTrackerName, null, msg);
 46             } catch(Exception e ) {
 47               LOG.info("Problem reporting to jobtracker: " + e);
 48             }
 49             return State.DENIED;
 50           }
 51           
 52           String dir = jobClient.getSystemDir();
 53           if (dir == null) {
 54             throw new IOException("Failed to get system directory");
 55           }
 56           systemDirectory = new Path(dir);
 57           systemFS = systemDirectory.getFileSystem(fConf);
 58         }
 59 
 60         now = System.currentTimeMillis();
 61         if (now > (lastCheckDirsTime + diskHealthCheckInterval)) {
 62           localStorage.checkDirs();
 63           lastCheckDirsTime = now;
 64           int numFailures = localStorage.numFailures();
 65           // Re-init the task tracker if there were any new failures
 66           if (numFailures > lastNumFailures) {
 67             lastNumFailures = numFailures;
 68             return State.STALE;
 69           }
 70         }
 71 
 72         // Send the heartbeat and process the jobtracker‘s directives
 73         //在transmitHeartBeat()函数处理中,
 74         //TaskTracker会创建一个新的TaskTrackerStatus对象记录目前任务的执行状况,
 75         //检查目前执行的Task数目以及本地磁盘的空间使用情况等,
 76         //如果可以接收新的Task则设置heartbeat()的askForNewTask参数为true。
 77         //然后通过IPC接口调用JobTracker的heartbeat()方法发送过去,
 78         //heartbeat()返回值TaskTrackerAction数组。
 79         HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);//发送Heartbeat到JobTracker,得到response 
 80 
 81         // Note the time when the heartbeat returned, use this to decide when to send the
 82         // next heartbeat   
 83         lastHeartbeat = System.currentTimeMillis();
 84         
 85         // Check if the map-event list needs purging
 86         Set<JobID> jobs = heartbeatResponse.getRecoveredJobs();
 87         if (jobs.size() > 0) {
 88           synchronized (this) {
 89             // purge the local map events list
 90             for (JobID job : jobs) {
 91               RunningJob rjob;
 92               synchronized (runningJobs) {
 93                 rjob = runningJobs.get(job);          
 94                 if (rjob != null) {
 95                   synchronized (rjob) {
 96                     FetchStatus f = rjob.getFetchStatus();
 97                     if (f != null) {
 98                       f.reset();
 99                     }
100                   }
101                 }
102               }
103             }
104 
105             // Mark the reducers in shuffle for rollback
106             synchronized (shouldReset) {
107               for (Map.Entry<TaskAttemptID, TaskInProgress> entry 
108                    : runningTasks.entrySet()) {
109                 if (entry.getValue().getStatus().getPhase() == Phase.SHUFFLE) {
110                   this.shouldReset.add(entry.getKey());
111                 }
112               }
113             }
114           }
115         }
116         //然后调用HeartbeatResponse的getActions()函数获得JobTracker传过来的所有指令即一个TaskTrackerAction数组。
117         TaskTrackerAction[] actions = heartbeatResponse.getActions();
118         if(LOG.isDebugEnabled()) {
119           LOG.debug("Got heartbeatResponse from JobTracker with responseId: " + 
120                     heartbeatResponse.getResponseId() + " and " + 
121                     ((actions != null) ? actions.length : 0) + " actions");
122         }
123         if (reinitTaskTracker(actions)) {
124           return State.STALE;
125         }
126             
127         // resetting heartbeat interval from the response.
128         heartbeatInterval = heartbeatResponse.getHeartbeatInterval();
129         justStarted = false;
130         justInited = false;
131         if (actions != null){ 
132             //遍历这个数组,如果是一个新任务指令即LaunchTaskAction则调用调用addToTaskQueue加入到待执行队列,
133             //否则加入到tasksToCleanup队列,交给一个taskCleanupThread线程来处理,如执行KillJobAction或者KillTaskAction等。
134           for(TaskTrackerAction action: actions) {
135             if (action instanceof LaunchTaskAction) {
136               addToTaskQueue((LaunchTaskAction)action);  //如果是运行一个新的Task,则将Action添加到任务队列中,加入TaskLauncher线程的执行队列
137             } else if (action instanceof CommitTaskAction) {
138               CommitTaskAction commitAction = (CommitTaskAction)action;
139               if (!commitResponses.contains(commitAction.getTaskID())) {
140                 LOG.info("Received commit task action for " + 
141                           commitAction.getTaskID());
142                 commitResponses.add(commitAction.getTaskID());
143               }
144             } else {//杀死任务或作业
145               tasksToCleanup.put(action);
146             }
147           }
148         }
149       //杀死一定时间没没有汇报进度的task 
150         markUnresponsiveTasks();
151       //当剩余磁盘空间小于mapred.local.dir.minspacekill(默认为0)时,寻找合适的任务将其杀掉以释放空间 
152         killOverflowingTasks();
153             
154         //we‘ve cleaned up, resume normal operation
155         if (!acceptNewTasks && isIdle()) {
156           acceptNewTasks=true;
157         }
158         //The check below may not be required every iteration but we are 
159         //erring on the side of caution here. We have seen many cases where
160         //the call to jetty‘s getLocalPort() returns different values at 
161         //different times. Being a real paranoid here.
162         checkJettyPort(server.getPort());
163       } catch (InterruptedException ie) {
164         LOG.info("Interrupted. Closing down.");
165         return State.INTERRUPTED;
166       } catch (DiskErrorException de) {
167         String msg = "Exiting task tracker for disk error:\n" +
168           StringUtils.stringifyException(de);
169         LOG.error(msg);
170         synchronized (this) {
171           jobClient.reportTaskTrackerError(taskTrackerName, 
172                                            "DiskErrorException", msg);
173         }
174         return State.STALE;
175       } catch (RemoteException re) {
176         String reClass = re.getClassName();
177         if (DisallowedTaskTrackerException.class.getName().equals(reClass)) {
178           LOG.info("Tasktracker disallowed by JobTracker.");
179           return State.DENIED;
180         }
181       } catch (Exception except) {
182         String msg = "Caught exception: " + 
183           StringUtils.stringifyException(except);
184         LOG.error(msg);
185       }
186     }
187 
188     return State.NORMAL;
189   }
View Code

  方法中有一个循环,没有其他状况会一直循环执行,循环内首先会阻塞心跳间隔时间,心跳间隔是动态的会不断修正的;如果TaskTracker是刚刚启动,需要先确认版本一致否则直接返回State.DENIED,然后获取system directory和filesystem;超过磁盘检查间隔就对磁盘进行检查,是否有损坏,然后检查localStorage.numFailures(),如果大于lastNumFailures,则直接返回State.STALE,对TaskTracker重新初始化; transmitHeartBeat(now)方法会发送心跳到JobTracker,并返回心跳响应信息heartbeatResponse,相应信息包括两部分,一个是作业集合recoveredJobs它是上次关闭JobTracker时正在运行的作业集合,重启JobTracker后需要恢复这些作业的运行状态(前提是用户启用了作业恢复功能),而TaskTracker收到该作业集合后,需重置这些作业对应Reduce Task的FetchStatus信息,从而迫使这些Reduce Task重新从Map Task端拷贝数据,另一部分是需要执行的命令列表;heartbeatResponse.getActions()获取JobTracker发送过来的指令数组,首先要检查的是是否有重新初始化TaskTracker的action,如果有则返回State.STALE,会重新初始化TaskTracker;修正心跳间隔;遍历actions数组,根据action的类型加入不同的数据结构中,是LaunchTaskAction:如果是action.getTask().isMapTask()==true,则将action加入mapLauncher,否则加入reduceLauncher,这两个launcher都是TaskLauncher(该类是TaskTracker类的一个内部类,具有一个数据成员,是 TaskTracker.TaskInProgress类型的队列。在此特别注意,在TaskTracker类内部所提到的TaskInProgress 类均为TaskTracker的内部类,我们用TaskTracker.TaskInProgress表示,一定要和mapred包中的 TaskInProgress类区分,后者我们直接用TaskInProgress表示);是CommitTaskAction:则放入commitResponses,commitResponses.add(commitAction.getTaskID());其他的类型放入tasksToCleanup.put(action),表示要清理,如执行KillJobAction或者KillTaskAction等。markUnresponsiveTasks()方法遍历runningTasks杀死一定时间没没有汇报进度的task(purgeTask(tip, true)方法)。killOverflowingTasks()方法,当剩余磁盘空间小于mapred.local.dir.minspacekill(默认为0)时,寻找合适的任务将其杀掉以释放空间,期间不接受任何新的task,acceptNewTasks=false,通过findTaskToKill(null)方法(会遍历runningTasks,优先考虑杀死reduce任务)找到合适的TaskInProgress killMe,执行purgeTask(killMe, false)。一旦空间不足而出现杀死task的情况出现,就会一直不接受任何新的task,直到所有的task执行完毕和所有的清理task也执行完毕,但仍会正常向JobTracker发送心跳信息,信息内容就会有所变化。

  transmitHeartBeat(now)方法是构造并将心跳信息发送到JobTracker。主要是构造一个TaskTrackerStatus这是要发送的东西,其内容包括任务的运行状态信息、TaskTracker资源信息、健康状态信息。如果可以接收新的Task则设置askForNewTask参数为true。当满足下面的条件的时候,此TaskTracker请求JobTracker为其分配一个新的Task来运行:当前TaskTracker正在运行的map task的个数小于可以运行的map task的最大个数;当前TaskTracker正在运行的reduce task的个数小于可以运行的reduce task的最大个数;acceptNewTasks==true。如果askForNewTask==true则对TaskTrackerStatus的实例status进行一些设置。然后对healthStatus = status.getHealthStatus()中的healthStatus进行一些设置。然后向JobTracker发送心跳并接受相应信息HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status, justStarted,justInited,askForNewTask, heartbeatResponseId)这是一个RPC调用,具体可以查看JobTracker.heartbeat方法,后续再讲。调用成功后更新heartbeatResponseId。遍历所有task的状态TaskStatus,对那些状态为SUCCEEDED或者FAILED或者KILLED,做一些统计信息,根据task类型使得mapTotal或者reduceTotal减一,runningTasks去除这个task,然后清空runningTasks中所有的TaskInProgress.getStatus().clearStatus()状态信息,这些状态信息是瞬时的,仅发送一次,status = null,这些瞬时的状态信息是在构造TaskTrackerStatus时通过cloneAndResetRunningTaskStatuses(sendCounters)生成的。最后返回心跳结果heartbeatResponse。

  这样offerService()方法通过while循环一直阻塞一定的心跳间隔,然后获取JobTracker的心跳应答信息,根据其中的action添加到不同的数据结构中,并做一些检查控制TaskTracker能够较为合理的运行,总是一遍又一遍的做这些。

  

  至此,TaskTracker的启动过程讲解完了,错误之处还请指正。

  参考:1、http://www.2cto.com/kf/201311/260826.html
     2、董西成,《hadoop技术内幕---深入理解MapReduce架构设计与实现原理》

TaskTracker启动过程源码级分析,布布扣,bubuko.com

TaskTracker启动过程源码级分析

标签:c   style   class   blog   code   java   

原文地址:http://www.cnblogs.com/lxf20061900/p/3745283.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!