Hadoop学习-05-Yarn

理论相关

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

基础架构

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

ResourceManager：
- 处理客户端请求
- 监控NodeManager
- 启动或监控c
- 资源的分配与调度
NodeManager：
- 管理单个节点上的资源
- 处理来自ResourceManager的命令
- 处理来自ApplicationMaster的命令
ApplicationMaster：
- 为应用程序申请资源并分配给内部的任务
- 任务的监控与容错
Container：
- YARN资源抽象，封装了某个节点上的多维度资源，如内存、CPU、磁盘、网络等

工作机制

MR程序提交到客户端所在节点
YarnRunner向ResourceManager申请一个Application
RM将该应用程序的资源路径返回给YarnRunner
该程序将运行所需资源提交到HDFS上
程序资源提交完毕后，申请运行mrAppMaster
RM将用户的请求初始化成一个Task
其中一个NodeManager领取到Task任务
该NodeManager创建容器Container，并产生MRAppMaster
Container从HDFS上拷贝资源到本地
MRAppmaster向RM申请运行MapTask资源
RM将运行MapTask任务分配给另外两个NodeManager，另外两个NodeManager分别领取任务并创建容器
MR向两个接收到任务的NodeManager发送程序启动脚本，这两个NodeManager分别启动MapTask，MapTask对数据分区排序
MrAppMaster等待所有MapTask运行完毕后，向RM申请容器，运行ReduceTask
ReduceTask向MapTask获取相应分区的数据
程序运行完毕后，MR会向RM申请注销自己

作业提交过程

作业提交
1. Client调用waitForCompletion方法向集群提交 MapReduce作业
2. Client向RM申请一个作业id
3. RM向Client返回该Job资源的提交路径和作业id
4. Client提交jar包、切片信息和配置文件到指定的资源提交路径
5. client提交完资源后，向RM申请运行MrAppmaster
作业初始化
1. 当RM收到Client请求后，将该job添加到容量调度器中
2. 某一个空闲的NM领取到该job
3. 该NM创建container，并产生MRAppmaster
4. 下载Client提交的资源到本地
任务分配
1. MrAppmaster向RM申请运行多个MapTask任务资源
2. RM将运行MapTask任务分配给另外两个NodeManager，另两个NodeManager分别领取任务并创建容器
任务运行
1. MR向两个接收到任务的NodeManager发送程序启动脚本，这两个NodeManager分别启动MapTask，MapTask对数据分区排序
2. MrAppMaster等待所有MapTask运行完毕后，向RM申请容器，运行ReduceTask
3. ReduceTask向MapTask获取相应分区的数据
4. 程序运行完毕后，MR会向RM申请注销自己
进度和状态更新
- Yarn中的任务将其进度和状态返回给应用管理器，客户端每秒向应用管理器请求进度更新，展示给用户
作业完成
- 除了向应用管理器请求作业进度外，客户端没5s都会通过调用waitForCompletion来检查作业是否完成，时间间隔可以设置，应用管理器和container会清理工作状态，作业的信息会被作业历史服务器存储以备之后用户核查

调度器和调度算法

主要调度器有三种：FIFO、容量（Capacity Scheduler）、公平（Fair Scheduler），Apache Hadoop3.1.3默认的资源调度器是容量（Capacity Scheduler），CDH框架默认调度器为（Fair Scheduler）

FIFO

先来先服务

优点：简单易懂，服务如其名

缺点：不支持多队列，生产环境很少使用

容量调度器

Capacity Scheduler是Yahoo开发的多用户调度器

特点：

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

算法：

队列资源分配：从root开始，使用深度优先算法，优先选择资源占用率最低的队列分配资源
作业资源分配：默认按照提交作业的优先级和提交时间顺序分配资源
容器资源分配：按照容器的优先级分配资源，如果优先级相同，按照数据本地性原则：任务和数据在同一个节点上，任务和数据在同一个机架上，任务和数据不再同一个节点也不在同一个机架上

公平调度器

Fair Scheduler 是Facebook开发的多用户调度器，通队列所有任务共享资源，在时间尺度上获得公平的资源

