hadoop之MapReduce框架TaskTracker端心跳机制分析(源码分析第六篇)

1、概述

   MapReduce框架中的master/slave心跳机制是整个集群运作的基础&＃xff0c;是沟通TaskTracker和JobTracker的桥梁。TaskTracker周期性地调用心跳RPC函数&＃xff0c;汇报节点和任务运行状态信息。MapReduce框架中通过心跳机制可以实现给TaskTracker分配任务、使JobTracker能够及时获取各个节点的资源使用情况和任务运行状态信息、判断TaskTracker的死活。本文主要从JobTracker和TaskTracker通信双方的角度分别去分析他们之间的心跳通信机制。

2、TaskTracker端心跳机制

    JobTracker和TaskTracker之前的心跳模式采取了“拉”的方式&＃xff0c;JobTracker不会主动向各个TaskTracker发送心跳信息&＃xff0c;而是各个TaskTracker主动向JobTracker发送信息&＃xff0c;同时领取JobTracker返回心跳包的各种命令。

   TaskTracker中有一个run方法&＃xff0c;其维护了一个无限循环用于通过心跳发送任务运行状态信息和接收JobTracker通过心跳返回的命令信息。其代码结构大概如下&＃xff1a;

/*** The server retry loop.* This while-loop attempts to connect to the JobTracker.  It only* loops when the old TaskTracker has gone bad (its state is* stale somehow) and we need to reinitialize everything.*/public void run() {try {getUserLogManager().start();startCleanupThreads();boolean denied &＃61; false;while (running && !shuttingDown && !denied) {boolean staleState &＃61; false;try {// This while-loop attempts reconnects if we get network errorswhile (running && !staleState && !shuttingDown && !denied) {try {State osState &＃61; offerService();if (osState &＃61;&＃61; State.STALE) {staleState &＃61; true;} else if (osState &＃61;&＃61; State.DENIED) {denied &＃61; true;}
....}

其中&＃xff0c;用于处理心跳相关信息的服务函数offerService代码大体框架&＃xff1a;

/*** Main service loop.  Will stay in this loop forever.*/State offerService() throws Exception {long lastHeartbeat &＃61; System.currentTimeMillis();//上一次发心跳距现在时间此循环主要根据控制完成task个数控制心跳间隔。while (running && !shuttingDown) {try {long now &＃61; System.currentTimeMillis();//获得当前时间// accelerate to account for multiple finished tasks up-front//通过完成的任务数动态控制心跳间隔时间long remaining &＃61;(lastHeartbeat &＃43; getHeartbeatInterval(finishedCount.get())) - now;while (remaining > 0) {// sleeps for the wait time or// until there are *enough* empty slots to schedule taskssynchronized (finishedCount) {finishedCount.wait(remaining);// Recomputenow &＃61; System.currentTimeMillis();remaining &＃61;(lastHeartbeat &＃43; getHeartbeatInterval(finishedCount.get())) - now;if (remaining <&＃61; 0) {// Reset countfinishedCount.set(0);break;}}}
...
//发送心跳
// Send the heartbeat and process the jobtracker&＃39;s directivesHeartbeatResponse heartbeatResponse &＃61; transmitHeartBeat(now);//真正想JobTracker发送心跳
.....
//开始处理JobTracker返回的命令
TaskTrackerAction[] actions &＃61; heartbeatResponse.getActions();
...
//杀死一定时间没没有汇报进度的task
markUnresponsiveTasks();
//当剩余磁盘空间小于mapred.local.dir.minspacekill(默认为0)时&＃xff0c;寻找合适的任务将其杀掉以释放空间
killOverflowingTasks();

从整个源码看&＃xff0c;TaskTracker向JobTracker发送一次心跳的流程如下&＃xff1a;

下面描述一下上面流程图中的几个重要过程&＃xff1a;

&＃xff08;1&＃xff09;过程一&＃xff0c;判断是否达到心跳间隔。

 TaskTracker的心跳间隔是由task完成情况以及整个集群规模规模动态觉得的。

• Task完成情况完成对心跳的动态调

   为了提高系统资源的利用效率和任务的相应的时间&＃xff0c;MapReduce框架提供了一种基于已经运行完毕的任务数的机制用于动态地缩短TaskTracker的发送心跳间隔&＃xff0c;从源码看&＃xff0c;这种机制叫做“outOfBand”。当存在某个Task运行完成或者失败&＃xff0c;TaskTracker会马上缩短心跳间隔以更快的速度将Task运行完成或者失败的消息告诉JobTracker&＃xff0c;让其重新快速分配任务。具体的实现机制我们来看源码分析&＃xff1a;

  源码定位到TaskTracker中的offerService方法

State offerService() throws Exception {//上一次发心跳距现在时间long lastHeartbeat &＃61; System.currentTimeMillis();//此循环主要根据控制完成task个数控制心跳间隔。while (running && !shuttingDown) {try {long now &＃61; System.currentTimeMillis();//获得当前时间// accelerate to account for multiple finished tasks up-front//通过完成的任务数动态控制心跳间隔时间long remaining &＃61;(lastHeartbeat &＃43; getHeartbeatInterval(finishedCount.get())) - now;while (remaining > 0) {// sleeps for the wait time or// until there are *enough* empty slots to schedule taskssynchronized (finishedCount) {finishedCount.wait(remaining);// Recomputenow &＃61; System.currentTimeMillis();remaining &＃61;(lastHeartbeat &＃43; getHeartbeatInterval(finishedCount.get())) - now;if (remaining <&＃61; 0) {// Reset countfinishedCount.set(0);//将已经完成的Task个数计数器归零....

上面代码的第九行&＃xff0c;就是用来实现根据Task的运行完成或者失败数目来动态的缩短心跳间隔。其中finishedCount.get()用于获得获得已经运行完毕的Task的计数。再来看看这个计数是怎么incream的&＃xff0c;定位到TaskTracker类的notifyTTAboutTaskCompletion方法&＃xff1a;

/*** Notify the tasktracker to send an out-of-band heartbeat.*/private void notifyTTAboutTaskCompletion() {if (oobHeartbeatOnTaskCompletion) {//判断是否启动“外带心跳”配置&＃xff08;默认为false&＃xff09;synchronized (finishedCount) {finishedCount.incrementAndGet();//运行完毕的Task计数器自增finishedCount.notify();}}}

其中oobHeartbeatOnTaskCompletion可以由mapreduce.tasktracker.outofband.heartbeat配置&＃xff08;默认为false&＃xff09;&＃xff0c;也就是说要当启动“外带心跳”时&＃xff0c;才会启动根据Task完成或者失败数来动态调整心跳间隔机制。下面看看动态调整心跳的具体算法&＃xff0c;进入getHeartbeatInterval(finishedCount.get())方法&＃xff1a;

private long getHeartbeatInterval(int numFinishedTasks) {return (heartbeatInterval / (numFinishedTasks * oobHeartbeatDamper &＃43; 1));}

其中&＃xff0c;numFinishedTasks代码已经运行完成或者失败的Task数目&＃xff0c;oobHeartbeatDamper简称“心跳收缩因子”由mapreduce.tasktracker.outofband.heartbeat.damper配置&＃xff08;默认为1000000&＃xff09;。当    启动外带心跳机制时&＃xff0c;如果某个时刻有numFinishedTasks个任务运行完成&＃xff0c;则心跳间隔就会调整为(heartbeatInterval / (numFinishedTasks * oobHeartbeatDamper &＃43; 1))。当不启动“外带心跳”机制时numFinishedTasks默认就为0了&＃xff0c;那么整个心跳间隔还是heartbeatInterval。

• 集群规模调整动态调整心跳间隔

  JobTracker有种机制&＃xff0c;能够根据集群中TaskTracker数量&＃xff0c;动态地调整TaskTracker的心跳间隔&＃xff0c;并将下一次心跳间隔放到TaskTracker的本次心跳应答中。具体分析将下一篇blog&＃xff0c;分析JobTracker端的心跳机制的时候会进行分析。

&＃xff08;2&＃xff09;过程二&＃xff0c;判断TaskTracker是否第一次启动

 当到达心跳间隔后发送心跳前&＃xff0c;会判断TaskTracker是否是第一次启动&＃xff0c;如果是第一次启动的话则会检测当前的TaskTracker版本是否和JobTracker的版本是否一致&＃xff0c;如果版本号一致才会向JobTracker发送心跳。看看源代码,还是TaskTracker类&＃xff1a;

......if(justInited) {//第一次启动justInited默认为trueString jobTrackerBV &＃61; jobClient.getBuildVersion();//获得JobTracker的版本号if(!VersionInfo.getBuildVersion().equals(jobTrackerBV)) {//获得TaskTracker的版本号&＃xff0c;并且判断JobTracker和TaskTracker的版本是否一致String msg &＃61; "Shutting down. Incompatible buildVersion." &＃43;"\nJobTracker&＃39;s: " &＃43; jobTrackerBV &＃43;"\nTaskTracker&＃39;s: "&＃43; VersionInfo.getBuildVersion();
....
justInited &＃61; false;//TaskTracker启动后将其设置为false
......

justInited默认为true&＃xff0c;当TaskTracker初次启动后会被改为false。当TaskTracker初次启动&＃xff0c;进入检测TaskTracker和JobTracker版本一致性环节。跟进代码中看看是如何判断版本一致性的&＃xff0c;

/*** Returns the buildVersion which includes version,* revision, user and date.*/public static String getBuildVersion(){return VersionInfo.getVersion() &＃43;" from " &＃43; VersionInfo.getRevision() &＃43;" by " &＃43; VersionInfo.getUser() &＃43;" source checksum " &＃43; VersionInfo.getSrcChecksum();}

上面代码是获得JobTracker和TaskTracker版本号的返回格式字符串&＃xff0c;getVersion()返回Hadoop版本号&＃xff0c;getRevision()返回Hadoop的修订版本号&＃xff0c;getUser()返回代码编译用户&＃xff0c;getSrcChecksum()返回校验和。验证版本的一致性就是验证这些。

&＃xff08;3&＃xff09;过程三&＃xff0c;检测磁盘是否读写是否正常。

 MapReduce框架中&＃xff0c;在map任务计算过程中会将输出结果保存在mapred.local.dir指定的本地目录中&＃xff08;可以由多块磁盘组成&＃xff0c;配置的时候用逗号隔开&＃xff09;&＃xff0c;这些本地目录是没有备份的&＃xff08;不像HDFS上有副本&＃xff09;一旦丢失或者损害整个Map任务需要重新进行计算。TaskTracker初始化时会对这些目录进行一次检测&＃xff0c;并将正常的目录保存起来。之后&＃xff0c;TaskTracker会周期性(由mapred.disk.healthChecker.interval配置&＃xff0c;默认60s)地对这些正常目录进行检测&＃xff0c;如果发现故障目录&＃xff0c;TaskTracker就会重新对自己进行初始化。看看源代码&＃xff0c;定位到TaskTracker的offerService方法&＃xff1a;

......now &＃61; System.currentTimeMillis();if (now > (lastCheckDirsTime &＃43; diskHealthCheckInterval)) {//判断是否达到检测磁盘的时间间隔localStorage.checkDirs();//检测硬盘读写是否正常lastCheckDirsTime &＃61; now;int numFailures &＃61; localStorage.numFailures();//出现读写错误的目录数// Re-init the task tracker if there were any new failuresif (numFailures > lastNumFailures) {//检测本次检测中是否存在损害目录lastNumFailures &＃61; numFailures;return State.STALE;//硬盘读写检测错误&＃xff0c;返回需要从新初始化状态}}
......

其中&＃xff0c;diskHealthCheckInterval代表检测磁盘的时间间隔&＃xff0c;由mapred.disk.healthChecker.interval配置&＃xff0c;默认60s。

&＃xff08;4&＃xff09;过程四&＃xff0c;发送心跳。

TaskTracker将当前节点运行时信息&＃xff0c;例如TaskTracker基本情况、资源使用情况、任务运行状态等&＃xff0c;通过心跳信息向JobTracker进行汇报&＃xff0c;同时接受来自JobTracker的各种指令。

// Send the heartbeat and process the jobtracker&＃39;s directivesHeartbeatResponse heartbeatResponse &＃61; transmitHeartBeat(now);//真正向JobTracker发送心跳

TaskTracker基本情况、资源使用情况、任务运行状态等信息会被封装到一个可序列化的类TaskTrackerStatus中&＃xff0c;并会伴随心跳发送给JobTracker。每次发送心跳时&＃xff0c;TaskTracker根据最新的信息重新构造TaskTrackerStatus。但是从源代码看并不是每次心跳都会发送节点资源信息申请新任务&＃xff0c;看代码&＃xff1a;

// Check if we should ask for a new Task//boolean askForNewTask;long localMinSpaceStart;//存在空闲的map或者reduce slot&＃xff0c;并且map输出目录大于mapred.local.dir.minspackekill才去向JobTracker发送节点资源使用情况申请新任务。synchronized (this) {askForNewTask &＃61;((status.countOccupiedMapSlots() < maxMapSlots ||status.countOccupiedReduceSlots() < maxReduceSlots) &&acceptNewTasks);localMinSpaceStart &＃61; minSpaceStart;}if (askForNewTask) {askForNewTask &＃61; enoughFreeSpace(localMinSpaceStart);//判断map中间结果输出路径空间long freeDiskSpace &＃61; getFreeSpace();//获得map中间结果输出大小long totVmem &＃61; getTotalVirtualMemoryOnTT();//获得虚拟内存总量long totPmem &＃61; getTotalPhysicalMemoryOnTT();//获得物理内存总量long availableVmem &＃61; getAvailableVirtualMemoryOnTT();//获得可用虚拟内存量long availablePmem &＃61; getAvailablePhysicalMemoryOnTT();//获得可用物理内存量long cumuCpuTime &＃61; getCumulativeCpuTimeOnTT();//获得TaskTracker自从集群启动到现在的累计使用时间long cpuFreq &＃61; getCpuFrequencyOnTT();//获得cpu频率int numCpu &＃61; getNumProcessorsOnTT();//获得cpu核心数float cpuUsage &＃61; getCpuUsageOnTT();//获得cpu使用率//将这些资源信息封装到TaskTracker中的resStatus对象(ResourceStatus类实例)status.getResourceStatus().setAvailableSpace(freeDiskSpace);status.getResourceStatus().setTotalVirtualMemory(totVmem);status.getResourceStatus().setTotalPhysicalMemory(totPmem);status.getResourceStatus().setMapSlotMemorySizeOnTT(mapSlotMemorySizeOnTT);status.getResourceStatus().setReduceSlotMemorySizeOnTT(reduceSlotSizeMemoryOnTT);status.getResourceStatus().setAvailableVirtualMemory(availableVmem);status.getResourceStatus().setAvailablePhysicalMemory(availablePmem);status.getResourceStatus().setCumulativeCpuTime(cumuCpuTime);status.getResourceStatus().setCpuFrequency(cpuFreq);status.getResourceStatus().setNumProcessors(numCpu);status.getResourceStatus().setCpuUsage(cpuUsage);}

只有当存在空闲的map或者reduce slot&＃xff0c;并且map输出目录大于mapred.local.dir.minspackekill才会将上面的节点资源信息放到TaskTrackerStatus中,向JobTracker发送节点资源使用情况申请新任务。再来看看通过心跳传给JobTracker的TaskTrackerStatus封装的具信息&＃xff1a;

String trackerName;//taskTracker名称String host;//主机名int httpPort;//TaskTracker对外的http端口int failures;//TaskTracker上运行失败的任务总数List taskReports;//记录了当前TaskTracker上各个任务的运行状态volatile long lastSeen;//上次汇报心跳的时间private int maxMapTasks;//map slot总数private int maxReduceTasks;//reduce slot总数private TaskTrackerHealthStatus healthStatus;//记录TaskTracker的健康情况
...private ResourceStatus resStatus;TaskTracker资源信息&＃xff0c;包括cpu、虚拟内存、物理内存等信息

以下主要说说healthStatus变量&＃xff0c;以及针对这个数据结构的节点健康监测机制。(注意&＃xff1a;该部分来自参考文献[1],p176~p177&＃xff09;

 healthStatus保存了节点的健康情况&＃xff0c;该变量对应TaskTrackerHealthStatus类&＃xff0c;结构如下&＃xff1a;

static class TaskTrackerHealthStatus implements Writable {private boolean isNodeHealthy;//节点是否健康private String healthReport;//如果节点不健康&＃xff0c;则记录导致不健康的原因private long lastReported;//上一次汇报健康状态的时间
...}

 healthStatus是由NodeHealthCheckerService线程计算得到&＃xff0c;该线程允许管理员配置一个“监控监测脚本”以监测节点健康状况&＃xff0c;且管理员可以在该脚本中添加任务监测语句作为节点是否健康运行的依据。如果脚本检测到该节点处于不健康状态&＃xff0c;它需要再标准输出中打印一条以字符串“ERROR”开头的输出语句。NodeHealthCheckerService线程周期性调用健康检测脚本并检测其输出&＃xff0c;一旦发现脚本输出是以“ERROR”开头的字符串&＃xff0c;则认为该节点不健康&＃xff0c;进而将其标注为“unhealthy”并通过心跳告诉JobTracker&＃xff0c;而JobTracker得知节点状态为“unhealthy”后&＃xff0c;会将其加入黑名单&＃xff0c;此后不再为它分配新任务。需要注意的是&＃xff0c;只要TaskTracker服务是活的&＃xff0c;该线程就会一直运行该脚本&＃xff0c;一旦发现节点又变为“healthy”&＃xff0c;JobTracker会立刻将其从黑名单中移除&＃xff0c;从而又会为之分配新任务。通过引入该机制&＃xff0c;可以带来很多好处&＃xff1a;

• 可作为节点负载的反馈&＃xff1a;比如&＃xff0c;可让健康检测脚本检测网络、磁盘、文件系统等运行情况&＃xff0c;一旦发现特殊情况&＃xff0c;比如网络拥塞、磁盘空间不足或者文件系统出现问题&＃xff0c;可以将健康状态变为“unhealthy”&＃xff0c;暂时不接收新的任务&＃xff0c;等待它们恢复正常后再继续接收新任务。

• 人为暂时维护TaskTracker&＃xff1a;如果发现TaskTracker所在节点出现故障&＃xff0c;可以通过控制脚本输出暂时让该TaskTracker停止接收新任务以便进行维护&＃xff0c;等待维护完成后&＃xff0c;修改脚本输出以让TaskTracker继续正常接收任务。

NodeHealthCheckerService线程包含了4个可配置参数&＃xff0c;用户可以再mapred-site.xml中记性配置&＃xff1a;

(1)mapred.healthChecker.script.path&＃xff1a;健康检测脚本所在的绝对路径&＃xff0c;NodeHealthCheckerService会周期性执行该脚本以判断节点健康状态&＃xff0c;如果该值为空&＃xff0c;则不会启动该线程。

(2)mapred.healthChecker.interval&＃xff1a;健康检测脚本调用频率&＃xff08;单位&＃xff1a;ms&＃xff09;.

(3)mapred.healthChecker.script.timeout&＃xff1a;如果检测脚本在一定时间内没有响应&＃xff0c;则NodeHealthCheckerService线程会将该节点的监控状态标注为“unhealthy”。

(4)mapred.healthChecker.script.args&＃xff1a;监控脚本的输入参数&＃xff0c;如果有多个参数&＃xff0c;则用逗号分开。

下面是一个健康监控脚本实例。在shell脚本中&＃xff0c;当一个节点上的空闲内存量低于10%时&＃xff0c;打印“ERROR”开头的字符串&＃xff0c;这样该节点将不再向JobTracker请求信任务。

#!/bin/bash
MEMORY_RATIO&＃61;0.1
freeMem&＃61;&＃96;grep MemFree /proc/meminfo |awk &＃39;{print $2}&＃39;&＃96;
totalMem&＃61;&＃96;grep MemTotal /proc/meminfo | awk &＃39;{print $2}&＃39;&＃96;
limitMem&＃61;&＃96;echo | awk &＃39;{print int("&＃39;$totalMem&＃39;"*"&＃39;$MEMORY_RATIO&＃39;")}&＃39;&＃96;
if [$freeMem -lt $limitMem];thenecho "ERROR,totalMem&＃61;$totalMem, freeMem&＃61;$freeMem, limitMem&＃61;$limitMem"
elseecho "Ok,totalMem&＃61;$totalMem, freeMem&＃61;$freeMem, limitMem&＃61;$limitMem"
fi

&＃xff08;5&＃xff09;过程五&＃xff0c;发送心跳。

  接收并执行JobTracker通过心跳返回的指令,详细流程看上面的流程图。

3、JobTracker端心跳机制

  由于本blog篇幅有点长&＃xff0c;JobTracker端心跳机制在下一篇blog中续写..........

参考文献&＃xff1a;

[1]《Hadoop技术内幕&＃xff1a;深入解析MapReduce架构设计与实现原理》

[2] http://hadoop.apache.org