<>1、Yarn资源调度器

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

<>1.1 Yarn基础架构

Yarn主要由ResourceManager、NodeManager、ApplicationMaster和Container等组件构成。

1、ResourceManager(RM)主要作用:
(1)处理客户端请求
(2)监控NodeManager
(3)启动或监控ApplicationMaster
(4)资源的分配和调度

2、NodeManager(NM)主要作用:
(1)管理单个节点上的资源
(2)处理来自ResourceManager的命令
(3)处理来自ApplicationMaster的命令

3、ApplicationMaster(AM)的主要作用·:
(1)为应用程序申请资源并分配给内部的任务
(2)任务的监控与容错

4、Container
Container是Yarn中的资源抽象,它封装了某个节点上的多维度资源,如内存、CPU、磁盘、网络等。

<>1.2 Yarn工作机制

(1)MR程序提交到客户端所在的节点
(2)YarnRunner向ResourceManager申请一个Application
(3)RM将该应用程序的资源路径返回给YarnRunner。
(4)该程序将运行所需资源提交到HDFS上
(5)程序资源提交完毕后,申请运行mrAppMaster
(6)RM将用户的请求初始化成一个Task
(7)其中一个NodeManager领取到Task任务
(8)该NodeManager创建容器,并尝试MRAppmaster
(9)Container从HDFS上拷贝资源到本地
(10)MRAppMaster向RM申请运行MapTask资源
(11)RM将运行MapTask任务分配给另外两个NodeManager,另外两个NodeManager分别领取任务并创建容器。
(12)MR向两个接受到任务的NodeManager发生程序启动脚本,这两个NodeManager分别启动MapTask,MapTask对数据分区排序。
(13)MRAppMaster等待所以MapTask运行完毕后,向RM申请容器,运行Reduce Task。
(14)ReduceTask向MapTask获取相应分区的数据
(15)程序运行完毕后,MR会向RM申请注销自己。

<>1.3 作业提交全过程

1、作业提交过程之YARN

2、作业提交之HDFS&&MapReduce

作业提交全过程详解
(1)作业提交
(a)client调用job.waitForCompletion()方法,向整个集群提交MapReduce作业。
(b)client向RM申请一个作业id
(c)RM给Client返回该job资源的提交路径和作业id
(d)client提交jar包、切片信息和配置文件到指定的资源提交路径
(e)client提交完资源后,向RM申请运行MRAppMaster

(2)作业初始化
(a)当RM收到client的请求后,将该job添加到容量调度器中
(b)某一个空间的NM领到该Job
(c)该NM创建Container,并产生MRAppMaster
(d)下载Client提交的资源到本地

(3)任务分配
(a)MRAppMaster向RM申请运行多个MapTask任务资源
(b)RM将运行MapTask任务分配给另外两个NodeManager,这两个NodeManager分别领取任务并创建容器。

(4)任务运行
(a)MR向两个接受到任务的NodeManager发送程序启动脚本,这两个NodeManager分别启动MapTask,MapTask对数据分区排序。
(b)MRAppMaster等待所有MapTask运行完毕后,向RM申请容器,运行ReduceTask。
(c)ReduceTask向MapTask获取相应分区的数据
(d)程序运行完毕后·,MR会向RM申请注销自己。

(5)进度和状态更新

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

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

<>1.4 Yarn调度器和调度算法

目前,Hadoop作业调度器主要有三种:FIFO、容量(Capacity Scheduler)和公平(Fair Scheduler)。Apache
Hadoop3.1.3默认的资源调度器是Capacity Scheduler。
CDH框架默认调度器是Fair 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.4.1 先进先出调度器(FIFO)
FIFO调度器(First In First Out):单队列,根据提交·作业的先后顺序,先来先服务。

优点:简单易懂
缺点:不支持多队列,生产环境很少使用
<>1.4.2 容量调度器(Capacity Scheduler)
Yahoo开发的多用户调度器

1、多队列:每个队列可配置一定的资源量,每个队列采用FIFO调度策略
2、容量保证:管理员可为每个队列设置资源最低保证和资源使用上限
3、灵活性:如果一个队列中的资源有剩余,可以暂时共享给那些需要资源的队列,而一旦该队列有新的应用程序,则其他队列借调的资源会归还给该队列。
4、多租户:
(1)支持多用户共享集群和多应用程序同时运行
(2)为了防止同一个用户的作业独占队列中的资源,该调度器会对象同一用户提交的作业所占资源量进行限定。

