TaskTracker节点的内部设计与实现

青旅半醒 2022-06-13 10:21 200阅读 0赞

众所周知，Hadoop采用master-slave的架构设计来实现Map-Reduce框架，它的JobTracker节点作为主控节点来管理和调度用户提交的作业，TaskTracker节点作为工作节点来负责执行JobTracker节点分配的Map/Reduce任务。整个集群由一个JobTracker节点和若干个TaskTracker节点组成，当然，JobTracker节点也负责对TaskTracker节点进行管理。在前面一系列的博文中，我已经比较系统地讲述了JobTracker节点内部的设计与实现，而在本文，我将对TaskTracker节点的内部设计与实现进行一次全面的概述。

TaskTracker节点作为工作节点不仅要和JobTracker节点进行频繁的交互来获取作业的任务并负责在本地执行他们，而且也要和其它的TaskTracker节点交互来协同完成同一个作业。因此，在目前的Hadoop-0.20.2.0实现版本中，对工作节点TaskTracker的设计主要包含三类组件：服务组件、管理组件、工作组件。服务组件不仅负责与其它的TaskTracker节点而且还负责与JobTracker节点之间的通信服务，管理组件负责对该节点上的任务、作业、JVM实例以及内存进行管理，工作组件则负责调度Map/Reduce任务的执行。这三大组件的详细构成如下：

![1333807599_3775.jpg][]

下面来详细的介绍这三类组件：

## 服务组件 ##

TaskTracker节点内部的服务组件不仅用来为TaskTracker节点、客户端提供服务，而且还负责向TaskTracker节点请求服务，这一类组件主要包括HttpServer、TaskReportServer、JobClient三大组件。

1.HttpServer

TaskTracker节点在其内部使用Jetty Web容器来开启http服务，这个http服务一是用来为客户端提供Task日志查询服务，二是用来提供数据传输服务，即在执行Reduce任务时是通过TaskTracker节点提供的该http服务来获取属于自己的map输出数据。这里需要详细介绍的是与该服务相关的配置参数，集群管理者可以通过TaskTracker节点的配置文件来配置该服务地址和端口号，对应的配置项为：mapred.task.tracker.http.address。同时，为了能够灵活的控制该该服务的吞吐量，管理者还可以设置该http服务的内部工作线程数量，对应的配置为：tasktracker.http.threads。

2.Task Reporter

TaskTracker节点在接收到JobTracker节点发送过来的Map/Reduce任务之后，会把它们交给JVM实例来执行，而自己则需要收集这些任务的执行进度信息，这就使得Task在JVM实例中执行的时候需要不断地向TaskTracker节点报告当前的执行情况。虽然TaskTracker节点和JVM实例在同一台机器上，但是它们之间的进程通信却是通过网络I/O来完成的(此处并不讨论这种通信方式的性能)，也就是TaskTracker节点在其内部开启一个端口来专门为任务实例提供进度报告服务。该服务地址可以通过配置项mapred.task.tracker.report.address来设置，而服务内部的工作线程的数量取2倍于该TaskTracker节点上的Map/Reduce Slot数量中的大者。该服务接口如下：

interface TaskUmbilicalProtocol extends VersionedProtocol {
    
      public static final long versionID = 16L;
      
      /**
       *  从TaskTracker节点上获取一个属于该JVM实例的任务
       */
      JvmTask getTask(JVMId jvmId) throws IOException;
    
      /**
       * 向TaskTracker节点报告当前任务实例执行的进度及状态信息
       */
      boolean statusUpdate(TaskAttemptID taskId, TaskStatus taskStatus) 
      throws IOException, InterruptedException;
      
      /** 
       * 向TaskTracker节点报告当前任务实例发生异常的原因
       */
      void reportDiagnosticInfo(TaskAttemptID taskid, String trace) throws IOException;
      
      /**
       * 
       */
      void reportNextRecordRange(TaskAttemptID taskid, SortedRanges.Range range) 
        throws IOException;
    
      /** 
       */
      boolean ping(TaskAttemptID taskid) throws IOException;
    
      /** 
       * 向TaskTracker节点发送一个心跳包，以此来告知TaskTracker节点自己还在正常执行
       */
      void done(TaskAttemptID taskid) throws IOException;
      
      /** 
       * 向TaskTracker节点发送任务提交预约请求
       */
      void commitPending(TaskAttemptID taskId, TaskStatus taskStatus) 
      throws IOException, InterruptedException;  
    
      /**
       * 向TaskTracker节点询问自己是否能够提交任务
       */
      boolean canCommit(TaskAttemptID taskid) throws IOException;
    
      /** 
       * 向TaskTracker节点报告获取一个Map输出失败
       */
      void shuffleError(TaskAttemptID taskId, String message) throws IOException;
      
      /** 
       * 向TaskTracker节点报告执行一个任务实例是发送文件系统的错误 
       */
      void fsError(TaskAttemptID taskId, String message) throws IOException;
    
      /** 
       * 向TaskTracker节点报告一个任务实例执行失败
       */
      void fatalError(TaskAttemptID taskId, String message) throws IOException;
      
      /** 
       * 从TaskTracker节点获取所属作业刚刚完成的Map任务
       */
      MapTaskCompletionEventsUpdate getMapCompletionEvents(JobID jobId, 
                                                           int fromIndex, 
                                                           int maxLocs,
                                                           TaskAttemptID id) 
      throws IOException;
    
    }

