hadoop基本操作2

2020-10-28 本文已影响0人 __method__

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

验证

[hadoop@hadoop000 sbin]$ jps
13000 ResourceManager
13199 Jps
13104 NodeManager

停止

./ stop-yarn.sh

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

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

可视化yarn和MapReduce

http://10.25.187.18:8088/cluster

分布式处理框架 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);

    }
}

使用 maven进行打包 hadoop_java_op_hdfs-1.0-SNAPSHOT.jar 上传到用户目录的lib文件夹下进行命令操作```