容量调度器资源分配算法:
(1)队列资源分配
从root开始,使用深度优先算法,优先选择资源占用率最低的队列分配资源。
(2)作业资源分配
默认按照提交作业的优先级和提交时间顺序分配资源
(3)容器资源分配
按容器的优先级分配资源:
如果优先级相同,按照数据本地性原则:
(a)任务和数据在同一个节点
(b)任务和数据在同一个机架
(c)任务和数据不同节点和不同机架
<>1.4.3 公平调度器(Fair Scheduler)
是Facebook开发的多用户调度器

1、与容量调度器相同点
(1)多队列:支持多队列多作业
(2)容量保证:管理员可为每个队列设置资源最低保证和资源使用上限
(3)灵活性:如果一个队列的资源有剩余,可以暂时共享给那些需要资源的队列,而一旦该队列有新的应用程序提交,则其他队列借调的资源会归还给该队列
(4)多租户:支持多用户共享集群和多应用程序同时运行;为了防止同一个用户的作业独占队列中的资源,该调度器会对同一个用户提交的作业所占资源量进行限定。

2、与容量调度器不同点
(1)核心调度策略不同
(a)容量调度器:优先选择资源利用率低的队列
(b)公平调度器:优先选择对资源缺额比例大的资源
(2)每个队列可以单独设置资源分配方式
(a)容量调度器:FIFO、DRF
(b)公平调度器:FIFO、FAIR、DRF

3、公平调度器队列资源分配方式
(1)FIFO策略
公平调度器每个队列资源分配策略如果选择FIFO的话,此时公平调度器相当于上面讲过的容量调度器。
(2)Fair策略

Fair策略(默认)是一种基于最大最小公平算法实现的资源多路复用方式。默认情况下,每个队列内部采用该方式分配资源。这意味着,如果一个队列中有两个应用程序在同时运行,则每个应用程序可以得到1/2的资源;如果三个应用程序同时运行,则每个应用程序可以得到1/3的资源。

具体资源分配流程和容量调度器一致:
(1)选择队列
(2)选择作业
(3)选择容器
以上三步,每一步都是按照公平策略分配资源
实际最小资源份额:mindshare = Min(资源需求量,配置的最小资源)
是否饥饿:isNeedy = 资源使用量 < mindshare(实际最小资源份额)
资源分配比:minShareRatio = 资源使用量 / Max(mindshare, 1)
资源使用权重比:useToWeightRatio = 资源使用量 / 权重

(3)公平调度器资源分配算法

(4)公平调度算法队列资源分配方式

(5)DRF策略
DRF(Dominant Resource
Fairness),我们之前说的资源,都是单一标准,例如只考虑内存(也是Yarn默认的情况)。但是很多时候我们资源有很多种,例如内存,CPU,网络带宽等,这样我们很难衡量两个应用应该分配的资源比例。

那么在YARN中,我们用DRF来决定如何调度:
假设集群一共有100 CPU和10T 内存,而应用A需要(2 CPU, 300GB),应用B需要(6
CPU,100GB)。则两个应用分别需要A(2%CPU, 3%内存)和B(6%CPU, 1%内存)的资源,这就意味着A是内存主导的,
B是CPU主导的,针对这种情况,我们可以选择DRF策略对不同应用进行不同资源(CPU和内存)的一个不同比例的限制。

<>1.5 Yarn常用命令

Yarn状态的查询,除了可以在ResourceManager节点ip+端口号8088在页面查看外,还可以通过命令操作。常见的命令操作如下:
<>1.5.1 yarn Application查看任务
(1)列出所有Application:
yarn application -list

(2)根据Application状态过滤 (所有状态:ALL、NEW、
NEW_SAVING、SUBMITTED、ACCEPTED、RUNNING、FINISHED、FAILED、KILLED)
yarn application -list -appStates
(3)kill 掉Application
yarn application -kill application_1612577921195_0001 <>1.5.2 yarn logs查看日志
(1)查询 Application 日志:yarn logs -applicationId
(2)查询 Container 日志:yarn logs -applicationId -containerId
<>1.5.3 yarn applicationattempt 查看尝试运行的任务
(1)列出所有 Application 尝试的列表:yarn applicationattempt -list
(2)打印 ApplicationAttemp 状态:yarn applicationattempt -status
<>1.5.4 yarn Container查看容器
(1)列出所有 Container:yarn container -list
(2)打印 Container 状态:yarn container -status

注意:只有在任务跑的途中才能看到 container 的状态
<>1.5.5 yarn node 查看节点状态
列出所有节点:yarn node -list -all
<>1.5.6 yarn rmadmin 更新配置
加载队列配置:yarn rmadmin -refreshQueues
<>1.5.6 yarn queue 查看队列
打印队列信息:yarn queue -status

