您当前的位置:首页 > 电脑百科 > 程序开发 > 框架

Hadoop框架:MapReduce基本原理和入门案例

时间:2020-11-23 13:32:23  来源:  作者:

一、MapReduce概述

1、基本概念

Hadoop核心组件之一:分布式计算的方案MapReduce,是一种编程模型,用于大规模数据集的并行运算,其中Map(映射)和Reduce(归约)。

MapReduce既是一个编程模型,也是一个计算组件,处理的过程分为两个阶段,Map阶段:负责把任务分解为多个小任务,Reduce负责把多个小任务的处理结果进行汇总。其中Map阶段主要输入是一对Key-Value,经过map计算后输出一对Key-Value值;然后将相同Key合并,形成Key-Value集合;再将这个Key-Value集合转入Reduce阶段,经过计算输出最终Key-Value结果集。

2、特点描述

MapReduce可以实现基于上千台服务器并发工作,提供很强大的数据处理能力,如果其中单台服务挂掉,计算任务会自动转义到另外节点执行,保证高容错性;但是MapReduce不适应于实时计算与流式计算,计算的数据是静态的。

二、操作案例

1、流程描述

Hadoop框架:MapReduce基本原理和入门案例

 

数据文件一般以CSV格式居多,数据行通常以空格分隔,这里需要考虑数据内容特点;

文件经过切片分配在不同的MapTask任务中并发执行;

MapTask任务执行完毕之后,执行ReduceTask任务,依赖Map阶段的数据;

ReduceTask任务执行完毕后,输出文件结果。

2、基础配置

hadoop:
  # 读取的文件源
  inputPath: hdfs://hop01:9000/hopdir/JAVANew.txt
  # 该路径必须是程序运行前不存在的
  outputPath: /wordOut

3、MApper程序

public class WordMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    Text mapKey = new Text();
    IntWritable mapValue = new IntWritable(1);

    @Override
    protected void map (LongWritable key, Text value, Context context)
                        throws IOException, InterruptedException {
        // 1、读取行
        String line = value.toString();
        // 2、行内容切割,根据文件中分隔符
        String[] words = line.split(" ");
        // 3、存储
        for (String word : words) {
            mapKey.set(word);
            context.write(mapKey, mapValue);
        }
    }
}

4、Reducer程序

public class WordReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    int sum ;
    IntWritable value = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values,Context context)
                        throws IOException, InterruptedException {
        // 1、累加求和统计
        sum = 0;
        for (IntWritable count : values) {
            sum += count.get();
        }
        // 2、输出结果
        value.set(sum);
        context.write(key,value);
    }
}

5、执行程序

@RestController
public class WordWeb {

    @Resource
    private MapReduceConfig mapReduceConfig ;

    @GetMapping("/getWord")
    public String getWord () throws IOException, ClassNotFoundException, InterruptedException {
        // 声明配置
        Configuration hadoopConfig = new Configuration();
        hadoopConfig.set("fs.hdfs.impl",
                org.Apache.hadoop.hdfs.DistributedFileSystem.class.getName()
        );
        hadoopConfig.set("fs.file.impl",
                org.apache.hadoop.fs.LocalFileSystem.class.getName()
        );
        Job job = Job.getInstance(hadoopConfig);

        // Job执行作业 输入路径
        FileInputFormat.addInputPath(job, new Path(mapReduceConfig.getInputPath()));
        // Job执行作业 输出路径
        FileOutputFormat.setOutputPath(job, new Path(mapReduceConfig.getOutputPath()));

        // 自定义 Mapper和Reducer 两个阶段的任务处理类
        job.setMapperClass(WordMapper.class);
        job.setReducerClass(WordReducer.class);

        // 设置输出结果的Key和Value的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //执行Job直到完成
        job.waitForCompletion(true);
        return "success" ;
    }
}

6、执行结果查看

将应用程序打包放到hop01服务上执行;

java -jar map-reduce-case01.jar
Hadoop框架:MapReduce基本原理和入门案例

 

