您的位置: 首页  >  文章  >  Hadoop之TaskTraker分析（转）

Hadoop之TaskTraker分析（转）

分类: 文章 • 2023-11-22 22:16:04

TaskTracker的工作职责之前已经和大家提过，主要负责维护，申请和监控Task，通过heartbeat和JobTracker进行通信。

     TaskTracker的init过程：

     1.读取配置文件，解析参数

     2.将TaskTraker上原有的用户local files删除并新建新的dir和file

     3. Map<TaskAttemptID, TaskInProgress> tasks = new HashMap<TaskAttemptID, TaskInProgress>(); 清除map

     4.    this.runningTasks = new LinkedHashMap<TaskAttemptID, TaskInProgress>();记录task的链表
            this.runningJobs = new TreeMap<JobID, RunningJob>();记录job的id信息

     5.初始化JVMManager：

[java] view plaincopyprint?
 
  1. mapJvmManager = new JvmManagerForType(tracker.getMaxCurrentMapTasks(),   
  2.       true, tracker);  
  3.   reduceJvmManager = new JvmManagerForType(tracker.getMaxCurrentReduceTasks(),  
  4.       false, tracker);  

     6.初始化RPC，获取JobTracker client用于heartbeat通信；

 

     7.new一个 后台线程用于监听map完成的事件

[java] view plaincopyprint?
 
  1. this.mapEventsFetcher = new MapEventsFetcherThread();  
  2. mapEventsFetcher.setDaemon(true);  
  3. mapEventsFetcher.setName(  
  4.                          "Map-events fetcher for all reduce tasks " + "on " +   
  5.                          taskTrackerName);  
  6. mapEventsFetcher.start();  

    后台线程的run方法如下：

 

 

[java] view plaincopyprint?
 
  1. while (running) {  
  2.        try {  
  3.          List <FetchStatus> fList = null;  
  4.          synchronized (runningJobs) {  
  5.            while (((fList = reducesInShuffle()).size()) == 0) {  
  6.              try {  
  7.                runningJobs.wait();  
  8.              } catch (InterruptedException e) {  
  9.                LOG.info("Shutting down: " + this.getName());  
  10.                return;  
  11.              }  
  12.            }  
  13.          }  
  14.          // now fetch all the map task events for all the reduce tasks  
  15.          // possibly belonging to different jobs  
  16.          boolean fetchAgain = false; //flag signifying whether we want to fetch  
  17.                                      //immediately again.  
  18.          for (FetchStatus f : fList) {  
  19.            long currentTime = System.currentTimeMillis();  
  20.            try {  
  21.              //the method below will return true when we have not   
  22.              //fetched all available events yet  
  23.              if (f.fetchMapCompletionEvents(currentTime)) {  
  24.                fetchAgain = true;  
  25.              }  
  26.            } catch (Exception e) {  
  27.              LOG.warn(  
  28.                       "Ignoring exception that fetch for map completion" +  
  29.                       " events threw for " + f.jobId + " threw: " +  
  30.                       StringUtils.stringifyException(e));   
  31.            }  
  32.            if (!running) {  
  33.              break;  
  34.            }  
  35.          }  
  36.          synchronized (waitingOn) {  
  37.            try {  
  38.              if (!fetchAgain) {  
  39.                waitingOn.wait(heartbeatInterval);  
  40.              }  
  41.            } catch (InterruptedException ie) {  
  42.              LOG.info("Shutting down: " + this.getName());  
  43.              return;  
  44.            }  
  45.          }  
  46.        } catch (Exception e) {  
  47.          LOG.info("Ignoring exception "  + e.getMessage());  
  48.        }  
  49.      }  
  50.    }   

8.initializeMemoryManagement，初始化每个TrackTask的内存设置

 

9.new一个Map和Reducer的Launcher后台线程

 

[java] view plaincopyprint?
 
  1. mapLauncher = new TaskLauncher(TaskType.MAP, maxMapSlots);  
  2.  reduceLauncher = new TaskLauncher(TaskType.REDUCE, maxReduceSlots);  
  3.  mapLauncher.start();  
  4.  reduceLauncher.start();  

用于后面创建子JVM来执行map、reduce task

 

看一下

