YARN之MRAppMaster深入剖析—ContainerAllocator分析

1. ContainerAllocator概述

ContainerAllocator负责与ResourceManager通信,为作业申请资源。作业的每个任务资源需求可描述为四元组,分别表示作业优先级、期望资源所在的host,资源量(当前仅支持内存),container数目,比如:

<10, “node1”, “memory:1G”, 3>//优先级是一个正整数,优先级值越小,优先级越高
<10, “node2”, “memory:2G”, 10>
<2, “*”, “memory:1G”, 20> //*表示这样的资源可来自任意一个节点,即不考虑数据本地性

ContainerAllocator周期性通过心跳与ResourceManager通信,ResourceManager每次会返回已经分配的container列表,完成的container列表等信息。

2. ContainerAllocator工作流程

当用户提交作业之后,MRAppMaster会为之初始化,并创建一系列map task和reduce task,由于reduce task依赖于map task之间结果,所以reduce task会延后调度。在ContainerAllocator中,当map task数目完成一定比例(由mapreduce.job.reduce.slowstart.completedmaps指定,默认是0.05,即5%)且Reduce Task可允许占用的资源(Reduce Task可占用资源比由yarn.app.mapreduce.am.job.reduce.rampup.limit指定)能够折合成整数个任务时,才会调度Reduce Task。

考虑到Map Task和Reduce Task之间的依赖关系,因此,它们之间的数据结构转移也是不一样的,对于Map Task而言,会依次转移到以下几个数据结构中:

scheduled->assigned->completed

对于Reduce Task而言,则按照以下流程:

pending->scheduled->assigned->completed

其中,pengding表示等待ContainerAllocator发送资源请求,scheduled表示已经将资源请求发送给RM,但还没有收到分配的资源,assigned是已经收到RM分配的资源,completed已经未完成。

Reduce Task之所有多出一个pending,主要是为了根据Map Task情况调整Reduce Task状态(在pengding和scheduled中相互转移)。进一步说,这主要是为了防止Map Task饿死,因为在YARN中不再有map slot和reduce slot的概念(这两个概念从一定程度上减少了作业饿死的可能性),只有内存、CPU等真实的资源,需要由ApplicationMaster控制资源申请的顺序,以防止可能产生的作业饿死。
此外,ContainerAllocator将所有任务划分成三类,分别是failed Map、Map和Reduce,并分别赋予它们优先级5、20和10,也就是说,当三种任务同时有资源需求是,会优先分配给failed map,然后是reduce,最后是map。

总结起来,ContainerAllocator工作流程如下:

步骤1 将所有map task的资源需求一次性发送给RM

步骤2 如果达到了Reduce task调度条件,则开始为Reduce Task申请资源。

步骤3 如果为某个task申请到了资源,则取消其他重复资源的申请。由于在HDFS中,任何一个任务
通常有三备份,而对于一个任务而言,考虑到rack和any级别的本地性,它可能会对应7个资源请求,分别是:

<20, “node1”, “memory:1G”, 1>
<20, “node2”, “memory:1G”, 1>
<20, “node3”, “memory:1G”, 1>
<20, “rack1”, “memory:1G”, 1>
<20, “rack2”, “memory:1G”, 1>
<20, “rack3”, “memory:1G”, 1>
<20, “*”, “memory:1G”, 1>

一旦该任务获取了以上任何一种资源,则会取消其他6个的资源申请。

步骤4 如果任务运行失败,则会重新为该任务申请资源。

步骤5 如果一个任务运行速度过慢,则会为其额外申请资源以启动备份任务(如果启动了推测执行功能)。

步骤6 如果一个节点失败的任务数目过多,则会撤销对该节点的所有资源申请请求。

3.ContainerAllocator类图

ContainerAllocator实际上是一接口,它只定义了三个事件:CONTAINER_REQ,CONTAINER_DEALLOCATE和CONTAINER_FAILED,分别表示请求container,释放container和container运行失败。

ContainerAllocator的实现是RMContainerAllocator,它只接收和处理ContainerAllocator接口中定义的三种事件,它的运行是这三种事件驱动的。

RMContainerAllocator中最核心的框架是维护了一个心跳信息,在RMCommunicator类中实现如下:

while (!stopped.get() && !Thread.currentThread().isInterrupted()) {

   try {

     Thread.sleep(rmPollInterval);

     try {

       heartbeat();

     } catch (YarnException e) {

     LOG.error("Error communicating with RM: " + e.getMessage() , e);

     return;

 } catch (Exception e) {

   LOG.error("ERROR IN CONTACTING RM. ", e);

 }

 } catch (InterruptedException e) {

   LOG.warn("Allocated thread interrupted. Returning.");

   return;

  }

}

其中,heartbeat()函数定义(在RMContainerAllocator类中)如下:

protected synchronized void heartbeat() throws Exception {

  LOG.info("Before Scheduling: " + getStat());

  List<Container> allocatedContainers = getResources();

  LOG.info("After Scheduling: " + getStat());

  if (allocatedContainers.size() > 0) {

    LOG.info("Before Assign: " + getStat());

    scheduledRequests.assign(allocatedContainers);

    LOG.info("After Assign: " + getStat());

  }

  ……

}

其中,getResources()函数用于向RM发送心跳信息,并处理心跳应答。需要注意的是,有些情况下,心跳信息中并不包含新的资源请求信息,即空的心跳信息,这有以下几个作用:

(1)周期性发送心跳,告诉RM自己还活着。

(2)周期性询问RM,以获取新分配的资源和各个container运行状况。

assign()函数是将收到的container分配给某个任务,如果这个container无法分配下去(比如内存空间不够),则是在下次心跳中通知RM释放该container,如果container可以分下去,则会释放对应任务的其他资源请求,同时会向TaskAttempt发送一个TA_ASSIGNED事件,以通知ContainerLauncher启动container。

坚持原创技术分享,您的支持将鼓励我继续创作!