Hadoop源码系列(一)FairScheduler申请和分配container的过程
1、如何申请资源
1.1 如何启动AM并申请资源
1.1.1 如何启动AM
val yarnClient = YarnClient.createYarnClient
setupCredentials()
yarnClient.init(yarnConf)
yarnClient.start()
// Get a new application from our RM
val newApp = yarnClient.createApplication()
val newAppResponse = newApp.getNewApplicationResponse()
appId = newAppResponse.getApplicationId()
// Set up the appropriate contexts to launch our AM
val containerContext = createContainerLaunchContext(newAppResponse)
val appContext = createApplicationSubmissionContext(newApp, containerContext)
// Finally, submit and monitor the application
logInfo(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)
}
}
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、首先对子队列进行排序,优先顺序请参照章节 2.3.4 如何确定优先顺序
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 * maxAMShare
if (amResourceUsage + amResource) > maxAMResource) {
// 可以运行
return true
} else {
// 不可以运行
return false
}
4、给该APP分配container
下面以一个例子来说明分配的过程是如何选择队列的:
假设队列的结构是这样子的
root
---->BU_1
-------->A
-------->B
---->BU_2
-------->C
-------->D
2.3.3 任务分配Container的本地性
任务分配Container的时候会考虑请求的本地性,对于调度器来说,它的本地性分为三种:NODE_LOCAL, RACK_LOCAL, OFF_SWITCH
具体方法位于FSAppAttempt的assignContainer方法
遍历优先级
给该优先级的调度机会+1
获取RackLocal和NodeLocal的任务
计算允许分配的本地性级别allowedLocality,默认是NODE_LOCAL
1、心跳分配方式
计算调度机会,如果该优先级的任务的调度机会超过了(节点数 * NODE_LOCAL阈值),降级为RACK_LOCAL,如果该优先级的任务的调度机会超过了(节点数 * RACK_LOCAL阈值),降级为OFF_SWITCH
2、连续分配方式
计算等待时间waitTime -= lastScheduledContainer.get(priority);
如果waitTime超过了NODE_LOCAL允许的delay时间,就降级为RACK_LOCAL,再超过RACK_LOCAL允许的delay的时间,就降级为OFF_SWITCH
分配NODE_LOCAL的container
允许分配的本地性级别>=RACK_LOCAL,分配RACK_LOCAL的container
允许分配的本地性级别=OFF_SWITCH,分配OFF_SWITCH的container
都分不到,等待下一次机会
相关参数:
默认值全是-1,则允许的本地性级别是OFF_SWITCH
yarn.scheduler.fair.locality-delay-node-ms -1
yarn.scheduler.fair.locality-delay-rack-ms -1
yarn.scheduler.fair.locality.threshold.node -1
yarn.scheduler.fair.locality.threshold.rack -1
2.3.4 Container分配
1、检查该节点的资源是否足够,如果资源充足
2、如果当前的allowedLocality比实际分配的本地性低,则重置allowedLocality
3、把新分配的Container加到newlyAllocatedContainers和liveContainers列表中
4、把分配的container信息同步到appSchedulingInfo当中
5、发送RMContainerEventType.START事件
6、更新FSSchedulerNode记录的container信息
7、如果被分配的是AM,则设置amRunning为true
如果资源不够,则检查是否可以预留资源
条件:
1)Container的资源请求必须小于Scheduler的增量分配内存 * 倍数(默认应该是2g)
2)如果已经存在的预留数 < 本地性对应的可用节点 * 预留比例
3)一个节点只允许同时为一个APP预留资源
相关参数:
yarn.scheduler.increment-allocation-mb 1024
yarn.scheduler.increment-allocation-vcores 1
yarn.scheduler.reservation-threshold.increment-multiple 2
yarn.scheduler.fair.reservable-nodes 0.05
2.3.4 如何确定优先顺序
该比较规则同时适用于队列和APP,详细代码位于FairSharePolicy当中
MinShare = Math.min(getMinShare(), getDemand())
1、(当前资源使用量 / MinShare)的比值越小,优先级越高
2、如果双方资源使用量都超过MinShare,则(当前资源使用量 / 权重)的比值越小,优先级越高
3、启动时间越早,优先级越高
4、最后实在比不出来,就比名字...
从上面分配的规则当中能看出来MinShare是非常重要的一个指标,当资源使用量没有超过MinShare之前,队列在分配的时候就会比较优先,切记一定要设置啊!
注:getMinShare()是FairScheduler当中队列的minResources
<minResources>6887116 mb,4491 vcores</minResources>
- OpenCV角点检测源代码分析(Harris和ShiTomasi角点)
- Java基础-day03-代码题
- mongodb11天之屠龙宝刀(九)js函数入门:MongoDB基于js的数据类型修改
- Go语言社区 APP --问答模块数据存储流程及代码
- Java基础-day09-重构随机点名器
- OpenCV3.4两种立体匹配算法效果对比
- 文件操作常用函数
- Java基础-day09-对象;类;封装 学生管理系统
- two Pass方法连通域检测
- 【Java入门提高篇】Day14 Java中的泛型初探
- 使用shell脚本快速得到主备关系(r9笔记第93天)
- 【Java入门提高篇】Day13 Java中的反射机制
- 仿腾讯课堂固定滚动列表ReactNative组件
- Golang通过socket与java通讯
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- 私藏的5个好用的Pandas函数!
- 漫画:如何在数组中找到和为 “特定值” 的两个数?
- 当你在百度搜索关键字的时候,哪个网站会排在最前面?今天给大家科普一下“网站SEO”
- Mysql系列第十九讲 异常捕获及处理详解
- 【原创】经验分享:一个小小emoji尽然牵扯出来这么多东西?
- jQuery 事件对象,拷贝对象,多库共存
- jQuery 事件注册与事件处理
- v-charts 修改文本颜色
- 使用 jsDelivr 免费加速 GitHub Pages 博客的静态资源(二)
- 竞赛比完,代码、模型怎么处理?Kaggle大神:别删,这都是宝藏
- 显示DataGrid序号的一个适用的方法
- SAP Spartacus cms-components.service.ts里的config.cmsComponents
- ng-template和ng-container的嵌套使用
- SAP Spartacus pageSlot一览
- Angular @Hostbinding工作原理