[java] view plaincopyprint?
 
  1. TaskLauncher的run方法：  
  2.  //before preparing the job localize   
  3.       //all the archives  
  4.       TaskAttemptID taskid = t.getTaskID();  
  5.       final LocalDirAllocator lDirAlloc = new LocalDirAllocator("mapred.local.dir");  
  6.       //simply get the location of the workDir and pass it to the child. The  
  7.       //child will do the actual dir creation  
  8.       final File workDir =  
  9.       new File(new Path(localdirs[rand.nextInt(localdirs.length)],   
  10.           TaskTracker.getTaskWorkDir(t.getUser(), taskid.getJobID().toString(),   
  11.           taskid.toString(),  
  12.           t.isTaskCleanupTask())).toString());  
  13.         
  14.       String user = tip.getUGI().getUserName();  
  15.         
  16.       // Set up the child task's configuration. After this call, no localization  
  17.       // of files should happen in the TaskTracker's process space. Any changes to  
  18.       // the conf object after this will NOT be reflected to the child.  
  19.       // setupChildTaskConfiguration(lDirAlloc);  
  20.   
  21.       if (!prepare()) {  
  22.         return;  
  23.       }  
  24.         
  25.       // Accumulates class paths for child.  
  26.       List<String> classPaths = getClassPaths(conf, workDir,  
  27.                                               taskDistributedCacheManager);  
  28.   
  29.       long logSize = TaskLog.getTaskLogLength(conf);  
  30.         
  31.       //  Build exec child JVM args.  
  32.       Vector<String> vargs = getVMArgs(taskid, workDir, classPaths, logSize);  
  33.         
  34.       tracker.addToMemoryManager(t.getTaskID(), t.isMapTask(), conf);  
  35.   
  36.       // set memory limit using ulimit if feasible and necessary ...  
  37.       String setup = getVMSetupCmd();  
  38.       // Set up the redirection of the task's stdout and stderr streams  
  39.       File[] logFiles = prepareLogFiles(taskid, t.isTaskCleanupTask());  
  40.       File stdout = logFiles[0];  
  41.       File stderr = logFiles[1];  
  42.       tracker.getTaskTrackerInstrumentation().reportTaskLaunch(taskid, stdout,  
  43.                  stderr);  
  44.         
  45.       Map<String, String> env = new HashMap<String, String>();  
  46.       errorInfo = getVMEnvironment(errorInfo, user, workDir, conf, env, taskid,  
  47.                                    logSize);  
  48.         
  49.       // flatten the env as a set of export commands  
  50.       List <String> setupCmds = new ArrayList<String>();  
  51.       for(Entry<String, String> entry : env.entrySet()) {  
  52.         StringBuffer sb = new StringBuffer();  
  53.         sb.append("export ");  
  54.         sb.append(entry.getKey());  
  55.         sb.append("=\"");  
  56.         sb.append(entry.getValue());  
  57.         sb.append("\"");  
  58.         setupCmds.add(sb.toString());  
  59.       }  
  60.       setupCmds.add(setup);  
  61.         
  62.       launchJvmAndWait(setupCmds, vargs, stdout, stderr, logSize, workDir);  
  63.       tracker.getTaskTrackerInstrumentation().reportTaskEnd(t.getTaskID());  
  64.       if (exitCodeSet) {  
  65.         if (!killed && exitCode != 0) {  
  66.           if (exitCode == 65) {  
  67.             tracker.getTaskTrackerInstrumentation().taskFailedPing(t.getTaskID());  
  68.           }  
  69.           throw new IOException("Task process exit with nonzero status of " +  
  70.               exitCode + ".");  
  71.         }  
  72.       }  
  73.     }  

run方法为当前task new一个child JVM，为其设置文件路径，上下文环境，JVM启动参数和启动命令等信息，然后调用TaskControll方法启动新的JVM执行对应的Task工作。

 

各个类关系图如下所示：

Hadoop之TaskTraker分析（转）

最后以TaskController的launchTask截至

10.然后开始  startHealthMonitor(this.fConf);

 

 

再来看看TaskLauncher的run方法，就是不停的循环去获取TaskTracker中新的task，然后调用startNewTask方法

 