三、案例分析

1、数据类型

Java数据类型与对应的Hadoop数据序列化类型;

Hadoop框架:MapReduce基本原理和入门案例

 

2、核心模块

Mapper模块:处理输入的数据,业务逻辑在map()方法中完成,输出的数据也是KV格式;

Reducer模块:处理Map程序输出的KV数据,业务逻辑在reduce()方法中;

Driver模块:将程序提交到yarn进行调度,提交封装了运行参数的job对象;

四、序列化操作

1、序列化简介

序列化:将内存中对象转换为二进制的字节序列,可以通过输出流持久化存储或者网络传输;

反序列化:接收输入字节流或者读取磁盘持久化的数据,加载到内存的对象过程;

Hadoop序列化相关接口:Writable实现的序列化机制、Comparable管理Key的排序问题;

2、案例实现

案例描述:读取文件,并对文件相同的行做数据累加计算,输出计算结果;该案例演示在本地执行,不把Jar包上传的hadoop服务器,驱动配置一致。

实体对象属性

public class AddEntity implements Writable {

    private long addNum01;
    private long addNum02;
    private long resNum;

    // 构造方法
    public AddEntity() {
        super();
    }
    public AddEntity(long addNum01, long addNum02) {
        super();
        this.addNum01 = addNum01;
        this.addNum02 = addNum02;
        this.resNum = addNum01 + addNum02;
    }

    // 序列化
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeLong(addNum01);
        dataOutput.writeLong(addNum02);
        dataOutput.writeLong(resNum);
    }
    // 反序列化
    @Override
    public void readFields(DataInput dataInput) throws IOException {
        // 注意:反序列化顺序和写序列化顺序一致
        this.addNum01  = dataInput.readLong();
        this.addNum02 = dataInput.readLong();
        this.resNum = dataInput.readLong();
    }
    // 省略Get和Set方法
}

Mapper机制

public class AddMapper extends Mapper<LongWritable, Text, Text, AddEntity> {

    Text myKey = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        // 读取行
        String line = value.toString();

        // 行内容切割
        String[] lineArr = line.split(",");

        // 内容格式处理
        String lineNum = lineArr[0];
        long addNum01 = Long.parseLong(lineArr[1]);
        long addNum02 = Long.parseLong(lineArr[2]);

        myKey.set(lineNum);
        AddEntity myValue = new AddEntity(addNum01,addNum02);

        // 输出
        context.write(myKey, myValue);
    }
}

Reducer机制

public class AddReducer extends Reducer<Text, AddEntity, Text, AddEntity> {

    @Override
    protected void reduce(Text key, Iterable<AddEntity> values, Context context)
            throws IOException, InterruptedException {

        long addNum01Sum = 0;
        long addNum02Sum = 0;

        // 处理Key相同
        for (AddEntity addEntity : values) {
            addNum01Sum += addEntity.getAddNum01();
            addNum02Sum += addEntity.getAddNum02();
        }

        // 最终输出
        AddEntity addRes = new AddEntity(addNum01Sum, addNum02Sum);
        context.write(key, addRes);
    }
}

案例最终结果:

Hadoop框架:MapReduce基本原理和入门案例