与容量调度器相同点
- 多队列：每一个队列可以配置一定的资源量，每个队列采用FIFO调度策略
- 容量保证：管理员可以为每个队列设置资源最低保证和资源使用上限
- 灵活性：如果一个队列中的资源有剩余，可以暂时共享给哪些需要资源的队列，而一旦该队列有新的应用程序提交，则其它队列借调的资源会归还给该队列
- 多租户：支持多用户共享集群和多应用程序同时运行，为了防止同一个用户的作业独占队列中的资源，该调度器会对统一用户提交的作业所占资源量进行限定
不同点
- 核心调度策略不同
  - 容量调度器优先选择资源利用率低的队列
  - 公平调度器优先选择对资源的缺额比例大的
- 每个队列可以单独设置资源分配方式
  - 容量调度器：FIFO、DFR
  - 公平调度器：FIFO、FAIR、DRF
缺额：
- 公平调度器设计目标是在时间尺度上，所有作业获得公平的资源，某一时刻一个作业应获资源和实际获取资源的差距叫做缺额
- 调度器会优先为缺额大的作业分配资源
FAIR资源分配方式
- 默认是一种基于最大最小公平算法实现的资源多路复用方式，默认情况下，每个队列内部采用该方式分配资源，这意味如果一个队列中有两个应用程序同时运行，则每个应用程序可得到1/2的资源：如果三个应用程序同时运行，则每个应用程序可得到1/3的资源
- 具体资源分配流程与容量调度器一致
  - 选择队列
  - 选择作业
  - 选择容器
  - 以上三步每一步都是按照公平测量分配资源
- 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和内存）的一个不同比例的限制。

常用命令

# 列出所有 Application
yarn application -list
# 根据  Application 状态过滤：yarn application -list -appStates （所有状态：ALL、NEW、 NEW_SAVING、SUBMITTED、ACCEPTED、RUNNING、FINISHED、FAILED、KILLED）
yarn  application  -list  -appStates FINISHED
# Kill 掉 Application
yarn application -kill application_1612577921195_0001

#查询 Application 日志
yarn logs -applicationId <ApplicationId>
#查询 Container 日志
yarn logs -applicationId <ApplicationId> -containerId <ContainerId> 

#列出所有 Application 尝试的列表
yarn applicationattempt -list <ApplicationId>
#打印  ApplicationAttemp 状态
yarn applicationattempt -status <ApplicationAttemptId>

#列出所有 Container
yarn container -list <ApplicationAttemptId>
#打印 Container 状态 注：只有在任务跑的途中才能看到  container 的状态
yarn container -status <ContainerId>

#列出所有节点
yarn node -list -all

#加载队列配置
yarn rmadmin -refreshQueues
#打印队列信息
yarn queue -status <QueueName>

Yarn核心参数

RM：
- yarn.resource.manager.scheduler.class, 配置调度器，默认容量
- yarn.resource.manager.scheduler.client.thread-count，处理调度器请求的线程数量，默认50
NM：
- 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-mb NodeManager使用内存，默认8G
- yarn.nodemanager.resource.system-reserved-memory-mb NodeManager为系统保留多少内存
以上两个参数配置一个即可
- yarn.nodemanager.resource.cpu-vcores NodeManager使用CPU核数，默认8个
- yarn.nodemanager.pmem-check-enabled 是否开启物理内存检查限制container，默认打开
  
  调度器
- yarn.nodemanager.vmem-check-enabled 是否开启虚拟内存检查限制container，默认打开
- yarn.nodemanager.vmem-pmem-ratio 虚拟内存物理内存比例，默认2.1
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个

核心参数配置

yarn-site.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.capaci ty.CapacityScheduler</value>
</property>
<!-- ResourceManager处理调度器请求的线程数量,默认  50；如果提交的任务数大于  50，可以 增加该值，但是不能超过  3台   * 4线程   = 12线程（去除其他应用程序实际不能超过  8）   -->
<property>
    <description>Number of threads interface.</description>
    <name>yarn.resourcemanager.scheduler.client.thread-count</name>
    <value>8</value>
</property>
<!-- 是否让  yarn自动检测硬件进行配置，默认是  false，如果该节点有很多其他应用程序，建议 手动配置。如果该节点没有其他应用程序，可以采用自动   -->
<property>
    <name>yarn.nodemanager.resource.detect-hardware-capabilities</name>
    <value>false</value>
