1. ContainerAllocator概述

ContainerAllocator负责与ResourceManager通信,为作业申请资源。作业的每个任务资源需求可描述为四元组<Priority, hostname,capability,containers>,分别表示作业优先级、期望资源所在的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发送心跳信息,并处理心跳应答。需要注意的是,有些情况下,心跳信息中并不包含新的资源请求信息,即空的心跳信息,这有以下几个作用:

  • 周期性发送心跳,告诉RM自己还活着。
  • 周期性询问RM,以获取新分配的资源和各个container运行状况。

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

原创文章,转载请注明: 转载自董的博客

本文链接地址: YARN/MRv2 MRAppMaster深入剖析—ContainerAllocator分析

微信公众号:hadoop-123,专注于大数据技术分享,欢迎加入!

说点什么

avatar
  Subscribe  
提醒