Tags:MapReduce   点击:()  评论:()
声明:本站部分内容及图片来自互联网,转载是出于传递更多信息之目的,内容观点仅代表作者本人,如有任何标注错误或版权侵犯请与我们联系(Email:2595517585@qq.com),我们将及时更正、删除,谢谢。
▌相关推荐
一、MapReduce概述1、基本概念Hadoop核心组件之一:分布式计算的方案MapReduce,是一种编程模型,用于大规模数据集的并行运算,其中Map(映射)和Reduce(归约)。MapReduce既是一个编程模...【详细内容】
2020-11-23  Tags: MapReduce  点击:(136)  评论:(0)  加入收藏
前言我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。这篇文章的内容,更多地主要是描述处理/交互流程性的东西,大部分流程图都是经过我梳理后画出来的(开始我打算使用序...【详细内容】
2020-07-08  Tags: MapReduce  点击:(98)  评论:(0)  加入收藏
什么是MapReduce分布式离线的计算框架,是一个分布式运算程序的编程框架,是用户开发基于hadoop的数据分析应用的核心框架,mapreduce的核心功能是将用户编写的业务逻辑代码和自带...【详细内容】
2019-11-12  Tags: MapReduce  点击:(96)  评论:(0)  加入收藏
▌简易百科推荐
近日只是为了想尽办法为 Flask 实现 Swagger UI 文档功能,基本上要让 Flask 配合 Flasgger, 所以写了篇 Flask 应用集成 Swagger UI 。然而不断的 Google 过程中偶然间发现了...【详细内容】
2021-12-23  Python阿杰    Tags:FastAPI   点击:(6)  评论:(0)  加入收藏
文章目录1、Quartz1.1 引入依赖<dependency> <groupId>org.quartz-scheduler</groupId> <artifactId>quartz</artifactId> <version>2.3.2</version></dependency>...【详细内容】
2021-12-22  java老人头    Tags:框架   点击:(11)  评论:(0)  加入收藏
今天来梳理下 Spring 的整体脉络啦,为后面的文章做个铺垫~后面几篇文章应该会讲讲这些内容啦 Spring AOP 插件 (了好久都忘了 ) 分享下 4ye 在项目中利用 AOP + MybatisPlus 对...【详细内容】
2021-12-07  Java4ye    Tags:Spring   点击:(14)  评论:(0)  加入收藏
&emsp;前面通过入门案例介绍,我们发现在SpringSecurity中如果我们没有使用自定义的登录界面,那么SpringSecurity会给我们提供一个系统登录界面。但真实项目中我们一般都会使用...【详细内容】
2021-12-06  波哥带你学Java    Tags:SpringSecurity   点击:(18)  评论:(0)  加入收藏
React 简介 React 基本使用<div id="test"></div><script type="text/javascript" src="../js/react.development.js"></script><script type="text/javascript" src="../js...【详细内容】
2021-11-30  清闲的帆船先生    Tags:框架   点击:(19)  评论:(0)  加入收藏
流水线(Pipeline)是把一个重复的过程分解为若干个子过程,使每个子过程与其他子过程并行进行的技术。本文主要介绍了诞生于云原生时代的流水线框架 Argo。 什么是流水线?在计算机...【详细内容】
2021-11-30  叼着猫的鱼    Tags:框架   点击:(21)  评论:(0)  加入收藏
TKinterThinter 是标准的python包,你可以在linx,macos,windows上使用它,你不需要安装它,因为它是python自带的扩展包。 它采用TCL的控制接口,你可以非常方便地写出图形界面,如...【详细内容】
2021-11-30    梦回故里归来  Tags:框架   点击:(26)  评论:(0)  加入收藏
前言项目中的配置文件会有密码的存在,例如数据库的密码、邮箱的密码、FTP的密码等。配置的密码以明文的方式暴露,并不是一种安全的方式,特别是大型项目的生产环境中,因为配置文...【详细内容】
2021-11-17  充满元气的java爱好者  博客园  Tags:SpringBoot   点击:(25)  评论:(0)  加入收藏
一、搭建环境1、创建数据库表和表结构create table account(id INT identity(1,1) primary key,name varchar(20),[money] DECIMAL2、创建maven的工程SSM,在pom.xml文件引入...【详细内容】
2021-11-11  AT小白在线中  搜狐号  Tags:开发框架   点击:(29)  评论:(0)  加入收藏
SpringBoot开发的物联网通信平台系统项目功能模块 功能 说明 MQTT 1.SSL支持 2.集群化部署时暂不支持retain&will类型消 UDP ...【详细内容】
2021-11-05  小程序建站    Tags:SpringBoot   点击:(55)  评论:(0)  加入收藏
最新更新
栏目热门
栏目头条