大数据框架之Hadoop：MapReduce（五）Yarn资源调度器

Apache YARN (Yet Another Resource Negotiator) 是 hadoop 2.0 引入的集群资源管理系统。用户可以将各种服务框架部署在 YARN 上由 YARN 进行统一地管理和资源分配。

简言之Yarn是一个资源调度平台负责为运算程序提供服务器运算资源相当于一个分布式的操作系统平台而MapReduce等运算程序则相当于运行于操作系统之上的应用程序。

5.1Yarn基本架构

YARN主要由ResourceManager、NodeManager、ApplicationMaster和Container等组件构成如下图所示。

1. ResourceManager

ResourceManager 通常在独立的机器上以后台进程的形式运行它是整个集群资源的主要协调者和管理者。ResourceManager 负责给用户提交的所有应用程序分配资源它根据应用程序优先级、队列容量、ACLs、数据位置等信息做出决策然后以共享的、安全的、多租户的方式制定分配策略调度集群资源。

2. NodeManager

NodeManager 是 YARN 集群中的每个具体节点的管理者。主要负责该节点内所有容器的生命周期的管理监视资源和跟踪节点健康。具体如下

• 启动时向 ResourceManager 注册并定时发送心跳消息等待 ResourceManager 的指令
• 维护 Container 的生命周期监控 Container 的资源使用情况
• 管理任务运行时的相关依赖根据 ApplicationMaster 的需要在启动 Container 之前将需要的程序及其依赖拷贝到本地。

3. ApplicationMaster

在用户提交一个应用程序时YARN 会启动一个轻量级的进程 ApplicationMaster。ApplicationMaster 负责协调来自 ResourceManager 的资源并通过 NodeManager 监视容器内资源的使用情况同时还负责任务的监控与容错。具体如下

• 根据应用的运行状态来决定动态计算资源需求
• 向 ResourceManager 申请资源监控申请的资源的使用情况
• 跟踪任务状态和进度报告资源的使用情况和应用的进度信息
• 负责任务的容错。

4. Container

Container 是 YARN 中的资源抽象它封装了某个节点上的多维度资源如内存、CPU、磁盘、网络等。当 AM 向 RM 申请资源时RM 为 AM 返回的资源是用 Container 表示的。YARN 会为每个任务分配一个 Container该任务只能使用该 Container 中描述的资源。ApplicationMaster 可在 Container 内运行任何类型的任务。例如MapReduce ApplicationMaster 请求一个容器来启动 map 或 reduce 任务而 Giraph ApplicationMaster 请求一个容器来运行 Giraph 任务。

5.2Yarn工作机制

2.1 工作原理简述

1. Client 提交作业到 YARN 上
2. Resource Manager 选择一个 Node Manager启动一个 Container 并运行 Application Master 实例
3. Application Master 根据实际需要向 Resource Manager 请求更多的 Container 资源如果作业很小, 应用管理器会选择在其自己的 JVM 中运行任务
4. Application Master 通过获取到的 Container 资源执行分布式计算。

2.2 工作原理详述

1. 作业提交

client 调用 job.waitForCompletion 方法向整个集群提交 MapReduce 作业 (第 1 步) 。新的作业 ID(应用 ID) 由资源管理器分配 (第 2 步)。作业的 client 核实作业的输出, 计算输入的 split, 将作业的资源 (包括 Jar 包配置文件, split 信息) 拷贝给 HDFS(第 3 步)。 最后, 通过调用资源管理器的 submitApplication() 来提交作业 (第 4 步)。

2. 作业初始化

当资源管理器收到 submitApplciation() 的请求时, 就将该请求发给调度器 (scheduler), 调度器分配 container, 然后资源管理器在该 container 内启动应用管理器进程, 由节点管理器监控 (第 5 步)。

MapReduce 作业的应用管理器是一个主类为 MRAppMaster 的 Java 应用其通过创造一些 bookkeeping 对象来监控作业的进度, 得到任务的进度和完成报告 (第 6 步)。然后其通过分布式文件系统得到由客户端计算好的输入 split(第 7 步)然后为每个输入 split 创建一个 map 任务, 根据 mapreduce.job.reduces 创建 reduce 任务对象。

3. 任务分配

如果作业很小, 应用管理器会选择在其自己的 JVM 中运行任务。

如果不是小作业, 那么应用管理器向资源管理器请求 container 来运行所有的 map 和 reduce 任务 (第 8 步)。这些请求是通过心跳来传输的, 包括每个 map 任务的数据位置比如存放输入 split 的主机名和机架 (rack)调度器利用这些信息来调度任务尽量将任务分配给存储数据的节点, 或者分配给和存放输入 split 的节点相同机架的节点。

4. 任务运行

