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

TaskTracker学习笔记

时间：2015-12-14 06:43:21      阅读：219      评论：0      收藏：0      [点我收藏+]

标签：

转自：http://blog.csdn.net/androidlushangderen/article/details/41477061

 上次分析完JobTracker通过TaskScheduler如何把作业分配给TaskTracker，这次把目光 移动到TaskTracker上面。TaskTracker在这里其实是一个slave的从属关系。我在后面的分析会通过TaskTracker的执行流程，主要讲他的2个过程的分析1.作业启动执行2.与JobTracker的heatbeat的过程。2个过程都是非常的典型。

         与JobTracker一样，TaskTracker也是作为一项服务所运行的，他也有自己的main函数入口。下面是一张全局的TaskTracker执行过程流程图：

技术分享

jvmManager负责为每个Task分配一个java虚拟机环境让其执行，避免任务之间的干扰，TaskMemoryManager负责任务内存的监控，对于某些任务恶意消耗资源内存，会给予杀死此任务的处理。

        1.TaskTracker任务启动

        下面从main函数的入口开始分析一下TaskTracker的执行流程：

 

[java] view plaincopyprint?
 
  1. /** 
  2.    * Start the TaskTracker, point toward the indicated JobTracker 
  3.    * taskTracker同样也是一个服务程序，main函数开始执行 
  4.    */  
  5.   public static void main(String argv[]) throws Exception {  
  6.     StringUtils.startupShutdownMessage(TaskTracker.class, argv, LOG);  
  7.     if (argv.length != 0) {  
  8.       System.out.println("usage: TaskTracker");  
  9.       System.exit(-1);  
  10.     }  
  11.     try {  
  12.       //初始化作业配置  
  13.       JobConf conf=new JobConf();  
  14.       // enable the server to track time spent waiting on locks  
  15.       ReflectionUtils.setContentionTracing  
  16.         (conf.getBoolean("tasktracker.contention.tracking", false));  
  17.       //初始化度量统计系统  
  18.       DefaultMetricsSystem.initialize("TaskTracker");  
  19.       //根据作业配置初始化TaskTracker  
  20.       TaskTracker tt = new TaskTracker(conf);  
  21.       //注册MBean,方便外界工具检测TaskTracker的状态  
  22.       MBeans.register("TaskTracker", "TaskTrackerInfo", tt);  
  23.       //执行TaskTracker服务主程序  
  24.       tt.run();  
  25.     } catch (Throwable e) {  
  26.       LOG.error("Can not start task tracker because "+  
  27.                 StringUtils.stringifyException(e));  
  28.       System.exit(-1);  
  29.     }  
  30.   }  

让后我们进入其中的执行主程序tt.run():

 

 

