分享

hadoop中YARN的ContainerAllocator分析

1. ContainerAllocator概述
ContainerAllocator负责与ResourceManager通信,为作业申请资源。作业的每个任务资源需求可描述为四元组<Priority, hostname,capability,containers>,分别表示作业优先级、期望资源所在的host,资源量(当前仅支持内存),container数目,比如:
  1. <10, “node1”, “memory:1G”, 3>//优先级是一个正整数,优先级值越小,优先级越高
  2. <10, “node2”, “memory:2G”, 10>
  3. <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而言,会依次转移到以下几个数据结构中:
  1. scheduled->assigned->completed
复制代码
对于Reduce Task而言,则按照以下流程:
  1. 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个资源请求,分别是:
  1. <20, “node1”, “memory:1G”, 1>
  2. <20, “node2”, “memory:1G”, 1>
  3. <20, “node3”, “memory:1G”, 1>
  4. <20, “rack1”, “memory:1G”, 1>
  5. <20, “rack2”, “memory:1G”, 1>
  6. <20, “rack3”, “memory:1G”, 1>
  7. <20, “*”, “memory:1G”, 1>
复制代码
一旦该任务获取了以上任何一种资源,则会取消其他6个的资源申请。
步骤4      如果任务运行失败,则会重新为该任务申请资源。
步骤5      如果一个任务运行速度过慢,则会为其额外申请资源以启动备份任务(如果启动了推测执行功能)。
步骤6      如果一个节点失败的任务数目过多,则会撤销对该节点的所有资源申请请求。

3. ContainerAllocator类图

ContainerAllocator实际上是一接口,它只定义了三个事件:CONTAINER_REQ,,CONTAINER_DEALLOCATE和CONTAINER_FAILED,分别表示请求container,释放container和container运行失败。
ContainerAllocator的实现是RMContainerAllocator,它只接收和处理ContainerAllocator接口中定义的三种事件,它的运行是这三种事件驱动的。

containerallocator.jpg



RMContainerAllocator中最核心的框架是维护了一个心跳信息,在RMCommunicator类中实现如下:
  1. while (!stopped.get() && !Thread.currentThread().isInterrupted()) {
  2.    try {
  3.      Thread.sleep(rmPollInterval);
  4.      try {
  5.        heartbeat();
  6.      } catch (YarnException e) {
  7.      LOG.error("Error communicating with RM: " + e.getMessage() , e);
  8.      return;
  9. } catch (Exception e) {
  10.    LOG.error("ERROR IN CONTACTING RM. ", e);
  11. }
  12. } catch (InterruptedException e) {
  13.    LOG.warn("Allocated thread interrupted. Returning.");
  14.    return;
  15.   }
  16. }
复制代码
其中,heartbeat()函数定义(在RMContainerAllocator类中)如下:
  1. protected synchronized void heartbeat() throws Exception {
  2.   LOG.info("Before Scheduling: " + getStat());
  3.   List<Container> allocatedContainers = getResources();
  4.   LOG.info("After Scheduling: " + getStat());
  5.   if (allocatedContainers.size() > 0) {
  6.     LOG.info("Before Assign: " + getStat());
  7.     scheduledRequests.assign(allocatedContainers);
  8.     LOG.info("After Assign: " + getStat());
  9.   }
  10.   ……
  11. }
复制代码
其中,getResources()函数用于向RM发送心跳信息,并处理心跳应答。需要注意的是,有些情况下,心跳信息中并不包含新的资源请求信息,即空的心跳信息,这有以下几个作用:
(1)周期性发送心跳,告诉RM自己还活着。
(2)周期性询问RM,以获取新分配的资源和各个container运行状况。

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


来自群组: Hadoop技术组

没找到任何评论,期待你打破沉寂

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条