MapReduce V1:JobTracker处理Heartbeat流程分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。这篇文章的内容,更多地主要是描述处理/交互流程性的东西,大部分流程图都是经过我梳理后画出来的(开始我打算使用序列图来描述流程,但是发现很多流程在单个对象内部都已经非常复杂,想要通过序列图表达有点担心描述不清,所以选择最基本的程序流程图),可能看起来比较枯燥,重点还是关注主要的处理流程要点,特别的地方我会刻意标示出来,便于理解。
JobTracker与TaskTracker之间通过org.apache.hadoop.mapred.InterTrackerProtocol协议来进行通信,TaskTracker通过该接口进行远程调用实现Heartbeat消息的发送,协议方法定义如下所示:

  HeartbeatResponse heartbeat(TaskTrackerStatus status,
                              boolean restarted,
                              boolean initialContact,
                              boolean acceptNewTasks,
                              short responseId) throws IOException;

通过该方法可以看出,最核心的Heartbeat报告数据都封装在TaskTrackerStatus对象中,JobTracker端会接收TaskTracker周期性地发送的心跳报告,根据这些心跳信息来更新整个Hadoop集群中计算资源的状态/数量,以及Task的运行状态。
另外,在JobTracker端维护的对象的数据结构,主要包括如下3个:

  • TaskTracker:这个类是在JobTracker端定义的,描述了TaskTracker的基本信息和状态(需要注意的是:它与TaskTracker进程的实现类同名,但是含义完全不同)
  • JobInProgress:简写JIP,在JobTracker端用来描述,JobClient提交的Job运行状态的数据结构,一个JIP对象还包含了组成一个Job的Task对应的一组TIP的信息
  • TaskInProgress:简写TIP,在JobTracker端用来描述,在TaskTracker上运行的Task状态的数据结构(需要注意的是:在TaskTracker端也对应一个TaskInProgress实现类,它与JobTracker端的同名,但是所包含的内容也并不完全相同)
  • TaskAttemptID:简写TAID,它是唯一标识了组成一个Job的Task的一个运行实例,一个Task(MapTask/ReduceTask)可能运行多次,比如第一次运行失败,对应一个失败的TAID,第二次调度又运行,又对应一个新的TAID;再比如,推测执行,可能会对应着同一个Task的、具有2个不同TAID的Task运行实例
  • TaskTrackerStatus结构

    TaskTrackerStatus对象要在网络间进行序列化传输,所以实现了接口org.apache.hadoop.io.Writable,该对象的数据结构,如下图所示:
    jt-hb-TaskTrackerStatus
    TaskTrackerStatus中各个数据项的含义,说明如下表所示:

    字段名称 字段类型 说明
    trackerName String TaskTracker的名称,例如:tracker_ + localHostname + : + taskReportAddress
    host String TaskTracker所在主机名称
    httpPort int HTTP端口号,默认50030
    taskFailures int 在该TaskTracker上运行失败的Task的个数
    dirFailures int 在TaskTracker节点上,配置的mapred.local.dir目录失败的个数
    maxMapTasks int 在该TaskTracker上同时运行Map的最大个数,通过mapred.tasktracker.map.tasks.maximum配置的,默认值是2
    maxReduceTasks int 在该TaskTracker上同时运行Map的最大个数,通过 mapred.tasktracker.reduce.tasks.maximum配置的,默认值是2
    resStatus ResourceStatus 在该TaskTracker上的资源情况,主要包括如下内容:虚拟内存大小、物理内存大小、Map slot数量、Reduce slot数量、可用磁盘空间、可用虚拟内存大小、可用物理内存大小、处理器数量、CPU频率、CPU使用百分比、累积CPU时间
    taskStatus TaskStatus 在该TaskTracker上,当前task的状态,它有分为MapTaskStatus和ReduceTaskStatus,主要包含如下内容:taskid(TaskAttemptID)、运行进度百分比、运行状态、诊断信息、所在TaskTracker名称、slot数、开始时间、结束时间、执行阶段(Phase)、一组计数器信息
    taskTrackerHealthStatus TaskTrackerHealthStatus TaskTracker的健康状态信息

    下面,主要对ResourceStatus、TaskStatus、TaskTrackerHealthStatus进行说明:

    • ResourceStatus

    ResourceStatus封装了一个TaskTracker节点的资源信息,结构如下图所示:
    jt-hb-ResourceStatus

    • TaskStatus

    TaskStatus封装了一个TaskTracker节点上运行的Task的状态信息,结构如下图所示:
    jt-hb-TaskStatus
    上图将TaskStatus的包含的数据结构全部展开,可以根据字段含义来了解它所描述的一些信息。

    • TaskTrackerHealthStatus

    TaskTrackerHealthStatus封装了TaskTracker的健康状态信息,如下图所示:
    jt-hb-TaskTrackerHealthStatus

    JobTracker处理Heartbeat流程

    JobTracker处理Heartbeat的流程,如果把每个处理细节都详细地展开,非常地复杂,可能从头到尾描述下来会感觉枯燥无味,所以这里我先概要地描述JobTracker处理Heartbeat的整体流程,然后再按照功能划分出一个个看似还算独立的子处理流程,单独地进行详细说明,这样能够更容易理解。整体处理流程,如下图所示:
    jt-hb-overview
    下面,我们根据上面的Heartbeat处理流程图,概要地说明Heartbeat是如何处理的,流程描述如下所示:

    1. TaskTracker创建一个TaskTrackerStatus对象,TaskTrackerStatus内部封装的信息包括:TaskTracker所在节点的基本信息、运行在TaskTracker上的Task的状态信息、TaskTracker服务的健康状态信息、TaskTracker的资源信息,另外发送心跳的RPC方法还包括restarted(TaskTracker是否重启)、initialContact(TaskTracker是否初次连接JobTracker)、acceptNewTasks(TaskTracker是否能够运行新的Task)、responseId(心跳响应ID),通过InterTrackerProtocol协议的heartbeat方法发送给JobTracker。
    2. JobTracker接收到TaskTracker发送的心跳数据。
    3. JobTracker检查TaskTracker的host是否在黑名单中,如果TaskTracker在黑名单中,则直接抛出异常终止RPC调用,否则继续下一步流程。
    4. 检查TaskTracker RPC调用参数restarted的值,如果TaskTracker重启了,则标记TaskTracker状态为健康状态;如果TaskTracker没有重启,则检查是否可以指派任务在该TaskTracker上运行。
    5. 如果TaskTracker不是初次连接JobTracker,检查JobTracker是否存在上一次向该TaskTracker发送的Heartbeat响应数据,存在的话则说明TaskTracker因为失去了与JobTracker之间的RPC连接而没有接收到,JobTracker直接再给TaskTracker重新发送该响应数据;不存在的话,若JobTracker重启了,使TaskTracker重新加入集群,需要通知Recovery Manager从恢复列表中移除该TaskTracker,若JobTracker未重启,这种情况几乎是不可能存在的(既然TaskTracker不是初次连接,JobTracker也没有重启,JobTracker端不可能没有保存Heartbeat响应数据)。
    6. 处理JobTracker接收到的TaskTracker的Heartbeat信息,主要是TaskTrackerStatus封装的数据。
    7. 根据处理Heartbeat数据结果,如果TaskTracker需要重新初始化,则发送一个带有ReinitTrackerAction指令的Heartbeat响应数据,否则TaskTracker不需要重新初始化则继续下一步流程。
    8. 检查是否可以向该TaskTracker指派任务,如果可以可以向该TaskTracker指派任务,则直接使用TaskScheduler指定的调度策略,选择当前可以指派给TaskTracker的一组需要启动的Task(对应指令LaunchTaskAction)。
    9. 根据TaskScheduler调度策略选择的需要启动的Task,并根据TaskTracker发送的Task状态报告,继续选择一些已经完成/需要被清理的Task分配给TaskTracker:先检查在该TaskTracker上是否有完成的Job,计算属于这些Job的需要被Kill掉(对应指令KillTaskAction)的Task;再检查是否有完成的Job,并且对应在该TaskTracker上的Task需要被清理(对应指令KillTaskAction);最后检查是否有已经完成需要被提交的Task(以此来通知TaskTracker提交Task完成并更新状态,对应指令CommitTaskAction)。
    10. 构造一个包含可调度Task(LaunchTaskAction/KillTaskAction/CommitTaskAction)的HeartbeatResponse对象,更新JobTracker内部维护的trackerToHeartbeatResponseMap映射。根据TaskTracker的Heartbeat报告的Task状态信息,对标记为完成的Task,更新JobTracker内部维护的多个队列和Map:trackerToMarkedTasksMap、taskidToTrackerMap、trackerToTaskMap、taskidToTIPMap。最后,返回TaskTracker调用的结果:HeartbeatResponse对象。

    上面流程图中,黑色虚线所表示的处理流程,我们说明一下:这种情况是不可能出现的,因为TaskTracker不是第一次连接JobTracker,而JobTracker端还没有上一次TaskTracker发送的Heartbeat对应的HeartbeatResponse,同时JobTracker又没有重启动过,所以这种条件是不存在的,那么该流程分支也不可能执行,故而用虚线描述,指向发送一个带有ReinitTrackerAction的HeartbeatResponse。

    下面,我们细化整个流程,将一些比较重要的流程详细分析说明:

    TaskTracker与JobTracker失去连接,更新状态

    JobTracker如果在给定超时时间范围之内没有收到TaskTracker的Heartbeat报告,会认为该TaskTracker已经无法执行/指派任务,那么在JobTracker端与该TaskTracker相关的数据结构都需要更新,受到影响的Job和Task的数据结构也需要更新,具体处理流程如下图所示:
    jt-hb-lost-TaskTracker
    上述流程图描述的流程,如下所示:

    1. 从队列Map<String, Set<JobID>> trackerToJobsToCleanup中移除在该TaskTracker上已经完成且需要清理的所有Job。
    2. 从队列Map<String, Set<TaskAttemptID>> trackerToTasksToCleanup中移除在TaskTracker上已经运行完成且需要清理的所有Task。
    3. 通知Recovery Manager从其维护的Set<String>类型的恢复列表JobTracker.RecoveryManager.recoveredTrackers中移除该TaskTracker。
    4. 从TreeMap<String, Set<TaskAttemptID>> trackerToTaskMap中删除在该TaskTracker上运行的所有Task。
    5. 对在该TaskTracker上的运行的每一个Task(在队列trackerToTaskMap中),进行如下2步处理:
    6. (1)从队列Map<TaskAttemptID, TaskInProgress> taskidToTIPMap中取出TaskAttemptID对应的TaskInProgress tip结构,再根据tip获取到JobInProgress:JobInProgress job = tip.getJob();;
    7. (2)如果ReduceTask已经完成,以及具有0个ReduceTask的所有MapTask已经完成,则将这些Task放入到队列TreeMap<String, Set<TaskAttemptID>> trackerToMarkedTasksMap中;如果tip标记Task没有完成,或者满足条件tip.isMapTask() && !tip.isJobSetupTask() && job.desiredReduces() != 0,检查Job运行状态,当job.getStatus().getRunState() == JobStatus.RUNNING || job.getStatus().getRunState() == JobStatus.PREP成立时,则该Task运行失败,并更新Task状态,同时收集这类Job,放入集合Set<JobInProgress> jobsWithFailures中,后续对这些Job进行处理;
    8. 由于该TaskTracker被JobTracker标记为lost状态,则对上面收集到的jobsWithFailures集合中的Job,只要存在属于该Job的Task被分配到该TaskTracker上运行,会通过累加计算在该TaskTracker上失败的Task计数,给该TaskTracker以惩罚,并释放所有在该TaskTracker上预留的Slot。
    9. 从队列TreeMap<String, Set<TaskAttemptID>> trackerToMarkedTasksMap中移除所有被标记完成的Task,同时更新JobTracker内部维护的如下3个队列:TreeMap<TaskAttemptID, String> taskidToTrackerMap、TreeMap<String, Set<TaskAttemptID>> trackerToTaskMap、Map<TaskAttemptID, TaskInProgress> taskidToTIPMap。
    10. 如果在该TaskTracker上的运行的Task还有没处理的,则转第6步进行处理;否则,流程结束。

    检查是否可以向TaskTracker指派运行Task

    当TaskTracker发送Heartbeat标志其没有重启,那么会执行该子流程,如下图所示:
    jt-hb-whether-assign-tasks
    在JobTracker端,既然TaskTracker汇报状态表明其没有重启,那么就需要检查该TaskTracker对应的黑名单和灰名单情况,如果TaskTracker状态一切正常,则恢复其正常被指派任务并运行Task的能力。

    标记TaskTracker为Health状态

    当TaskTracker重启了,然后再次连接JobTracker时,发送Heartbeat的过程中,会执行该流程。重启的TaskTracker,JobTracker会将一个TaskTracker标记为Health状态,说明该TaskTracker对应的资源信息(内存/CPU)应该在JobTracker端做记录,表示这些资源是可用的,更新JobTracker端的几个可用资源的变量计数。但是,很有可能TaskTracker重启之前,其上运行Task失败了很多次,在JobTracker端记录该失败计数,当满足一定条件后,会将TaskTracker加入灰名单,如果TaskTracker重启了,应该将其从灰名单中移除,以便不影响任务分派,具体处理流程如下图所示:
    jt-hb-mark-TaskTracker-health
    上述流程图比较简单不再累述。

    更新TaskTracker状态

    如果TaskTracker不是第一次连接JobTracker,那么在JobTracker端的队列HashMap<String, TaskTracker> taskTrackers中会保存上一次TaskTracker向JobTracker汇报的状态TaskTrackerStatus,如果该TaskTrackerStatus不存在,则直接处理当前汇报的TaskTracker的状态报告,使得JobTracker端维护的该TaskTracker的状态是最新的,具体的处理流程,如下图所示:
    jt-hb-update-TaskTracker-status
    上图中处理流程,描述如下所示:

    1. 检查是否JobTracker端存在该TaskTracker上一次汇报的状态报告,如果不存在,则直接处理当前发送的状态报告;否则,会更新JobTracker端维护的如下4个全局计数器:totalMaps(MapTask总数)、totalReduces(ReduceTask总数)、occupiedMapSlots(占用的Map Slot总数)、occupiedReduceSlots(占用的Reduce Slot总数),在当前计数值的基础上,减去上次汇报的报告中的数量(实际上是假定上次汇报的全部指标都已完成,如果没完成,再通过本次汇报的状态报告再加回去);如果TaskTracker没有被加入到黑名单中,还需要更新下面2个JobTracker端全局计数器:totalMapTaskCapacity(该TaskTracker上最大Map Slot总数)、totalReduceTaskCapacity(该TaskTracker上最大Reduce Slot总数)。
    2. 处理TaskTracker当前汇报的状态报告,更新JobTracker内部维护的6个全局计数器:totalMaps、totalReduces、occupiedMapSlots、occupiedReduceSlots、totalMapTaskCapacity、totalReduceTaskCapacity,各个计数器具体含义见上一步说明。
    3. 如果TaskTracker是第一次汇报状态报告,则需要在JobTracker内部注册,构造一个org.apache.hadoop.mapreduce.server.jobtracker.TaskTracker对象(该TaskTracker对象是在JobTracker的视角看到的结构),加入到队列HashMap<String, TaskTracker> taskTrackers中,同时还要计算该TaskTracker所在的host节点上TaskTracker进程的个数,更新队列Map<String, Integer> uniqueHostsMap。

    更新TaskTracker上所有Task状态

    在JobTracker处理TaskTracker发送的Heartbeat的过程中,首先会更新JobTracker维护的TaskTracker的状态信息,因为一个TaskTracker上可能运行着很多Task,那么需要更新这些Task的状态,可以通过上面介绍的TaskTrackerStatus的结构看出,对应着一个TaskStatus的状态报告集合,所以这里有一个批量更新TaskStatus状态的操作,实际上会对每一个Task的状态分别进行更新,整体处理流程如下图所示:
    jt-hb-update-all-tasks-on-TaskTracker
    具体处理流程,描述如下所示:

    1. 从TaskTracker发送的TaskTrackerStatus对象可以提取Task状态报告集合,然后对每一个状态报告进行处理,直到所有的Task的状态都已经被更新到JobTracker内部维护的状态对象上,下面描述每一个TaskStatus的处理过程
    2. (1)如果一个Task的运行状态不为TaskStatus.State.UNASSIGNED,说明该Task还没有在TaskTracker上获得运行机会,则并不让该Task失败(当一个Task指派给一个TaskTracker运行时,会首先在JobTracker端加入到一个超时列表中,由一个独立的线程JobTracker.ExpireLaunchingTasks去检测,该Task是否在给定的时间内(默认是10分钟 )是否在TaskTracker上启动而且一直没有报告状态,如果没有报告,则会将该Task标记为失败),等待下一次被调度分配给TaskTracker去运行。
    3. (2)根据Task的ID,获取到它对应的JobInProgress信息,如果没有获取到则将该Task对应的JobInProgress对象加入到cleanup列表Map<String, Set<JobID>> trackerToJobsToCleanup中,直接返回继续处理下一个TaskStatus报告;如果能够获取到对应的JobInProgress信息,则检查该JobInProgress中包含的Job是否设置初始化完成状态,如果没有设置,则直接将该Task加入到队列Map<String, Set<TaskAttemptID>> trackerToTasksToCleanup中,等待JobTracker调度Kill掉该Task,直接返回继续处理下一个TaskStatus报告。
    4. (3)检查该TaskStatus报告中对应的TaskAttemptID,是否在JobTracker端存在对应的TaskInProgress对象,很有可能JobTracker重启,内存中维护的Map<TaskAttemptID, TaskInProgress> taskidToTIPMap队列中没有TaskInProgress对象,这时JobInProgress对象一定存在,可以通过JobInProgress对象获取到该Task对应的TaskInProgress对象(因为在JobTracker端创建Job的时候,会分别创建4类TIP:map、reduce、cleanup、setup),再将其加入到Map<TaskAttemptID, TaskInProgress> taskidToTIPMap队列中,同时触发已知的一组JobInProgressListener的jobUpdated方法,去更新Job状态。
    5. (4)根据TaskStatus能够获取到所有Fetch失败的Task,查询该Task对应的TaskInProgress对象,从而进一步通知JobInProgress对象,根据设定的允许Task Fetch失败的最大次数限制,确定是否要让该Task失败,并更新TaskInProgress状态。

    更新Task状态

    当Task的状态发生变化的情况下,可能需要更新Task的状态,我们根据JobTracker定义的updateTaskStatus方法,方法声明如下所示:

    public synchronized void updateTaskStatus(TaskInProgress tip, TaskStatus status)
    

    其中,tip是当前在JobTracker端维护的Task的状态,status是TaskTracker汇报的Task状态,更新JobTracker端Task状态主要是根据心跳汇报的status来更新tip数据结构。更新Task状态的具体流程,如下图所示:
    jt-hb-update-task-status
    更新Task状态,主要是更新每个Task对应的在JobTracker端维护的TaskInProgress结构,处理流程描述如下:

    1. 如果心跳汇报的status中,Task运行状态为SUCCEEDED,当tip标识已经完成或标识被Kill掉,则统一修改status的运行状态为KILLED;如果心跳汇报的status对应的TaskAttemptID不是cleanup task,当该TaskAttemptID 对应的JobInProgress表示Job已经完成,或失败,或被Kill掉,那么status运行状态为FAILED_UNCLEAN则修改为FAILED,运行状态为KILLED_UNCLEAN则修改为KILLED。
    2. 调用TaskInProgress的updateStatus方法,传入当前TaskTracker汇报的status状态对象,更新tip的状态。TaskInProgress会维护每个Task对应的TaskStatus对象oldStatus,并根据汇报的status对更新替换oldStatus。有3种情况不需要更新:第一种是当status的运行状态不等于RUNNING/COMMIT_PENDING/FAILED_UNCLEAN/KILLED_UNCLEAN/UNASSIGNED中的任何一种状态;第二种是status的运行状态为RUNNING、UNASSIGNED中的任意一种状态,并且oldStatus的运行状态为FAILED/KILLED/FAILED_UNCLEAN/KILLED_UNCLEAN/SUCCEEDED/COMMIT_PENDING中任意一种状态;第三种是oldStatus的运行状态为FAILED/KILLED中的任意一种状态,这种情况会把该TaskAttemptID加入到队列TreeMap<TaskAttemptID, Boolean> tasksToKill中标识需要Kill掉该Task。
    3. 如果status的运行状态为FAILED状态,并且JobTracker在Safe模式下,则设置status的运行状态为KILLED。
    4. 此时,如果oldStatus与status不相等,即TaskAttemptID的状态已经发生变化,则会根据status的运行状态创建不同的TaskCompletionEvent事件(SUCCEEDED/FAILED/KILLED),这些 TaskCompletionEvent事件会被加入到JobInProgress的taskCompletionEvents列表中,供JobClient查询或供JobTracker检索;或者执行相应的操作:如果运行状态为FAILED_UNCLEAN/KILLED_UNCLEAN,则tip中该TaskAttemptID标记为失败并更新相关结构,然后加入到mapCleanupTasks/reduceCleanupTasks列表中等待被清理,同时将该TaskAttemptID对应的数据从JobTracker的taskidToTIPMap、taskidToTrackerMap、trackerToTaskMap这3个队列中删除。
    5. 根据构造的TaskCompletionEvent对象,并且如果status的运行状态为SUCCEEDED,则更新其对应的JobInProgress的状态为成功。
    Creative Commons License

    本文基于署名-非商业性使用-相同方式共享 4.0许可协议发布,欢迎转载、使用、重新发布,但务必保留文章署名时延军(包含链接:http://shiyanjun.cn),不得用于商业目的,基于本文修改后的作品务必以相同的许可发布。如有任何疑问,请与我联系

评论(5): “MapReduce V1:JobTracker处理Heartbeat流程分析

  1. 你好,博主,看了你的博客,给我感觉你的自我学习能力很强,有什么好的学习方法么,在短短的两年时间接触这么多东西?希望能回我邮件.谢谢.油箱:42666583#qq.com

发表评论

电子邮件地址不会被公开。 必填项已用*标注

您可以使用这些HTML标签和属性: <a href="" title=""> <abbr title=""> <acronym title=""> <b> <blockquote cite=""> <cite> <code> <del datetime=""> <em> <i> <q cite=""> <strike> <strong>