Hadoop中Yarn和Mapreduce_yarn mapreduce demo pi

yarn

• Yet Another Resource Negotiator, 另一种资源协调者
• 通用资源管理系统
• 为上层应用提供统一的资源管理和调度，为集群在利用率、资源统一管理和数据共享等方面带来了巨大好处

YARN的架构和执行流程

ResourceManager: RM 资源管理器

1、整个集群同一时间提供服务的RM只有一个，负责集群资源的统一管理和调度， 2、处理客户端的请求： submit, kill监控我们的NM，一旦某个NM挂了，那么该NM上运行的任务需要告诉我们的AM来如何进行处理

NodeManager: NM 节点管理器

整个集群中有多个，负责自己本身节点资源管理和使用，定时向RM汇报本节点的资源使用情况，接收并处理来自RM的各种命令：启动Container处理来自AM的命令

ApplicationMaster: AM

每个应用程序对应一个：MR、Spark，负责应用程序的管理
为应用程序向RM申请资源（core、memory），分配给内部task， 需要与NM通信：启动/停止task，task是运行在container里面，AM也是运行在container里面

• Container 容器: 封装了CPU、Memory等资源的一个容器,是一个任务运行环境的抽象， Client: 提交作业 查询作业的运行进度,杀死作业

启动YARN相关的进程

sbin目录下

./ start-yarn.sh
1

验证

[hadoop@hadoop000 sbin]$ jps
13000 ResourceManager
13199 Jps
13104 NodeManager
1
2
3
4

停止

./ stop-yarn.sh
1

yarn提交MapReduce作业演示，我们接下来使用hadoop进行π值的计算
在/home/hadoop/app/hadoop-2.6.0-cdh5.7.0/share/hadoop/mapreduce找到示例jar包hadoop-mapreduce-examples-2.6.0-cdh5.7.0.jar
执行以下命令进行π值计算

[hadoop@hadoop000 mapreduce]$ hadoop jar hadoop-mapreduce-examples-2.6.0-cdh5.7.0.jar  pi 2 3
1

2 是执行任务次数 3是执行每次任务投递次数

可视化yarn和MapReduce

http://10.25.187.18:8088/cluster
1

分布式处理框架 MapReduce

什么是MapReduce

• 源于Google的MapReduce论文(2004年12月)
• Hadoop的MapReduce是Google论文的开源实现
• MapReduce优点: 海量数据离线处理&易开发
• MapReduce缺点: 实时流式计算

MapReduce编程模型

MapReduce分而治之的思想

• 数钱实例：一堆钞票，各种面值分别是多少
  • 单点策略
    • 一个人数所有的钞票，数出各种面值有多少张
  • 分治策略
    • 每个人分得一堆钞票，数出各种面值有多少张
    • 汇总，每个人负责统计一种面值
  • 解决数据可以切割进行计算的应用
• MapReduce编程分Map和Reduce阶段
  • 将作业拆分成Map阶段和Reduce阶段
  • Map阶段 Map Tasks 分：把复杂的问题分解为若干"简单的任务"
  • Reduce阶段: Reduce Tasks 合：reduce
• MapReduce编程执行步骤
  • 准备MapReduce的输入数据
  • 准备Mapper数据
  • Shuffle
  • Reduce处理
  • 结果输出
• 编程模型
  • 借鉴函数式编程方式
  • 用户只需要实现两个函数接口：
    • Map(in_key,in_value)
      —>(out_key,intermediate_value) list
    • Reduce(out_key,intermediate_value) list
      —>out_value list
  • Word Count 词频统计案例
    

Java实现WordCount

编写WordCountApp

package com.neusoft.hadoop.mapreduce;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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 java.io.FileInputStream;
import java.io.IOException;

/**
 * @author Eric Lee
 * @date 2020/10/28 11:16
 * 使用Mapreduce 开发 WordCountApp
 */
public class WordCountApp {

    /**
     * Map阶段： 读取输入文件
     */
    public  static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{
        LongWritable one = new  LongWritable(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){
                // 通过上下文把map的处理结果进行输出
                context.write(new Text(word), one);
            }

        }
    }

    /**
     * Reduce阶段： 归并操作
     */

    public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            long sum = 0;
            for (LongWritable value: values){
                // 求出key出现的次数总和
                sum += value.get();
            }
            // 最终的统计结果进行输出
            context.write(key, new LongWritable(sum));
        }
    }

    /**
     * 定义main函数 使用Driver封装Maoreduce作业的所有信息
     */
    public static void main(String[] args) throws Exception {
        // 创建Configuration对象
        Configuration configuration = new Configuration();

        // 创建Job
        Job job = Job.getInstance(configuration, "wordcount");

        // 设置job处理类
        job.setJarByClass(WordCountApp.class);
        // 设置作业的输入路径  args[0] 将命令行的第一个值传给 Path
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        // 设置 map 相关参数
        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        // 设置 reduce 相关参数
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        //     设置作业的输出路径  args[1] 将命令行的第二个值传给 Path
        FileOutputFormat.setOutputPath(job, new Path(args[2]));
        System.exit(job.waitForCompletion(true)? 0: 1);

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

打包程序

• 1. mvn打包命令，需要设置mvn环境变量

mvn clean package -DskipTests
1

• 2.使用idea的mave插件
  

上传jar包

使用 maven进行打包 hadoop_java_op_hdfs-1.0-SNAPSHOT.jar 上传到 用户目录的lib文件夹下进行命令操作
接下来在命令行下进行操作（类似计算π值）

hadoop jar 主函数全限定名  输入  输出
1

hadoop jar shenyangbig_data_hadoop-1.0-SNAPSHOT.jar com.neusoft.hadoop.mapreduce.WordCountApp  输入  输出
1

注意

相同的脚本和代码在这里执行会报错;

• 通过shell方式将输出文件夹删除

hadoop fs -rm -r /output/wc
1

参考命令

hadoop jar shenyangbig_data_hadoop-1.0-SNAPSHOT.jar com.neusoft.hadoop.mapreduce.WordCountApp hdfs://hadoop000:8020/hello.txt hdfs://hadoop000:8020/output/wc
1

打开http://192.168.1.109:8088/cluster

查看词频统计结果