Hadoop源码系列（一）FairScheduler申请和分配container的过程

阅读量：6515 次

发布时间：2019-06-24

本文共 7625 字，大约阅读时间需要 25 分钟。

1、如何申请资源

1.1 如何启动AM并申请资源

1.1.1 如何启动AM

val yarnClient = YarnClient.createYarnClientsetupCredentials()yarnClient.init(yarnConf)yarnClient.start()// Get a new application from our RMval newApp = yarnClient.createApplication()val newAppResponse = newApp.getNewApplicationResponse()appId = newAppResponse.getApplicationId() // Set up the appropriate contexts to launch our AMval containerContext = createContainerLaunchContext(newAppResponse)val appContext = createApplicationSubmissionContext(newApp, containerContext) // Finally, submit and monitor the applicationlogInfo(s"Submitting application $appId to ResourceManager")yarnClient.submitApplication(appContext)

1.1.2 FairScheduler如何处理AM的ResourceRequest

1、FairScheduler接收到SchedulerEventType.APP_ADDED之后，调用addApplication方法把把RMApp添加到队列里面，结束之后发送RMAppEventType.APP_ACCEPTED给RMApp

2、RMApp启动RMAttempt之后，发送SchedulerEventType.APP_ATTEMPT_ADDED给FairScheduler

LOG.info("Added Application Attempt " + applicationAttemptId + " to scheduler from user: " + user);

3、FairScheduler调用addApplicationAttempt方法，发送RMAppAttemptEventType.ATTEMPT_ADDED事件给RMAppAttempt，RMAppAttempt随后调用Scheduler的allocate方法发送AM的ResourceRequest

4、FairScheduler在allocate方法里面对该请求进行处理，FairScheduler对于AM的资源请求的优先级上并没有特殊的照顾，详细请看章节2 如何分配资源

1.2 AM启动之后如何申请资源

1.2.1、注册AM

amClient = AMRMClient.createAMRMClient()amClient.init(conf)amClient.start()amClient.registerApplicationMaster(Utils.localHostName(), 0, uiAddress)

1.2.2、发送资源请求

// 1.创建资源请求amClient.addContainerRequest(request)// 2.发送资源请求val allocateResponse = amClient.allocate(progressIndicator)val allocatedContainers = allocateResponse.getAllocatedContainers()if (allocatedContainers.size > 0) {  // 3.请求返回之后处理Container  handleAllocatedContainers(allocatedContainers.asScala)}

1.2.3、启动Container