</property>
        <!-- 是否将虚拟核数当作  CPU核数，默认是  false，采用物理  CPU核数   -->
<property>
    <name>yarn.nodemanager.resource.count-logical-processors-as- cores</name>
    <value>false</value>
</property>
        <!-- 虚拟核数和物理核数乘数，默认是  1.0 -->
<property>
    <name>yarn.nodemanager.resource.pcores-vcores-multiplier</name>
    <value>1.0</value>
</property>
        <!-- NodeManager使用内存数，默认  8G，修改为  4G内存    -->
<property>
    <name>yarn.nodemanager.resource.memory-mb</name>
    <value>4096</value>
</property>
        <!-- nodemanager的  CPU核数，不按照硬件环境自动设定时默认是  8个，修改为  4个   -->
<property>
    <name>yarn.nodemanager.resource.cpu-vcores</name>
    <value>4</value>
</property>
<!-- 容器最小内存，默认  1G -->
<property>
    <name>yarn.scheduler.minimum-allocation-mb</name>
    <value>1024</value>
</property>
<!-- 容器最大内存，默认  8G，修改为  2G -->
<property>
    <name>yarn.scheduler.maximum-allocation-mb</name>
    <value>2048</value>
</property>
        <!-- 容器最小  CPU核数，默认  1个   -->
<property>
    <name>yarn.scheduler.minimum-allocation-vcores</name>
    <value>1</value>
</property>
        <!-- 容器最大  CPU核数，默认  4个，修改为  2个   -->
<property>
    <name>yarn.scheduler.maximum-allocation-vcores</name>
    <value>2</value>
</property>
        <!-- 虚拟内存检查，默认打开，修改为关闭    -->
<property>
    <name>yarn.nodemanager.vmem-check-enabled</name>
    <value>false</value>
</property>
        <!-- 虚拟内存和物理内存设置比例,默认  2.1 -->
<property>
    <name>yarn.nodemanager.vmem-pmem-ratio</name>
    <value>2.1</value>
</property>
<!--任务优先级配置-->
<property>
	<name>yarn.cluster.max-application-priority</name>
    <value>5</value>
</property>
<property>
    <name>yarn.resourcemanager.scheduler.class</name>
    <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairS cheduler</value>
    <description>配置使用公平调度器</description>
</property>
<property>
    <name>yarn.scheduler.fair.allocation.file</name>
    <value>/opt/module/hadoop-3.1.3/etc/hadoop/fair-scheduler.xml</value>
    <description>指明公平调度器队列分配配置文件</description>
</property> 
<property>
    <name>yarn.scheduler.fair.preemption</name>
    <value>false</value>
    <description>禁止队列间资源抢占</description>
</property>

capacity-scheduler.xml

<!-- 指定多队列，增加  hive队列   -->
<property>
    <name>yarn.scheduler.capacity.root.queues</name>
    <value>default,hive</value>
</property>
        <!-- 降低  default队列资源额定容量为  40%，默认  100% -->
<property>
    <name>yarn.scheduler.capacity.root.default.capacity</name>
    <value>40</value>
</property>
        <!-- 降低  default队列资源最大容量为  60%，默认  100% -->
<property>
    <name>yarn.scheduler.capacity.root.default.maximum-capacity</name>
    <value>60</value>
</property>
<!-- 指定  hive队列的资源额定容量   -->
<property>
    <name>yarn.scheduler.capacity.root.hive.capacity</name>
    <value>60</value>
</property>
        <!-- 用户最多可以使用队列多少资源，1表示   -->
<property>
    <name>yarn.scheduler.capacity.root.hive.user-limit-factor</name>
    <value>1</value>
</property>
        <!-- 指定  hive队列的资源最大容量   -->
<property>
    <name>yarn.scheduler.capacity.root.hive.maximum-capacity</name>
    <value>80</value>
</property>
        <!-- 启动  hive队列   -->
<property>
    <name>yarn.scheduler.capacity.root.hive.state</name>
    <value>RUNNING</value>
</property>
<!-- 哪些用户有权向队列提交作业   -->
<property>
    <name>yarn.scheduler.capacity.root.hive.acl_submit_applications</name>
    <value>*</value>