3.JobClient

TaskTracker节点与JobTracker节点进行通信的客户端组件，该组件负责为TaskTracker节点提供与系统有关的查询服务，该客户端的API为：

interface InterTrackerProtocol extends VersionedProtocol {
      
      public static final long versionID = 25L;
      
      public final static int TRACKERS_OK = 0;
      public final static int UNKNOWN_TASKTRACKER = 1;
    
      /**
       * 向JobTracker节点发送心跳跑同时报告自己当前的状态及任务实例执行的进度信息等，并且顺便获取分配的新任务实例
       */
      HeartbeatResponse heartbeat(TaskTrackerStatus status, 
                                  boolean restarted, 
                                  boolean initialContact,
                                  boolean acceptNewTasks,
                                  short responseId)
        throws IOException;
      
      /**
       * 获取集群依赖的文件系统的类型
       */
      public String getFilesystemName() throws IOException;
    
      /**
       * 向JobTracker节点报告TaskTracker节点出错的原因
       */
      public void reportTaskTrackerError(String taskTracker,
                                         String errorClass,
                                         String errorMessage) throws IOException;
      /**
       * 从JobTracker节点获取某一作业最近的任务完成事件
       */
      TaskCompletionEvent[] getTaskCompletionEvents(JobID jobid, int fromEventId
          , int maxEvents) throws IOException;
    
      /**
       * 获取集群的系统目录
       */
      public String getSystemDir();
      
      
      /**
       * 从JobTracker节点获取其提供该服务的版本号
       */
      public String getBuildVersion() throws IOException;
    }

##     ##

## 管理组件 ##

1.JVM Manager

TaskTracker节点对任务实例执行的核心就是将其交由一个的合适的JVM进程来负责执行，这里要强调的是JVM实例和TaskTracker节点是在同一台机器上的两个不同的进程，因此运行在每一个TaskTracker节点上的JVM实例是有限制的，同时又两类JVM实例，一类用来运行Map任务，另一类则用来运行Reduce任务。这两类JVM实例的最大数量可分别由配置文件中的mapred.tasktracker.map.tasks.maximum、mapred.tasktracker.reduce.task.maximum项来设置。TaskTracker节点利用JvmManager来管理该节点上所有的JVM实例，一方面它要保证运行在该TaskTracker节点上的各类JVM实例数量不能超过对应的上限值；另一方面还要负责分配合适的任务实例到空闲的JVM上执行，这是因为频繁地关闭/开始JVM线程是很耗时间的，而且一个JVM实例只能运行同一个作业的Map/Reduce任务。

2.Memory Manager

Hadoop集群的成本低主要是因为它部署在普通的低端PC上，以至于集群中的每一个节点的物理资源是有限的，特别是内存资源。任何作业的任务在TaskTracker节点上执行时都或多或少地需要一定的内部，但不同类型的作业需要的内存量是不一样的。很明显，仅仅通过开启一个JVM实例时设置该进程可使用的内存的上限值是不准确的、不够优化的，所以TaskTracker节点通过TaskMemoryManager来统计该节点当前内存的使用情况，并将其报告给JobTracker节点，JobTracker节点就可以根据该TaskTracker节点内存的剩余为其分配合适的任务了。

3.Task/Job Manager

TaskTracker节点通过Task、Job两个级别来共同的管理正在其上运行的任务，本来在Task级别管理任务是足以的，但为什么还要在Job级别再设置一层管理呢？这是因为，在前面刚介绍给，TaskTracker节点对JVM实例的调度级别是Job一级的，而JVM实例执行的单位是Task级别。

#     #

# 工作组件 #

1.MapEvent Fetcher

MapEvent Fetcher组件是TaskTracker节点内部的一个后台工作线程，用来向JobTracker节点请求某一个作业的任务完成事件，这个作业是正在TaskTracker节点上执行的。这个组件存在的意义在于Reduce任务在执行的时候需要Map任务的输出数据，因此Reduce任务在执行reduce操作之前需要shuffle所有的map输出到本地。

2.Map/Reduce Launcher

Map/Reduce Launcher组件也是TaskTracker节点内部的两个后台工作线程，它们分别负责调度合适的Map/Reduce任务到对应空闲/新的JVM实例上执行。当然，在将Task交给JVM实例运行之前还负责该任务对应的作业即自身的本地化。

3.TaskCleanup

TaskTracker节点每一次向JobTracker节点发送心跳包之后，除了带回一些需要JVM实例执行的任务之外，还带回来了一些操作命令，如KillJob、KillTask等，这一类任务并不需要JVM实例来执行，因此，TaskTracker节点在其内部开启了一个后台线程来专门的处理这些由于某些原因需要kill掉在正在其上执行的Job和Task。Job或Task的kill操作主要包括关闭对应的JVM实例，清理它们在TaskTracker节点上占用的临时Disk/Mem存储空间。对于在TaskTracker节点上执行的Job/Task所占用的临时磁盘存储空间的清理，也是由一个后台线程来全权负责处理的，这样做一是因为文件或目录的删除操作可能比较消耗时间，二是这些文件或目录已经没有任何用处了而不需要急着清除。

[1333807599_3775.jpg]: https://img-my.csdn.net/uploads/201204/07/1333807599_3775.jpg