大数据学习YARN资源调度原理-每日运维

Hadoop主要是由三部分组成，除了分布式文件系统HDFS、分布式计算框架MapReduce，还有一个是分布式集群资源调度框架Yarn。但Yarn也并不是Hadoop从1.0架构中就有的，在2.0架构中，将MapReduce中的JobTracker作业控制和资源管理两个功能剥离出来，形成独立的框架，这个框架就是Yarn。

为什么需要Yarn

上一篇文章MapReduce编程模型学习了MapReduce，在MapReduce1.0架构中，JobTracker负责作业调度和资源管理，这种架构的缺点有：

可扩展性差
- JobTracker既要做资源管理，又要做任务监控，job的最大并发数受限制。
- 服务器集群资源调度管理和MapReduce执行过程耦合在一起，如果想在当前集群中运行其他计算任务，比如Spark或者Flink，就无法统一使用集群中的资源了。
可用性差
- JobTracker和NameNode一样，都存在单点故障问题，如果JobTracker挂了，整个MapReduce都不能用了
资源利用率低
- Map Slot和Reduce Slot的设计无法分享，造成资源利用不合理
无法支持更多的计算模型
- 只能支持MapReduce计算模型，不能调度流式计算，DAG计算，迭代计算等。

所以我们需要把MapReduce的资源管理和计算框架分开，这也是Hadoop2.0架构最主要的变化，就是将Yarn从MapReduce中分离出来，成为一个独立的资源调度框架。

Yarn基本组成

从上图中可以看出Yarn也是Master/Worker架构，分为两部分：

ResourceManager
- 负责整个集群的资源调度管理和分配，通常部署在独立的服务器上
- 负责处理Client的请求
- 负责启动和监控ApplicationMaster
- 监控NodeManager
NodeManager
- 负责具体服务器上的资源和任务管理，在集群的每一台计算服务器上都会启动，基本上跟HDFS的DataNode进程一起出现。
- 处理来自ResourceManger的命令
- 处理来自ApplicationMaster的命令

ResourceManager

ResourceManager除了上述的功能，还有两个重要的角色：调度器（Scheduler）和应用程序管理器（Applications Manager，AM）

Scheduler

调度器其实就是一个资源分配算法，根据应用程序（Client）提交的资源申请和当前服务器集群的资源状况进行资源分配。实际上Yarn内置了很多调度算法，也可以自定义调度算法，调度算法的作用是决定一个计算任务需要放在集群中的哪台机器上面。

资源调度算法

FIFO

先来的先被调用，先分配CPU、内存等资源，后来的在队列等待。只有等先来的应用程序资源满足后，再开始为下一个应用程序进行调度运行和分配资源。

优点：这种方式适合平均计算时间、耗用资源情况差不多的作业，优先级一样的按先来后到方式运行。

缺点：

所有的任务都按照同一优先级处理。
应用程序并发运行程度低。
如果前一个任务耗用资源特别久的作业，比如占用几个小时乃至几天的大部分机器的 CPU和内存的训练算法作业，导致排在后面的大量很短时间运行完、耗用资源比较少的作业很久才被调度，实际上他们优先调度更适合。

SJF

Shortest Job First，SJF，为了追求平均等待时间，平均周转时间，平均带权周转时间最短，人们提出了短作业优先算法。各个任务在开始执行之前，必须事先预计好它的执行时间，然后调度器将根据这些时间，从中选择用时较短的任务优先执行。

优点：这种方式已经解决FIFO算法的缺点，不会让短任务因为前面出现的大任务而堆积。

缺点：不利于长作业，对短作业有利，由此可能产生饥饿现象，如果一个用户一次提交了大量短作业就会导致其他用户提交的长作业长时间得不到服务。

Round Robin

时间片轮转调度算法：把系统当中的所有就绪任务按照先来先服务的原则，排成一个队列，然后再每次调度的时候，把处理器分派给队列当中的第一个任务，让它去执行一小段CPU时间(即时间片，time slice)。当这个时间片结束时，如果任务还没有执行完成的话，将会发生时钟中断，在时钟中断里面，调度器将会暂停当前任务的执行，并把它送到就绪队列的末尾，然后执行当前的队首任务。反之，如果一个任务在它的时间片用完之前就已经结束了或者阻塞了，那么它就会立即让出CPU给其他任务。

优点：不会产生饥饿问题，大家获得公平的资源分配。