[java] view plaincopyprint?
 
  1. if (this.taskStatus.getRunState() == TaskStatus.State.UNASSIGNED ||  
  2.          this.taskStatus.getRunState() == TaskStatus.State.FAILED_UNCLEAN ||  
  3.          this.taskStatus.getRunState() == TaskStatus.State.KILLED_UNCLEAN) {  
  4.        localizeTask(task);  
  5.        if (this.taskStatus.getRunState() == TaskStatus.State.UNASSIGNED) {  
  6.          this.taskStatus.setRunState(TaskStatus.State.RUNNING);  
  7.        }  
  8.        setTaskRunner(task.createRunner(TaskTracker.this, this, rjob));  
  9.        this.runner.start();  
  10.        long now = System.currentTimeMillis();  
  11.        this.taskStatus.setStartTime(now);  
  12.        this.lastProgressReport = now;  

TaskTracker的run方法：通过维护心跳和JobTracker通信，以获取、杀掉新的Task，重点看一下heartBeat通信过程：

 

 

[java] view plaincopyprint?
 
  1. synchronized (this) {  
  2.      askForNewTask =   
  3.        ((status.countOccupiedMapSlots() < maxMapSlots ||   
  4.          status.countOccupiedReduceSlots() < maxReduceSlots) &&   
  5.         acceptNewTasks);   
  6.      localMinSpaceStart = minSpaceStart;  
  7.    }  
  8.    if (askForNewTask) {  
  9.      askForNewTask = enoughFreeSpace(localMinSpaceStart);  
  10.      long freeDiskSpace = getFreeSpace();  
  11.      long totVmem = getTotalVirtualMemoryOnTT();  
  12.      long totPmem = getTotalPhysicalMemoryOnTT();  
  13.      long availableVmem = getAvailableVirtualMemoryOnTT();  
  14.      long availablePmem = getAvailablePhysicalMemoryOnTT();  
  15.      long cumuCpuTime = getCumulativeCpuTimeOnTT();  
  16.      long cpuFreq = getCpuFrequencyOnTT();  
  17.      int numCpu = getNumProcessorsOnTT();  
  18.      float cpuUsage = getCpuUsageOnTT();  
  19.   
  20.      status.getResourceStatus().setAvailableSpace(freeDiskSpace);  
  21.      status.getResourceStatus().setTotalVirtualMemory(totVmem);  
  22.      status.getResourceStatus().setTotalPhysicalMemory(totPmem);  
  23.      status.getResourceStatus().setMapSlotMemorySizeOnTT(  
  24.          mapSlotMemorySizeOnTT);  
  25.      status.getResourceStatus().setReduceSlotMemorySizeOnTT(  
  26.          reduceSlotSizeMemoryOnTT);  
  27.      status.getResourceStatus().setAvailableVirtualMemory(availableVmem);   
  28.      status.getResourceStatus().setAvailablePhysicalMemory(availablePmem);  
  29.      status.getResourceStatus().setCumulativeCpuTime(cumuCpuTime);  
  30.      status.getResourceStatus().setCpuFrequency(cpuFreq);  
  31.      status.getResourceStatus().setNumProcessors(numCpu);  
  32.      status.getResourceStatus().setCpuUsage(cpuUsage);  
  33.    }  
  34.    //add node health information  
  35.      
  36.    TaskTrackerHealthStatus healthStatus = status.getHealthStatus();  
  37.    synchronized (this) {  
  38.      if (healthChecker != null) {  
  39.        healthChecker.setHealthStatus(healthStatus);  
  40.      } else {  
  41.        healthStatus.setNodeHealthy(true);  
  42.        healthStatus.setLastReported(0L);  
  43.        healthStatus.setHealthReport("");  
  44.      }  
  45.    }  
  46.    //  
  47.    // Xmit the heartbeat  
  48.    //  
  49.    HeartbeatResponse heartbeatResponse = jobClient.heartbeat(status,   
  50.                                                              justStarted,  
  51.                                                              justInited,  
  52.                                                              askForNewTask,   
  53.                                                              heartbeatResponseId);  


该方法主要将TaskTracker上的各种性能参数信息反馈给JobTraker，调用其heartbeat方法然后解析返回的结果，下篇详细分析heartBeat机制

相关推荐