本篇内容主要讲解“Hadoop MapReduce基本原理是什么”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“Hadoop MapReduce基本原理是什么”吧!

专注于为中小企业提供网站设计制作、网站制作服务,电脑端+手机端+微信端的三站合一,更高效的管理,为中小企业泉港免费做网站提供优质的服务。我们立足成都，凝聚了一批互联网行业人才，有力地推动了近1000家企业的稳健成长，帮助中小企业通过网站建设实现规模扩充和转变。

一、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、流程描述

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

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

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

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

2、基础配置

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

3、Mapper程序

public class WordMapper extends Mapper {
    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 {

    int sum ;
    IntWritable value = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable 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

三、案例分析

1、数据类型

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

Java类型	Writable类型	Java类型	Writable类型
String	Text	float	FloatWritable
int	IntWritable	long	LongWritable
boolean	BooleanWritable	double	DoubleWritable
byte	ByteWritable	array	DoubleWritable
map	MapWritable

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 {

    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 {

    @Override
    protected void reduce(Text key, Iterable 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基本原理是什么”有了更深的了解，不妨来实际操作一番吧！这里是创新互联网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！

网页名称：HadoopMapReduce基本原理是什么
分享链接：http://pwwzsj.com/article/pghhej.html