[java] view plaincopyprint?
 
  1. /** 
  2.    * The server retry loop.   
  3.    * This while-loop attempts to connect to the JobTracker.  It only  
  4.    * loops when the old TaskTracker has gone bad (its state is 
  5.    * stale somehow) and we need to reinitialize everything. 
  6.    */  
  7.   public void run() {  
  8.     try {  
  9.       getUserLogManager().start();  
  10.       //开启CleanUp清理线程  
  11.       startCleanupThreads();  
  12.       boolean denied = false;  
  13.       while (running && !shuttingDown && !denied) {  
  14.         boolean staleState = false;  
  15.         try {  
  16.           // This while-loop attempts reconnects if we get network errors  
  17.           while (running && !staleState && !shuttingDown && !denied) {  
  18.             try {  
  19.               //offerService()执行了核心的启动操作  
  20.               State osState = offerService();  
  21.               if (osState == State.STALE) {  
  22.                 staleState = true;  
  23.               } else if (osState == State.DENIED) {  
  24.                 denied = true;  
  25.               }  
  26.               ......  

我们可以看到，这里通过while操作，循环进行服务操作，如果拒绝服务，则会shutdown中断服务，服务的主要操作又在offerService方法中：

 

 

[java] view plaincopyprint?
 
  1. /** 
  2.    * Main service loop.  Will stay in this loop forever. 
  3.    * 主要的循环服务操作 
  4.    */  
  5.   State offerService() throws Exception {  
  6.       .....  
  7.       // Send the heartbeat and process the jobtracker‘s directives  
  8.         //发送给JobTracker心跳包  
  9.         HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);  
  10.   
  11.         // Note the time when the heartbeat returned, use this to decide when to send the  
  12.         // next heartbeat     
  13.         lastHeartbeat = System.currentTimeMillis();  
  14.           
  15.         ....          
  16.         //在这里获取了心跳回应中的action命令  
  17.         TaskTrackerAction[] actions = heartbeatResponse.getActions();  
  18.         if(LOG.isDebugEnabled()) {  
  19.           LOG.debug("Got heartbeatResponse from JobTracker with responseId: " +   
  20.                     heartbeatResponse.getResponseId() + " and " +   
  21.                     ((actions != null) ? actions.length : 0) + " actions");  
  22.         }  
  23.         if (reinitTaskTracker(actions)) {  
  24.           return State.STALE;  
  25.         }  
  26.               
  27.         // resetting heartbeat interval from the response.  
  28.         heartbeatInterval = heartbeatResponse.getHeartbeatInterval();  
  29.         justStarted = false;  
  30.         justInited = false;  
  31.         if (actions != null){   
  32.           for(TaskTrackerAction action: actions) {  
  33.             if (action instanceof LaunchTaskAction) {  
  34.               //如果是执行Task任务指令，执行添加任务到任务队列中  
  35.               addToTaskQueue((LaunchTaskAction)action);  
  36.             } else if (action instanceof CommitTaskAction) {  
  37.              //如果是提交任务的指令，则执行后面的操作  
  38.               CommitTaskAction commitAction = (CommitTaskAction)action;  
  39.               if (!commitResponses.contains(commitAction.getTaskID())) {  
  40.                 LOG.info("Received commit task action for " +   
  41.                           commitAction.getTaskID());  
  42.                 commitResponses.add(commitAction.getTaskID());  
  43.               }  
  44.             } else {  
  45.               //其他的指令一律添加到tasksToCleanup队列中等待被处理  
  46.               tasksToCleanup.put(action);  
  47.             }  
  48.           }  
  49.         }  
  50.         .....  

在这里我省略了比较多的代码，把执行任务相关的核心操作保留了，在这里就开始执行了后面的和Task相关的很多操作了，当然这些任务都是通过收到JobTracker的心跳包Response来获得的，在通过获取里面的TaskTrackerAction命令来判断执行的。TaskTrackerAction里面包含了1枚举类，包括了以下的相关指令:

 

技术分享

具体什么意思，看上面的英文解释就能理解了吧，上面代表了6种命令操作，我们侧重看第一个launch_task的命令执行，在上面的判断执行方法是addToTaskQueue();方法:

 

[java] view plaincopyprint?
 
  1. private void addToTaskQueue(LaunchTaskAction action) {  
  2.     //任务类型加入到任务待执行的容器中  
  3.     if (action.getTask().isMapTask()) {  
  4.       mapLauncher.addToTaskQueue(action);  
  5.     } else {  
  6.       reduceLauncher.addToTaskQueue(action);  
  7.     }  
  8.   }  

这里的mapLauncher,reduceLauncher的类型是TaskLauncher，他是一个线程类：

 

 

[java] view plaincopyprint?
 
  1. class TaskLauncher extends Thread {  
  2.     private IntWritable numFreeSlots;  
  3.     private final int maxSlots;  
  4.     private List<TaskInProgress> tasksToLaunch;  
  5.    ....  

也就是说，待执行的map，Reduce任务都是添加到taskToLauch中的，

[java] view plaincopyprint?
 
  1. public void addToTaskQueue(LaunchTaskAction action) {  
  2.       //新建1个TIP，并加入tasksToLaunch列表  
  3.       synchronized (tasksToLaunch) {  
  4.         TaskInProgress tip = registerTask(action, this);  
  5.         tasksToLaunch.add(tip);  
  6.         //唤醒所有被tasksToLaunch wait的操作，说明此时有新的任务了  
  7.         tasksToLaunch.notifyAll();  
  8.       }  
  9.     }  

加入之后唤醒相应的操作，这个就很好理解了，一定是在empty的时候被阻塞住了，

 

 

[java] view plaincopyprint?
 
  1. public void run() {  
  2.       while (!Thread.interrupted()) {  
  3.         try {  
  4.           TaskInProgress tip;  
  5.           Task task;  
  6.           synchronized (tasksToLaunch) {  
  7.             while (tasksToLaunch.isEmpty()) {  
  8.               tasksToLaunch.wait();  
  9.             }  
  10.             //get the TIP  
  11.             tip = tasksToLaunch.remove(0);  
  12.             task = tip.getTask();  
  13.             LOG.info("Trying to launch : " + tip.getTask().getTaskID() +   
  14.                      " which needs " + task.getNumSlotsRequired() + " slots");  
  15.           }  
  16.           //wait for free slots to run  
  17.           .....  
  18.           //got a free slot. launch the task  
  19.           startNewTask(tip);  

到了startNewTask就是开始所谓的任务了。到此为止，TaskTracker的任务执行这条路，我们算彻底打通了，相关时序图如下：

 

技术分享

        2.Heateat过程

        下面我们看另外一个流程，心跳机制。此过程的实现同样的主要是在offerService的循环操作中。首先第一步，判断是否到了发送心跳包的时间，因为心跳包是隔周期性的时间发送的，所以这里必须会进行判读：

 

[java] view plaincopyprint?
 
  1. /** 
  2.    * Main service loop.  Will stay in this loop forever. 
  3.    * 主要的循环服务操作 
  4.    */  
  5.   State offerService() throws Exception {  
  6.     long lastHeartbeat = System.currentTimeMillis();  
  7.   
  8.     while (running && !shuttingDown) {  
  9.       try {  
  10.         long now = System.currentTimeMillis();  
  11.           
  12.         // accelerate to account for multiple finished tasks up-front  
  13.         //判断上次心跳的时间+心跳等待时间是否已经到了当前时间，如果到了可以发送新的心跳包  
  14.         long remaining =   
  15.           (lastHeartbeat + getHeartbeatInterval(finishedCount.get())) - now;  
  16.         //如果还没到，时间有剩余，则要强行等待剩余的时间  
  17.         while (remaining > 0) {  
  18.           // sleeps for the wait time or   
  19.           // until there are *enough* empty slots to schedule tasks  
  20.           synchronized (finishedCount) {  
  21.             finishedCount.wait(remaining);  
  22.               
  23.             // Recompute  
  24.             now = System.currentTimeMillis();  
  25.             remaining =   
  26.               (lastHeartbeat + getHeartbeatInterval(finishedCount.get())) - now;  
  27.               
  28.             if (remaining <= 0) {  
  29.               // Reset count   
  30.               finishedCount.set(0);  
  31.               break;  
  32.             }  
  33.           }  
  34.         }  
  35.         .....  

假设已经到达了发送时间了，会执行后面的操作，检测版本号，TaskTracker和JobTracker的版本号必须一致:

 

 

[java] view plaincopyprint?
 
  1. .....  
  2.         // If the TaskTracker is just starting up:  
  3.         // 1. Verify the buildVersion  
  4.         // 2. Get the system directory & filesystem  
  5.         if(justInited) {  
  6.          //验证版本号，如果版本号不对，则返回拒绝状态  
  7.           String jobTrackerBV = jobClient.getBuildVersion();  
  8.           if(!VersionInfo.getBuildVersion().equals(jobTrackerBV)) {  
  9.             String msg = "Shutting down. Incompatible buildVersion." +  
  10.             "\nJobTracker‘s: " + jobTrackerBV +   
  11.             "\nTaskTracker‘s: "+ VersionInfo.getBuildVersion();  
  12.             LOG.error(msg);  
  13.             try {  
  14.               jobClient.reportTaskTrackerError(taskTrackerName, null, msg);  
  15.             } catch(Exception e ) {  
  16.               LOG.info("Problem reporting to jobtracker: " + e);  
  17.             }  
  18.             return State.DENIED;  
  19.           }  

如果通过上述2个验证，基本上就达到了发送的条件了，下面就准备发送操作了:

 

 

[java] view plaincopyprint?
 
  1. // Send the heartbeat and process the jobtracker‘s directives  
  2.         //发送给JobTracker心跳包  
  3.         HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);  

就是在上面这个方法中实现了发送的操作，此方法的返回值是JobTracker的心跳回复包，里面就包含着刚刚的TaskTrackerAction命令信息。我们进入transmitHeartBeat。之前分析过，心跳机制的有1个主要的作用就是汇报TaskTracker的资源使用情况和作业执行情况给JobTracker节点。以此可以让主节点可以进行资源调配。所以在上面的这个方法必不可少的操作是构建TaskTracker的Status状态信息。这个类包含的信息还比较多。下面是主要的此类的关系结构:

 

技术分享

里面2大包含类ResourceStatus(TaskTracker资源使用情况)，TaskTrackerHealthStatus(TaskTracker节点健康状况)。首先当然是新建一个Status了:

 

[java] view plaincopyprint?
 
  1. /** 
  2.    * Build and transmit the heart beat to the JobTracker 
  3.    * 将TaskTracker自身的状态信息发送给JobTracker，并获得一个心跳包的回应 
  4.    * @param now current time 
  5.    * @return false if the tracker was unknown 
  6.    * @throws IOException 
  7.    */  
  8.   HeartbeatResponse transmitHeartBeat(long now) throws IOException {  
  9.     ....  
  10.     //   
  11.     // Check if the last heartbeat got through...   
  12.     // if so then build the heartbeat information for the JobTracker;  
  13.     // else resend the previous status information.  
  14.     //  
  15.     if (status == null) {  
  16.       synchronized (this) {  
  17.         status = new TaskTrackerStatus(taskTrackerName, localHostname,   
  18.                                        httpPort,   
  19.                                        cloneAndResetRunningTaskStatuses(  
  20.                                          sendCounters),   
  21.                                        failures,   
  22.                                        maxMapSlots,  
  23.                                        maxReduceSlots);   
  24.       }  
  25.    

后面就是各种获取节点CPU，内存等基本信息，这里就不列举了，不过这里提一点，对于TaskTracker是否还能运行任务，在这里是通过TaskTracker是否达到了它的maxSlot上限作为1个标准。一般1个Reduce Task占据1个slot单元，1个Map Task同样占据1个Slot单元，如果1个TaskTracker结点拥有好多slot单元，那么他就可以运行很多Task。

 

 

[java] view plaincopyprint?
 
  1. //  
  2.     // Check if we should ask for a new Task  
  3.     // 检测TaskTracker是否需要一个新 Task任务  
  4.     //  
  5.     boolean askForNewTask;  
  6.     long localMinSpaceStart;  
  7.     synchronized (this) {  
  8.       //通过判断当前所占据的slots数量是否已经达到最大slot的数量作为标准  
  9.       askForNewTask =   
  10.         ((status.countOccupiedMapSlots() < maxMapSlots ||   
  11.           status.countOccupiedReduceSlots() < maxReduceSlots) &&   
  12.          acceptNewTasks);   
  13.       localMinSpaceStart = minSpaceStart;  
  14.     }  

askForNewTask布尔类型就代表TaskTracker是否还能运行新的任务，封装好了这些Status信息之后，就要执行关键的发送步骤了:

 

 

[java] view plaincopyprint?
 
  1. //  
  2. // Xmit the heartbeat  
  3. // 通过JobClient发送给JobTracker,并获得1个回复  
  4. //  
  5. HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status,   
  6.                                                           justStarted,  
  7.                                                           justInited,  
  8.                                                           askForNewTask,   
  9.                                                           heartbeatResponseId);  

是通过JobClient的方法发送的。得到的heartbeatResponse返回结果就是JobTracker结果了。至于里面JobClient具体怎么发送就不是本次分析的重点了，HeartBeat也分析完毕。同样看一下流程图:

 

技术分享

      

     总结

     2个过程都是在offerService核心服务程序中执行的。了解完JobTracker和TaskTracker的工作原理，在聊了具体Task任务的执行的5个阶段，从微观Task细节的执行到宏观上作业调度的原理分析理解，的确对MapReduce计算模型的理解上升了许多的层次。

TaskTracker学习笔记

标签：

原文地址：http://www.cnblogs.com/cxzdy/p/5044009.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！