首页 > 网络 > 云计算 >

Hadoop心跳机制

2016-10-01

心跳是Jobtracker和Tasktracker的桥梁,它实际上是一个RPC函数,Tasktracker周期性的调用该函数汇报节点和任务状态信息,从而形成心跳。

心跳是Jobtracker和Tasktracker的桥梁,它实际上是一个RPC函数,Tasktracker周期性的调用该函数汇报节点和任务状态信息,从而形成心跳。在hadoop中,心跳主要有三个作用:
1、判断Tasktracker是否活着
2、及时让Jobtracker获取各个节点上的资源使用情况和任务运行状态
3、为Tasktracker分配任务
注意:Jobtracker与Tasktracker之间采用了Pull而不是Push模型,是Jobtracker不会主动向Tasktracker发送任何信息,而是由Tasktracker主动通过心跳领取属于自己的信息,Jobtracker只能通过心跳应答的形式为各个Tasktracker分配任务。

Tasktracker周期性的调用RPC函数heartbeat向Jobtracker汇报信息和领取任务,这个函数定义是:

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

这些参数含义是
Status:该类参数封装了Tasktracker上的各种状态信息,包括
String trackerName //TaskTracker名称,形式如 tracker_mymachine:localhost.localdomain/127.0.0.1:34196
String host //TaskTracker主机名
int httpPort //TaskTracker对外的HTTP端口号
int failures //该TaskTracker上已经失败的任务数
List taskReports //正在运行的各个任务状态
volatile long lastSeen //上次汇报心跳的时间
private int maxMapTasks //Map slot总数,即允许同时运行的Map Task总数,有参数mapred.tasktracker.map.tasks.maximum设定
private int maxReduceTasks//Reduce slot 总数
private TaskTrackerHealthStatus healthStatus//Tasktracker健康状态
private ResourceStatus resStatus //Tasktracker 资源(内存 CPU等)信息
Restarted:表示Tasktracker是否刚刚重新启动
initialContact:表示TaskTracker是否初次连接Jobtracker
acceptNewTasks:表Tasktracker是否可以接收新任务,这通常取决于slot是否有剩余和节点健康状态等
responseId:表示心跳响应编号,用于防止重复发送心跳,没接收一次心跳后,该值加1
该函数的返回值为一个HeartbeatResponse 对象,该对象主要封装了Jobtracker向tasktracker下达的命令,

class HeartbeatResponse implements Writable, Configurable {
  Configuration conf = null;
  short responseId;//心跳响应编号
  int heartbeatInterval;//下次心跳发送间隔
  TaskTrackerAction[] actions;//来自Jobtracker的命令, 杀死作业,杀死任务,提交任务,运行任务。等等
  Set recoveredJobs = new HashSet();//恢复完成的作业列表
 .......}

该函数的内部实现逻辑主要分为两个步骤:更新状态和下达命令。Jobtracker首先将Tasktracker汇报的最新任务运行状态保存到相应的数据结构中,然后根据这些状态信息和外界需求为其下达相应的命令。
1、更新状态
函数heartbeat首先会更新Tasktracker/job/task的状态信息。代码如下

 if (!acceptTaskTracker(status)) {
      throw new DisallowedTaskTrackerException(status);
    }//检查是否允许Tasktracker连接Jobtracker,当一个Tasktracker在Host list(由参数mapred.hosts指定)中,但不在exclude list(由参数mapred.hosts.excluede指定)中时,可接入Jobtracker

    String trackerName = status.getTrackerName();
    long now = clock.getTime();
    if (restarted) {
      faultyTrackers.markTrackerHealthy(status.getHost());
    } else {
      faultyTrackers.checkTrackerFaultTimeout(status.getHost(), now);
    }//如果该Tasktracker被重启了,则将之标注为健康的Tasktracker,并从黑名单或者灰名单中清除,否则,启动Tasktracker容错机制以检查它是否处于健康状态
    ........
    // Process this heartbeat 
    short newResponseId = (short)(responseId + 1);
    status.setLastSeen(now);
    if (!processHeartbeat(status, initialContact, now)) {
     ........
    }//记录心跳发送时间,以发现一定时间内未发送心跳的Tasktracker,并将之标注为死亡的Tasktracker,此后不可再向其分配新任务