当一个任务由资源管理器的调度器分配给一个 container 后应用管理器通过联系节点管理器来启动 container(第 9 步)。任务由一个主类为 YarnChild 的 Java 应用执行 在运行任务之前首先本地化任务需要的资源比如作业配置JAR 文件, 以及分布式缓存的所有文件 (第 10 步。 最后, 运行 map 或 reduce 任务 (第 11 步)。

YarnChild 运行在一个专用的 JVM 中, 但是 YARN 不支持 JVM 重用。

5. 进度和状态更新

YARN 中的任务将其进度和状态 (包括 counter) 返回给应用管理器, 客户端每秒 (通 mapreduce.client.progressmonitor.pollinterval 设置) 向应用管理器请求进度更新, 展示给用户。

6. 作业完成

除了向应用管理器请求作业进度外, 客户端每 5 分钟都会通过调用 waitForCompletion() 来检查作业是否完成时间间隔可以通过 mapreduce.client.completion.pollinterval 来设置。作业完成之后, 应用管理器和 container 会清理工作状态 OutputCommiter 的作业清理方法也会被调用。作业的信息会被作业历史服务器存储以备之后用户核查。

三、作业提交全过程

3.1 主要流程

1、Yarn运行机制

如下图所示。

2、工作机制详解

1MR程序提交到客户端所在的节点。

2YarnRunner向ResourceManager申请一个Application。

3RM将该应用程序的资源路径返回给YarnRunner。

4该程序将运行所需资源提交到HDFS上。

5程序资源提交完毕后申请运行mrAppMaster。

6RM将用户的请求初始化成一个Task。

7其中一个NodeManager领取到Task任务。

8该NodeManager创建容器Container并产生MRAppmaster。

9Container从HDFS上拷贝资源到本地。

10MRAppmaster向RM 申请运行MapTask资源。

11RM将运行MapTask任务分配给另外两个NodeManager另两个NodeManager分别领取任务并创建容器。

12MR向两个接收到任务的NodeManager发送程序启动脚本这两个NodeManager分别启动MapTaskMapTask对数据分区排序。

13MrAppMaster等待所有MapTask运行完毕后向RM申请容器运行ReduceTask。

14ReduceTask向MapTask获取相应分区的数据。

15程序运行完毕后MR会向RM申请注销自己。

3.2 详细过程

1、作业提交过程之YARN

如下图所示。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vHzsFWNT-1677842175783)(5Yarn%E8%B5%84%E6%BA%90%E8%B0%83%E5%BA%A6%E5%99%A8%2052c4bf3665e64b858624bae9765fda4e/Untitled%206.png)]

作业提交全过程详解

1作业提交

第1步Client调用job.waitForCompletion()方法向整个集群提交MapReduce作业。

第2步Client向RM申请一个作业id。

第3步RM向Client返回该job资源的提交路径和作业id。

第4步Client核实作业的输出, 计算输入的 split提交jar包、切片信息和配置文件到指定的资源提交路径(HDFS)。

第5步Client提交完资源后通过调用资源管理器的 submitApplication()向RM提交作业。

2作业初始化

第6步当RM收到Client 的submitApplciation() 的请求后将该job添加到容量调度器中。

第7步调度器分配某一个空闲的NM执行此Job。

第8步NM创建Container并启动MRAppmaster。

第9步下载Client提交的资源到本地。

3任务分配

第10步MrAppMaster向RM申请运行多个MapTask任务资源。

第11步RM将运行MapTask任务分配给另外两个NodeManager另两个NodeManager分别领取任务并创建容器。

4任务运行

第12步MR向两个接收到任务的NodeManager发送程序启动脚本这两个NodeManager分别启动MapTaskMapTask对数据分区排序。

第13步MrAppMaster等待所有MapTask运行完毕后向RM申请容器运行ReduceTask。

第14步ReduceTask向MapTask获取相应分区的数据。

第15步程序运行完毕后MR会向RM申请注销自己。

5进度和状态更新

YARN中的任务将其进度和状态(包括counter)返回给应用管理器, 客户端每秒(通过mapreduce.client.progressmonitor.pollinterval设置)向应用管理器请求进度更新, 展示给用户。

6作业完成

除了向应用管理器请求作业进度外, 客户端每5秒都会通过调用waitForCompletion()来检查作业是否完成。时间间隔可以通过mapreduce.client.completion.pollinterval来设置。作业完成之后, 应用管理器和Container会清理工作状态。作业的信息会被作业历史服务器存储以备之后用户核查。

2、作业提交过程之MapReduce

如下图所示

四、资源调度器

目前Hadoop作业调度器主要有三种FIFO、Capacity Scheduler和Fair Scheduler。Hadoop2.7.2默认的资源调度器是Capacity Scheduler。

具体设置详见yarn-default.xml文件

<property>
    <description>The class to use as the resource scheduler.</description>
    <name>yarn.resourcemanager.scheduler.class</name>
<value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value>
</property>

1、先进先出调度器FIFO如图所示

2、容量调度器Capacity Scheduler如图所示系统默认为容量调度器。

3、公平调度器Fair Scheduler如图所示

五、任务的推测执行

1、作业完成时间取决于最慢的任务完成时间

一个作业由若干个Map任务和Reduce任务构成。因硬件老化、软件Bug等某些任务可能运行非常慢。

思考系统中有99%的Map任务都完成了只有少数几个Map老是进度很慢完不成怎么办

2、推测执行机制

发现拖后腿的任务比如某个任务运行速度远慢于任务平均速度。为拖后腿任务启动一个备份任务同时运行。谁先运行完则采用谁的结果。

3、执行推测任务的前提条件

1每个Task只能有一个备份任务

2当前Job已完成的Task必须不小于0.055%

3开启推测执行参数设置。mapred-site.xml文件中默认是打开的。

<property>
  	<name>mapreduce.map.speculative</name>
  	<value>true</value>
  	<description>If true, then multiple instances of some map tasks may be executed in parallel.</description>
</property>

<property>
  	<name>mapreduce.reduce.speculative</name>
  	<value>true</value>
  	<description>If true, then multiple instances of some reduce tasks may be executed in parallel.</description>
</property>

4、不能启用推测执行机制情况

1任务间存在严重的负载倾斜

2特殊任务比如任务向数据库中写数据。

5、算法原理如图4-20所示