def startContainer(): java.util.Map[String, ByteBuffer] = { val ctx = Records.newRecord(classOf[ContainerLaunchContext])   .asInstanceOf[ContainerLaunchContext] val env = prepareEnvironment().asJava  ctx.setLocalResources(localResources.asJava) ctx.setEnvironment(env)  val credentials = UserGroupInformation.getCurrentUser().getCredentials() val dob = new DataOutputBuffer() credentials.writeTokenStorageToStream(dob) ctx.setTokens(ByteBuffer.wrap(dob.getData()))  val commands = prepareCommand()  ctx.setCommands(commands.asJava) ctx.setApplicationACLs(YarnSparkHadoopUtil.getApplicationAclsForYarn(securityMgr).asJava)  // If external shuffle service is enabled, register with the Yarn shuffle service already // started on the NodeManager and, if authentication is enabled, provide it with our secret // key for fetching shuffle files later if (sparkConf.get(SHUFFLE_SERVICE_ENABLED)) {   val secretString = securityMgr.getSecretKey()   val secretBytes =   if (secretString != null) {     // This conversion must match how the YarnShuffleService decodes our secret     JavaUtils.stringToBytes(secretString)   } else {     // Authentication is not enabled, so just provide dummy metadata     ByteBuffer.allocate(0)   }   ctx.setServiceData(Collections.singletonMap("spark_shuffle", secretBytes)) }  // Send the start request to the ContainerManager try {   nmClient.startContainer(container.get, ctx) } catch {   case ex: Exception =>     throw new SparkException(s"Exception while starting container ${container.get.getId}" +       s" on host $hostname", ex) }}

View Code

2、如何分配资源

2.1 接受资源请求步骤

在FairScheduler的allocate方法里面仅仅是记录ResourceRequest，并不会真正的立马分配。

流程如下：

1、检查该APP是否注册过

2、检查资源的请求是否超过最大内存和最大CPU的限制

3、记录资源请求的时间，最后container分配的延迟会体现在队列metrics的appAttemptFirstContainerAllocationDelay当中

4、释放AM发过来的已经不需要的资源，主要逻辑在FSAppAttempt的containerCompleted方法里

5、更新资源请求，所有资源请求都是记录在AppSchedulingInfo当中的requests（注意：只有是ANY的资源请求才会被立马更新到QueueMetrics的PendingResources里）

6、找出该APP被标记为抢占的container ID列表preemptionContainerIds

7、更新APP的黑名单列表，该信息被记录在AppSchedulingInfo当中

8、从FSAppAttempt的newlyAllocatedContainers当中获取最新被分配的container

9、返回preemptionContainerIds、HeadRoom、ContainerList、NMTokenList。(注：Headroom = Math.min(Math.min(queueFairShare - queueUsage, 0), maxAvailableResource)

2.2 请求和分配的关系

请求和分配的过程是异步的，关系如上图，每次调用allocate获得的container，其实是之前的请求被分配的结果

2.3 如何分配

2.3.1 分配方式

分配有两种方式：

1、接收到NodeManager的心跳的时候进行分配

NodeManager每隔一秒（yarn.resourcemanager.nodemanagers.heartbeat-interval-ms）给ResourceManager发送一个心跳事件NODE_UPDATE，接收到心跳事件之后，在FairScheduler的nodeUpdate方法里进行处理。

NodeManager会汇报新启动的Container列表newlyLaunchedContainers和已经结束的Container列表completedContainers。然后在attemptScheduling方法里面进行分配。

2、持续调度方式

它有一个单独的线程，线程名称是FairSchedulerContinuousScheduling，每5毫秒对所有节点的资源进行排序，然后遍历所有节点，调用attemptScheduling方法进行分配。

开启持续调度模式之后，在接收到心跳事件NODE_UPDATE的时候，只有在completedContainers不为空的情况下，才会进行调度

attemptScheduling首先会检查是否有资源预留，如果有预留，则直接为预留的APP分配container

没有预留的分配过程如下：

1、最大可分配资源为这台机器的可用资源的一半，从root队列开始自上而下进行分配Resource assignment = queueMgr.getRootQueue().assignContainer(node);

2、分配到一个Container之后，判断是否要连续分配多个，最大支持连续分配多少个？

以下是涉及到的各个参数以及参数的默认值：

yarn.scheduler.fair.assignmultiple false （建议设置为true）

yarn.scheduler.fair.dynamic.max.assign true （hadoop2.7之后就没有这个参数了）

yarn.scheduler.fair.max.assign -1 （建议设置为2~3，不要设置得太多，否则会有调度倾斜的问题）

2.3.2 如何从队列当中选出APP进行资源分配

入口在queueMgr.getRootQueue().assignContainer(node);

1、检查当前队列的使用量是否小于最大资源量

2、首先对子队列进行排序，优先顺序请参照章节

3、排序完再调用子队列的assignContainer方法分配container

4、一直递归到叶子队列

叶子队列如何进行分配？

1、先对runnableApps进行排序，排序完成之后，for循环遍历一下

2、先检查该Node是否在APP的黑名单当中

3、检查该队列是否可以运行该APP的AM，主要是检查是否超过了maxAMShare（根据amRunning字段判断是否已经启动了AM了）

检查逻辑的伪代码如下：

maxResource = getFairShare()if (maxResource == 0) {  // 最大资源是队列的MaxShare和集群总资源取一个小的值  maxResource = Math.min(getRootQueue().AvailableResource(), getMaxShare());}maxAMResource = maxResource * maxAMShareif (amResourceUsage + amResource) > maxAMResource) {  // 可以运行  return true} else {  // 不可以运行  return false}

View Code

4、给该APP分配container

下面以一个例子来说明分配的过程是如何选择队列的：

假设队列的结构是这样子的

root

---->BU_1

-------->A

-------->B

---->BU_2

-------->C

-------->D