</property>
        <!-- 哪些用户有权操作队列，管理员权限（查看/杀死）    -->
<property>
    <name>yarn.scheduler.capacity.root.hive.acl_administer_queue</name>
    <value>*</value>
</property>
        <!-- 哪些用户有权配置提交任务优先级    -->
<property>
    <name>yarn.scheduler.capacity.root.hive.acl_application_max_priority</name>
    <value>*</value>
</property>
        <!-- 任务的超时时间设置：yarn application -appId appId -updateLifetime Timeout 参考资料：https://blog.cloudera.com/enforcing-application-lifetime-slas- yarn/ -->
        <!-- 如果  application指定了超时时间，则提交到该队列的  application能够指定的最大超时 时间不能超过该值。
        -->
<property>
    <name>yarn.scheduler.capacity.root.hive.maximum-application- lifetime</name>
    <value>-1</value>
</property>
        <!-- 如果  application没指定超时时间，则用  default-application-lifetime作为默认 值   -->
<property>
    <name>yarn.scheduler.capacity.root.hive.default-application- lifetime</name>
    <value>-1</value>
</property>

fair-scheduler.xml需手动创建

<?xml version="1.0"?>
<allocations>
    <!-- 单个队列中  Application Master占用资源的最大比例,取值  0-1 ，企业一般配置  0.1 -->
    <queueMaxAMShareDefault>0.5</queueMaxAMShareDefault> <!-- 单个队列最大资源的默认值   test atguigu default -->
    <queueMaxResourcesDefault>4096mb,4vcores</queueMaxResourcesDefault> <!-- 增加一个队列  test -->
    <queue name="test">
        <!-- 队列最小资源   -->
        <minResources>2048mb,2vcores</minResources>
        <!-- 队列最大资源   -->
        <maxResources>4096mb,4vcores</maxResources>
        <!-- 队列中最多同时运行的应用数，默认  50，根据线程数配置   -->
        <maxRunningApps>4</maxRunningApps>
        <!-- 队列中  Application Master占用资源的最大比例   -->
        <maxAMShare>0.5</maxAMShare>
        <!-- 该队列资源权重,默认值为  1.0 -->
        <weight>1.0</weight>
        <!-- 队列内部的资源分配策略   -->
        <schedulingPolicy>fair</schedulingPolicy>
    </queue>
    <!-- 增加一个队列  atguigu -->
    <queue name="atguigu" type="parent">
        <!-- 队列最小资源   -->
        <minResources>2048mb,2vcores</minResources>
        <!-- 队列最大资源   -->
        <maxResources>4096mb,4vcores</maxResources>
        <!-- 队列中最多同时运行的应用数，默认  50，根据线程数配置   -->
        <maxRunningApps>4</maxRunningApps>
        <!-- 队列中  Application Master占用资源的最大比例   -->
        <maxAMShare>0.5</maxAMShare>
        <!-- 该队列资源权重,默认值为  1.0 -->
        <weight>1.0</weight>
        <!-- 队列内部的资源分配策略   -->
        <schedulingPolicy>fair</schedulingPolicy>
    </queue>
    <!-- 任务队列分配策略,可配置多层规则,从第一个规则开始匹配,直到匹配成功   -->
    <queuePlacementPolicy>
        <!-- 提交任务时指定队列,如未指定提交队列,则继续匹配下一个规则; false表示：如果指 定队列不存在,不允许自动创建-->
        <rule name="specified" create="false"/>
        <!-- 提交到  root.group.username队列,若  root.group不存在,不允许自动创建；若 root.group.user不存在,允许自动创建   -->
        <rule name="nestedUserQueue" create="true">
            <rule name="primaryGroup" create="false"/>
        </rule>
        <!-- 最后一个规则必须为  reject或者  default。Reject表示拒绝创建提交失败， default表示把任务提交到  default队列   -->
        <rule name="reject"/>
    </queuePlacementPolicy>
</allocations>

Tool接口

兼容-D等配置参数

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 conf) {
        this.conf = conf;
    }

    @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, 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, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable value : values) {
                sum += value.get();
            }
            outV.set(sum);
            context.write(key, outV);
        }
    }
}

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);
    }
}

原文链接：https://blog.csdn.net/Ty_0026/article/details/116934133