缺点：该算法要求计算框架支持中断。此外，时间片的大小要适当选取，如果选择不当，将会影响到系统的性能和效率。如果时间片太大，每个任务都在一个时间片内完成，这就退化为先来先服务算法了，如果太小，任务之间的切换次数增加，从而增大了系统的管理开销，降低了CPU的使用效率。

Min-Max Fair

最大最小公平调度(Min-Max Fair)

将资源平分成n份,每份都是S/n,把每份分给相应的用户

如果超过了用户的需求，就回收超过的部分

然后把总体回收的资源，平均分给上一轮分配中尚未得到满足的用户，依次类推，直到没有回收的资源为止

加权最大最小公平调度(Weighted Min-Max Fair)

令W=w1 + w2+ … + wn, 将资源按照权重分成n份，每份分别是：Sw1/W, Sw2/W,…, S*wn/W。把每份分给相应的用户。

如果超过了用户的需求，就回收超过的部分，假设有m个用户尚未得到满足

然后把总体回收的资源，按照目前尚未满足的用户的权重分成m份，给对应的用户。依次类推，直到没有回收的资源为卡。

优点：考虑到了公平性，无论是大任务还是小任务，都能相对公平的得到服务。

缺点：加权的设置对平台有着更高的要求，加权不合理可能会破坏公平性。

Capacity

首先划分多个队列，队列资源采用容量占比的方式进行分配。每个队列设置资源最低保证和资源使用上限。如果队列中的资源有剩余或者空闲，可以暂时共享给那些需要资源的队列，而一旦该队列有新的应用程序需要资源运行，则其他队列释放的资源会归还给该队列。

优点：资源可以得到最大化的利用

缺点：但是小任务仍然会等待较多的时间。而且无法自定义优先级。

调度器

FIFO Scheduler

FIFO先进先出调度器，同一时间队列中只有一个任务在执行。
FIFO调度器以集群资源独占的方式来运行作业，这样的好处是一个作业可以充分利用所有的集群资源

Capacity Scheduler

层次化队列设计，每个队列采用FIFO调度策略，同一时间队列中只有一个任务在执行。队列的并行度为队列的个数。
可以为每个队列分配一个资源占比，管理员可为每个队列设置资源最低保证和资源使用上限。
灵活性，空闲的资源可以分配给任何队列，出现资源竞争时又按照占比再平衡
多租户支持，每个队列都有访问控制，用户只能提交到有权限的队列
队列支持树形嵌套
YARN的默认调度器

Fair Scheduler

层次化队列设计，每个队列内部的作业案缺额分配资源，同一时间队列中有多个任务执行。队列的并行度大于等于队列的个数。
可以让短的作业在合理的时间内完成，而不必一直等待长作业的完成。
所有队列的资源在无竞争时是共享的。
作业可以设置优先级。

在资源有限的情况下，每个job理想情况下获得的资源和实际获得的资源存在一个差距，这个差距叫缺额。同一个队列中job的缺额越大，越先获得资源优先执行。同时，还可以为job设置优先级，优先级越高，越先分配资源。

Capacity Scheduler和Fair Scheduler的区别

相同点:（Capacity有的Fair都有)

以队列划分资源
设定最低保证和最大使用上限
在某个队列空闲时可以将资源共享给其他队列。

不同点:

Fair Scheduler队列内部支持多种调度策略，包括
- FIFO
- Fair(队列中的N个作业，每个获得该队列1 / N的资源)
- DRF(Dominant Resource Fairness)(多种资源类型e.g. CPU，内存的公平资源分配策略)
Capacity Scheduler队列内部支持的调度策略
- FIFO
- DRF
Fair Scheduler可以使得小应用快速获得资源，避免了饿死的情况。优先选择对资源的缺额比例大的
Capacity Scheduler优先选择资源利用率低的队列

ApplicationsManager

ApplicationsManager负责接受作业提交，协商执行特定于应用程序的ApplicationMaster的第一个容器，并提供在失败时重新启动ApplicationMaster容器的服务。每个应用程序的ApplicationMaster负责与Scheduler协商适当的资源容器，跟踪它们的状态并监视进度。

NodeManager

负责Container生命周期管理
监控每个Container的资源使用情况
跟踪节点健康状况
以心跳的方式与ResourceManager保持通信
向ResourceManager汇报作业资源使用情况和容器的运行状态
接收来自ApplicationMaster启动和停止容器的请求