<>1.6 Yarn生产环境核心参数

1、ResourceManager相关

核心参数说明
yarn.resourcemanager.scheduler.class配置调度器,默认容量
yarn.resourcemanager.scheduler.client.thread-count
ResourceManager处理调度器请求的线程数量,默认50
2、NodeManager相关

核心参数说明
yarn.nodemanager.resource.detect-hardware-capabilities是否让yarn自己检测硬件进行配置,默认false
yarn.nodemanager.resource.count-logical-processors-as-cores
是否将虚拟核数当作CPU核数,默认false
yarn.nodemanager.resource.pcores-vcores-multiplier
虚拟核数和物理核数乘数,例如:4核8线程,该参数就应设为2,默认1.0
yarn.nodemanager.resource.memory-mbNodeManager使用内存,默认8G
yarn.nodemanager.resource.system-reserved-memory-mb NodeManager
为系统保留多少内存以上二个参数配置一个即可
yarn.nodemanager.resource.cpu-vcoresNodeManager使用CPU核数,默认8个
yarn.nodemanager.pmem-check-enabled是否开启物理内存检查限制container,默认打开
yarn.nodemanager.vmem-check-enabled是否开启虚拟内存检查限制container,默认打开
yarn.nodemanager.vmem-pmem-ratio虚拟内存物理内存比例,默认2.1
3、Container相关

核心参数说明
yarn.scheduler.minimum-allocation-mb容器最最小内存,默认1G
yarn.scheduler.maximum-allocation-mb容器最最大内存,默认8G
yarn.scheduler.minimum-allocation-vcores容器最小CPU核数,默认1个
yarn.scheduler.maximum-allocation-vcores容器最大CPU核数,默认4个
<>2、案例实操

<>2.1 Yarn的Tool接口案例

(1)需求:自己写的程序可以动态传参。编写Yarn的Tool接口。
(2)编码
(a)WordCount类
package org.example._14yarntool; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.
Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.
mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.
output.FileOutputFormat; import org.apache.hadoop.util.Tool; import java.io.
IOException; /** * @ClassName WordCount * @Description TODO * @Author
Zouhuiming * @Date 2023/5/24 17:34 * @Version 1.0 */ public class WordCount
implements Tool { private Configuration conf; @Override public int run(String[]
args) throws Exception { Job job= Job.getInstance(conf); job.setJarByClass(
WordCountDriver.class); job.setMapperClass(WordCountMapper.class); job.
setReducerClass(WordCountReducer.class); job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class); job.setOutputKeyClass(Text.class)
; job.setOutputValueClass(IntWritable.class); FileInputFormat.setInputPaths(job,
new Path(args[0])); FileOutputFormat.setOutputPath(job,new Path(args[1]));
return job.waitForCompletion(true) ? 0 : 1; } @Override public void setConf(
Configuration configuration) { this.conf=configuration; } @Override public
Configuration getConf() { return conf; } public static class WordCountMapper
extends Mapper<LongWritable, Text,Text, IntWritable>{ private Text outK=new Text
(); private IntWritable outV=new IntWritable(1); @Override protected void map(
LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.
Context context) throws IOException, InterruptedException { String line=value.
toString(); String[] words = line.split(" "); for (String word : words) { outK.
set(word); context.write(outK,outV); } } } public static class WordCountReducer
extends Reducer<Text,IntWritable,Text,IntWritable>{ private IntWritable outV=new
IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable>
values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws
IOException, InterruptedException { int sum=0; for (IntWritable value : values)
{ sum+=value.get(); } outV.set(sum); context.write(key,outV); } } }
(b)Driver类
package org.example._14yarntool; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner;
import java.util.Arrays; /** * @ClassName WordCountDriver * @Description TODO *
@Author Zouhuiming * @Date 2023/5/24 17:42 * @Version 1.0 */ public class
WordCountDriver { private static Tool tool; public static void main(String[]
args) throws Exception { //1、创建配置文件 Configuration conf=new Configuration();
//2、判断是否有tool接口 switch (args[0]){ case "wordcount": tool=new WordCount(); break;
default: throw new RuntimeException("No such tool:"+args[0]); } //3、用Tool执行程序
//Arrays.copyOfRange()将老数组的元素放到新数组里面 int run=ToolRunner.run(conf,tool,Arrays.
copyOfRange(args,1,args.length)); System.exit(run); } }

技术
下载桌面版
GitHub
百度网盘(提取码:draw)
Gitee
云服务器优惠
阿里云优惠券
腾讯云优惠券
华为云优惠券
站点信息
问题反馈
邮箱:ixiaoyang8@qq.com
QQ群:766591547
关注微信