接下来跟踪进入函数processHeartbeat内部,该函数首先进行一系列异常情况检查,然后调用以下两个函数更新Tasktracker/job/task的状态信息
  updateTaskStatuses(trackerStatus);//更新Task状态信息
    updateNodeHealthStatus(trackerStatus, timeStamp);//更新节点健康状态

下达命令
在Jobtracker更新状态以后,Jobtracker要为Tasktracker构造一个HeartbeatResponse对象作为心跳应答,该对象主要有2个部分内容:下达给Tasktracker的命令和下次心跳的时间
下达命令
Jobtracker将下达给Tasktracker的命令封装成TasktrackerAction类,主要包括了ReinitTrackerAction(重新初始化)、LaunchTaskAction(运行新任务)、KillTaskAction(杀死任务)、KillJobAction(杀死作业)、CommitTaskAction(提交任务)五种。
ReinitTrackerAction:Jobtracker接到Tasktracker发送过来的心跳信息后,首先要进行一致性检查。如果发现异常情况,则会要求Tasktracker重新对自己进行初始化,已恢复到一致性的状态。当出现下面2中不一致情况时,Jobtracker会向Tasktracker下达ReinitTrackerAction命令
1、丢失上次心跳应答信息:Jobtracker会保存向每个Tasktracker发送的最近心跳应答信息,如果Jobtracker未刚刚重启且一个Tasktracker并非初次连接Jobtracker(initialContact!=true)而最近心跳应答丢失了,
2、丢失Tasktracker状态信息:jobTracker接收到任何一个心跳信息后,会将Tasktracker状态(封装在类TaskTrackerStatus中)信息保存起来,如果一个Tasktracker非初次连接Jobtracker但状态信息却不存在,

LaunchTaskAction:该类封装了Tasktracker新分配的任务,Tasktracker接收到该命令后会启动一个子进程运行该任务,Hadoop将一个作业分解后的任务分成2大类:计算型任务和辅助型任务,其中,计算型任务是处理实际数据的任务,包括Map Task和Reduce Task2种(对应的TaskType类中的MAP和REDUCE2种类型),由专门的任务调度器对它们进行调度,而辅助型任务则不会处理实际的数据,通常用户同步计算型任务或者清理磁盘上无用的目录,包括job-setup task、job-cleanup task、和task-cleanup task三种(对应的ActionType是JOB_SETUP JOB_CLEANUP TASK_CLEANUP),其中job-setup task 和job-cleanup task 分别用作计算型任务开始运行同步标识和结束运行同步标识,而task-cleanup task 则用于清理失败计算型任务已经写到磁盘的部分结果,这种任务由Jobtracker负责调度,运行优先级高于计算型任务。
如果一个正常的Tasktracker尚有空闲的slot(acceptNewTasks=true),则Jobtracker会为该Tasktracker分配新任务,任务顺序是先辅助型任务在计算型任务,选择顺序依次为job-cleanup task 、task-cleanup task 、job-setup task .
//优先选择辅助型任务,选择优先级从高到低依次是 job-cleanup task 、task-cleanup task 、job-setup task这样可以要完成了的作业快速结束,提交的的作业立刻进入运行状态

   List tasks = getSetupAndCleanupTasks(taskTrackerStatus);
        if (tasks == null ) {
          tasks = taskScheduler.assignTasks(taskTrackers.get(trackerName));
        }//如果没有辅助型任务,则选择计算型任务,由任务调度器选择一个或者多个计算型任务
        if (tasks != null) {
          for (Task task : tasks) {
            expireLaunchingTasks.addNewTask(task.getTaskID());
            if(LOG.isDebugEnabled()) {
              LOG.debug(trackerName + " -> LaunchTask: " + task.getTaskID());
            }/将分配的任务封装成LaunchTaskAction对象
            actions.add(new LaunchTaskAction(task));
          }
        }