ApplicationMaster

ApplicationMaster实际上是特定计算框架的一个实例，每种计算框架都有自己独特的ApplicationMaster，负责与ResourceManager协商资源，并和NodeManager协同来执行和监控Container。MapReduce只是可以运行在YARN上其中一种计算框架。

与ResourceManager协商获取资源，以Container的形式为ApplicationMaster分配资源
把获得的资源进一步分配给内部的各个任务
与NodeManager保持通信，进行App的启动，运行，监控，停止
向ResourceManager发送心跳消息，报告资源的使用情况和应用的进度信息
作业完成时，ApplicationMaster向ResourceManager注销容器

Container

Container是YARN中资源的抽象，它封装了某个节点上一定量的资源(CPU 和内存两类资源)。它跟Linux Container没有任何关系，仅仅是YARN提出的一个概念(从实现上看，可看做一个可序列化/反序列化的Java类)。

Container由ApplicationMaster向ResourceManager申请的，由ResouceManager中的资源调度器异步分配给ApplicationMaster，但不会立马为它返回满足要求的资源，ApplicationMaster会不断与ResourceManager通信，探测分配到资源。

Container的运行是由ApplicationMaster向资源所在的NodeManager发起的，Container运行时需提供内部执行的任务命令(可以使任何命令，比如java、Python、C++进程启动命令均可)以及该命令执行所需的环境变量和外部资源(比如词典文件、可执行文件、jar包等)，这些信息都会被封装到StartContainerRequest对象里。

Job提交流程

1.用户向 YARN 中提交应用程序，其中包括 MRAppMaster 程序、启动 MRAppMaster 的命令、用户程序等。(MRAppMstr 程序在客户端生成，这一步已经将应用程序先行程序提交到RM的Application Manager模块进行处理，但此时运行程序所需的jar包、环境变量、切片信息等提交到HDFS之上，需要等MRAPPMstr返回一个可用的Container的时候在节点提交执行。可以理解为惰性处理，减少资源占用，做到需要什么提交什么)
2.ResourceManager 为该应用程序分配第一个 Container，并与对应的 Node-Manager 通信，要求它在这个 Container 中启动应用程序的MRAppMaster。（APPMaster就是先头兵，这时的操作是RM的ApplicationManager来进行处理）
3.MRAppMaster 首先向 ResourceManager 注册，这样用户可以直接通过ResourceManage 查看应用程序的运行状态，然后它将为各个任务申请资源，并监控它的运行状态，直到整个应用运行结束。
4.MRAppMaster 采用轮询的方式通过 RPC 协议向 ResourceManager 申请和领取资源。
5.一旦 MRAppMaster 申请到资源后，便与对应的 NodeManager 通信，要求它启动任务。(理解集群中不同节点的资源动态变动，可用Container以及不同的Task可以在不同的节点运行)
6.NodeManager 为任务设置好运行环境（包括环境变量、JAR 包、二进制程序等）后，将任务启动命令写到一个脚本中，并通过运行该脚本启动任务。（此时，客户端才真正上传具体Task需要的Jar包等等运行资源，同时NodeManager通过调用资源运行对应的Task任务）；
7.各个任务通过某个 RPC 协议向 MRAppMaster 汇报自己的状态和进度，以让 MRApplicationMaster 随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务。在应用程序运行过程中，用户可随时通过 RPC 向 MRAppMaster 查询应用程序的当前运行状态。
步骤4~7是重复执行的
8.应用程序运行完成后，MRAppMaster 向 ResourceManager 注销并关闭自己。

Node Label

Yarn Node Label和K8s的Node Label+NodeSelector的机制类似。目前只有Capacity Scheduler调度器支持Node Labels分区调度，可以通过调度器配置或者计算引擎node-label-expression参数让队列上的任务容器调度到队列可访问的分区上。

HDFS异构存储只能设置让某些数据(以目录为单位)分别在不同的存储介质上，但是计算调度时无法保障作业运行的环境。
在节点标签(NodeLabel)出现之前，资源申请方是无法指定特定节点的资源的，比如需要运行GPU节点，需要有SSD存储的节点等等。应用是无法指定特定节点上面的特定资源的，我们也无法对集群中的各个节点进行分区。
同时，同一个集群中如果需要将作业跑在特殊的软件环境下，例如特定版本的JDK，特定版本的 Python库等等，我们也无法进行指定。
同一个集群中新采购的机器具有大内存和高CPU，机型的不同或者网络环境的不同等等导致同一个作业的不同任务执行时间差异过大。
同一个集群跨越不同机房，应尽量避免同一个作业跨机房执行的需求。

Node Label的特性

给集群中的node打上标签，每个node只能有一个标签。
整个集群node划分为互斥的若干分区，没有设置标签的node属于DEFAULT分区。
分区又可以配置为两类：
- exclusive分区：只允许请求和该分区匹配的容器调度到该分区的节点上。
- non-exclusive分区：除了分配该分区容器请求外，还允许在有空闲资源时将请求为DEFAULT分区的容器调度上来（或请求未特殊制定分区）。
队列可以绑定多个标签，不设置默认标签的队列，使用DEFAULT分区的node。
通过队列的默认标签，和指定特定标签，实现节点分配控制

Node Label的配置和使用

# 使用YARN默认管理员hadoop账户执行
sudo su hadoop
# 添加分区
yarn rmadmin -addToClusterNodeLabels DEMO,CORE
# 列出YARN节点列表
yarn node -list
# 配置指定节点分区映射
yarn rmadmin -replaceLabelsOnNode node1:45454,DEMO
yarn rmadmin -replaceLabelsOnNode node2:45454,CORE
# 查看标签
yarn node -status node1:45454 
# 也可以通过Yarn管理页面查看Node Label 
# node-label webUI http://RM-Address:port/cluster/nodelabels

Yarn HA

RM存在单点故障问题。YARN的HA架构和HDFS HA类似，需要启动两个ResourceManager，这两个ResourceManager会向ZooKeeper集群注册，通过 ZooKeeper管理它们的状态(Active或Standby)并进行自动故障转移。

和HDFS HA的ZKFC不同，这里的ZKFC是RM内部的一个线程
ZKFC线程定期向zk发送心跳
RMStateStore存储在zk的/rmstore目录下，相当于一个文件夹
RM将状态写入RMStateStore中

YARN HA故障转移机制

Container故障：Resource Manager可以分配其他的Container继续执行
App Master故障：分配新的Container，启动App Master，新的App Master从App Manager获取相关恢复信
NodeManager故障：移除这个节点，在其他的NodeManager重启继续任务。
ResourceManager故障：在Yarn集群中，ResourceManager可以启动多台，只有其中一台是active状态的，其他都处于待命状态。这台active 状态的ResourceManager执行的时候会向ZooKeeper集群写入它的状态；当它故障的时候这些RM首先选举出另外一台leader变为active状态，然后从ZooKeeper集群加载ResourceManager的状态；在转移的过程中它不接收新的Job，转移完成后才接收新Job。

Yarn Federation

对于HDFS的扩展性问题来说，我们讲过了HDFS federation的几种方案，通过横向扩展命名空间的做法来延展其扩展性。随着集群规模的扩张，不仅仅存储系统会有性能瓶颈问题，计算系统也会存在这样的问题。YARN的ResourceManager在管理这上千甚至上万个NodeManager节点时，会面临着许多性能问题，此外，大量在跑的应用会产生大量的event，也同样加重了RM的性能问题。一般这种情况，我们的一个简单直接的方案是再搭建一个新的YARN集群。

对于不方便增加一个YARN集群的情况来说，目前社区提供了一套类似HDFS Federation的方案:
YARN Federation
涉及的主要有以下两大模块:

多集群状态信息收集，存储让众多独立小集群变为逻辑意义上的一个超大资源池(State Store)

客户端请求路由服务实现对于客户端来说，它面对的将不是众多具体的独立小集群(Proxy Store)