KillTaskAction:该类封装了Tasktracker需杀死的任务,Tasktracker收到该命令后杀掉对应任务、清理工作目录和释放slot,导致Jobtracker向Tasktracker发送该命令的原因有很多主要:
1、用户使用 hadoop job -kill-task 或者 hadoop job -fail-task 杀死一个任务或者使一个任务失败
2、启用了推测执行机制后,同一份数据可能同时由2个task attempt 处理,当其中一个task attempt执行成功后,另一个处理相同数据的task attempt将会被杀掉
3、某个作业运行失败,它的其他任务将被杀掉
4、Tasktracker在一定时间内未汇报心跳,则Jobtracker认为其死掉,它上面所有task均标注为死亡

KillJobAction:该类封装了Tasktracker待清理的作业,tasktracker接收到该命令后,会清理作业的临时目录,导致Jobtracker向Tasktracker发送该命令的原因很多
1、用户使用 hadoop job -kill 或者hadoop job -fail杀死一个作业或者使一个作业失败
2、作业完成,通知Tasktracker清理该作业的工作目录
3、作业运行失败,即同一个作业失败的task数目超过一定比例

CommitTaskAction:该类封装了Tasktracker需要提交的任务。为了防止同一个TaskInProcess的2个同时运行的Task Attempt同时打开一个文件或者往一个文件中写数据而产生冲突,Hadoop让每个Task Attempt 写到单独一个文件中(以TaskAttemptID命名,比如attempt_201606092312_0008_r_000000_0)中,通常而言,hadoop让每个Task Attempt将计算结果写到临时目录{mapred.output.dir}/temporary/_{taskid}中,当某个task Attempt成功运行完成,再将运算结果转移到最终目录${mapred.output.dir}中,hadoop将一个成功运行完成的task Attempt结果文件从临时目录”提升”至最终目录的过程,称为“提交任务”。当TaskInProgress中一个任务被提交后,其他任务将会被杀死,同时意味着该TaskInProgress运行完成

调整心跳间隔
Tasktracker心跳时间间隔大小应该适度,如果太小,则Jobtracker需要处理高并发的心跳连接请求,必然产生不小的并发压力,如果太大,空闲的资源不能及时汇报给Jobtracker(进而个tasktracker分配任务),造成资源浪费,进而降低了系统的吞吐率。Tasktracker汇报心跳的时间间隔并不是一成不变的,它会随着集群的规模的动态调整而变化,以便能够合理利用Jobtracker的并发处理能力,在hadoop mapreduce中,只有Jobtracker知道某一时刻集群的规模,因此由Jobtracker为每个Tasktracker计算下一次汇报心跳的时间间隔,并通过心跳机制告诉Tasktracker。
Jobtracker允许用户通过参数配置心跳的时间间隔加速比,即每增加mapred.heartbeats.in.second(默认是100,最小1)个节点心跳时间间隔增加mapreduce.jobtracker.heartbeats.scaling.factor(默认是1,最小是0.01)秒。同时为了防止用户参数设置不合理而对Jobtracker产生较大的负载,Jobtracker要求心跳时间间隔至少为3秒

 public int getNextHeartbeatInterval() {
    // 获取当前Tasktracker的总数,
    int clusterSize = getClusterStatus().getTaskTrackers();
    //计算新的心跳间隔
    int heartbeatInterval =  Math.max((int)(1000 * HEARTBEATS_SCALING_FACTOR *((double)clusterSize /NUM_HEARTBEATS_IN_SECOND)), HEARTBEAT_INTERVAL_MIN) ;
    return heartbeatInterval;
  }
相关文章
最新文章
热点推荐