SubCluster：YARN 子集群，一般子集群要具有高可用性，即可以容忍 RM，NM 宕机，实现伤害最小化。如果一个子集群整个挂掉，联邦机制应该保证这个子集群上的作业被重新提交到另一个子集群上。在联邦集群里，子集群是一个扩展单位。我们可以通过添加一个或者多个联邦来扩展一个联邦集群。而且，各自独立的子集群可以决定是否并入联邦系统，贡献出自己一部分的资源
Router：为联邦架构下对外提供服务的接口，实现了ApplicationClientProtocol担当了以前RM的角色，屏蔽了底层多个子集群RMs的存在。对客户端的请求进行转发，根据策略将请求转发至子集群。该组件是无状态的，一个Federation 集群可以配置一组，但最少配置一个。用户提交应用时首先会访问其中一个 Router，然后 Router 会先从 State Store 中获得所有子集群信息（active rm 和其他一些使用率信息），之后根据配置的路由策略将应用程序提交请求转发到对应的 RM 上。一个作业在哪一个子集群中启动，这个子集群就被称为该作业的"home sub-cluster",其它子集群都是该作业的"secondary sub-cluster"。
StateStore：Federation 集群状态存储组件，其中主要记录了所有 sub-cluster 的信息。目前提供了 Memory、MySQL、Zookeeper 几种存储实现，可以根据自己的场景选择一种。一般使用 Zookeeper 存储。
AMRMProxy：是保证应用可以跨子集群运行的关键组件，充当应用程序和多个 RM 通讯的桥梁。AMRMProxy 会在每个 NM 上启动，实现了 ApplicationMasterProtocol，充当 AM 向 RM 的代理。YARN Federation 架构应用不允许和子集群中的 RM 直接通信，应用被限定只能与 AMRMProxy 通信，以此来实现多 RMs 的透明访问。
PolicyStore：路由策略存储组件，其中主要包含应用程序和资源请求如何路由到不同子集群的策略。目前的实现提供了多种策略，从 hashing、loadbased、weighted、priority 到更复杂的策略，这些策略包含子集群负载和本地化需求。
Global Policy Generator：全局策略生成器(GPG)会持续观测整个联邦系统，确保其配置合理性。设计中不要求 GPG 始终开启，可以更加准确地更新用户在子集群上的容量影射表，并且更加准确地更新 Router，AMRMProxy 甚至 RMs 中的策略。该特性暂未实现，初始实现代码已从 trunk 中删除（YARN-3657）。

AM跨集群资源调度

YARN Federation支持跨集群资源调度的情况。它的原理是在AM和RM之间增添了一个 Proxy Service做中间的拦截，此服务叫做AMRMProxy。

AM不仅仅可以与本地的RM申请资源，还能够向其它集群申请资源。这里借助的角色就是上文提到的AMRMProxy服务。AMRMProxy负责和其他非本地RM进行通信。

Router 接收到一个通过 YARN Application Client Protocol 提交的应用请求。

Router 通过查询路由(策略)表来为该应用选择一个＂home RM＂。

Router 访问 statestore 来确定该 home RM 的地址

Router 将应用提交请求重定向到 home RM。

Router 将包含 home RM 标识的应用状态信息保存到 statestore 中

当应用被提交到 home RM 上时，作业的执行流程就开始了。例如，应用会被提交到调度器队列上，它的 AM 会在 home-sub-cluster 上的第一个有足够资源的 Nodemanager 上被启动。 a. 在这个过程中，AM 的执行环境中会加入 AMRMProxy 的地址，以便和之后的 YARN RM 进行通信。 b. AM 的安全令牌也会被修改，这样，AM 就只能和 AMRMProxy 通信。之后 AM 和 RM 之间的一切通信都由 AMRMProxy 来完成。

AM 根据 HDFS 上的本地信息来申请 containers

AMRMProxy 可以基于某种策略，扮演成 AM 在其它子集群上启动一个 Unmanaged AM，并且向两个子集群同时汇报集群。

AMRMProxy 会同时根据本地信息和从 GPG 获得的调度策略来决定是从 home-sub-cluster 还是从 secondary sub cluster 来请求资源。在上图中，AMRMProxy 就是决定向 Secondary sub cluster 来请求资源。

Secondary RM 为 AMRMProxy 提供有效的 container 令牌来在该 RM 集群上启动一个 container。这样一来，每个 sub-cluster 都可以保证只使用到自己的安全令牌，从而避免需要一个全局的产生令牌的组件

AMRMProxy 把资源被分配的响应传回给 AM。

AM 使用标准的 YARN 协议在目标 Nodemanager（在 sub-cluster2）上启动 container。

参考文献

Hadoop-Yarn官网

Yarn源码分析4-资源调度算法

YARN Federation技术解析及应用_YARN_移动云大数据_InfoQ写作社区

Yarn技术原理

Hadoop Yarn 一文搞懂 Yarn架构原理和工作机制 - 清风画扇 - 博客园

大数据学习YARN资源调度原理