MapReduce概述

定义

MapReduce是一个分布式运算程序的编程框架，是用户开发“基于Hadoop的数据分析应用”的核心框架
核心功能，就是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个Hadoop集群上

优缺点

优点：
- 易于编程：简单的实现一些接口，就可以完成一个分布式程序。
- 良好的扩展性：当你的计算机资源不能得到满足的时候，可以通过简单的增加机器来扩展它的计算能力。
- 高容错性：假设一台机器挂了，它可以把上面的计算任务转移到到另一个正常的节点上运行。
- 适合PB级以上海量数据的离线处理
缺点：
- 不擅长实时计算：MapReduce无法像MySQL一样，在毫秒或者描记内返回结果
- 不擅长流式计算：流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的，不能动态变化的。
- 不擅长DAG计算：多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce 并不是不能做，而是使用后，每个MapReduce 作业的输出结果都会写入到磁盘，会造成大量的磁盘 IO，导致性能非常的低下。

核心思想

MapReduce运行程序一般需要分成2个阶段：Map阶段和Reduce阶段
Map阶段的并发MapTask，完全并行运行，互不相干。并行运行主要是将数据源分块进行计算，就比如任务一就计算数据源前100M的数据，任务二就计算数据源后100M的数据
Reduce阶段的并发ReduceTask，完全互不相干，但是他们的数据依赖于上一个阶段的输出
MapReduce编程模型只能包含一个Map阶段和一个Reduce阶段，如果用户业务逻辑复杂，那就只能多个MapReduce程序，串行运行

进程

一个完整的 MapReduce 程序在分布式运行时有三类实例进程：
- MrAppMaster：负责整个程序的过程调度及状态协调。
- MapTask：负责 Map 阶段的整个数据处理流程。
- ReduceTask：负责 Reduce 阶段的整个数据处理流程。

常用数据序列化类型

在这里插入图片描述

MapReduce编程规范

Mapper阶段
- 用户自定义的Mapper要继承自己的父类
- Mapper的输入数据是KV对的形式（KV的类型可自定义）
- Mapper中的业务逻辑写在map()方法中
- Mapper的输出数据是KV对的形式（KV的类型可自定义）
- map()方法（MapTask进程）对每一个<K,V>调用一次
Reducer阶段
- 用户自定义的Reducer要继承自己的父类
- Reducer的输入数据类型对应Mapper的输出数据类型，也是KV
- Reducer的业务逻辑写在reduce()方法中
- ReduceTask进程对每一组相同k的<k,v>组调用一次reduce()方法
Driver阶段
- 相当于YARN集群的客户端，用于提交我们整个程序到YARN集群，提交的是
  封装了MapReduce程序相关运行参数的job对象

实操

首先通过图来简单演示一下本地案例

搭建环境

创建maven工程，MapReduceDemo
在pom.xml文件中添加依赖：

<dependencies>
 <dependency>
 <groupId>org.apache.hadoop</groupId>
 <artifactId>hadoop-client</artifactId>
 <version>3.1.3</version>
 </dependency>
 <dependency>
 <groupId>junit</groupId>
 <artifactId>junit</artifactId>
 <version>4.12</version>
 </dependency>
 <dependency>
 <groupId>org.slf4j</groupId>
 <artifactId>slf4j-log4j12</artifactId>
 <version>1.7.30</version>
 </dependency>
</dependencies>

在项目的 src/main/resources 目录下，新建一个文件，命名为“log4j.properties”，在文件中填入。

log4j.rootLogger=INFO, stdout 
log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n 
log4j.appender.logfile=org.apache.log4j.FileAppender 
log4j.appender.logfile.File=target/spring.log 
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout 
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

编写程序

编写WordCountMapper类，下图是mapper类的相关介绍，代码是我们需要重写的部分

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

//KEYIN,map阶段输入的KEY的类型：LongWritable 是偏移量
//VALUEIN，map阶段输入的VALUE的类型，Text
//KEYOUT, map阶段输出的KEY类型：TEXT 标识单词
//VALUEOUT，map阶段输出的VALUE类型，IntWritable 单词次数
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    //创建封装类
    private Text outKey = new Text();
    private IntWritable intWritable = 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) {
            //4. 封装数据
            outKey.set(word);
            System.out.println("map阶段的Key：" + outKey);
            context.write(outKey, intWritable);
        }
    }
}

编写WordCountReducer类，下图是源码解析，代码是自己重写的业务逻辑

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

//KEYIN,reduce阶段输入的KEY的类型：Text
//VALUEIN，reduce阶段输入的VALUE的类型，IntWritable
//KEYOUT, reduce阶段输出的KEY类型：TEXT 标识单词
//VALUEOUT，reduce阶段输出的VALUE类型，IntWritable 单词次数
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    //创建封装类型
    private IntWritable intWritable = new IntWritable();
    
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        super.reduce(key, values, context);
        //创建累加的值
        int sum = 0;

        //循环累加
        for (IntWritable value:values) {
            sum += value.get();
        }
        
        //封装数据
        intWritable.set(sum);
        context.write(key,intWritable);
    }
}

编写WordCountDriver类

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class WordCountDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //1.获取job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        //2.设置jar包路径
        job.setJarByClass(WordCountDriver.class);

        //3.关联mapper和reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        //4.设置map的输出的KV类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5.设置最终输出的KV类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //6.设置输入路径和输出路径
        FileInputFormat.setInputPaths(job,new Path("D:\\input"));
        FileOutputFormat.setOutputPath(job,new Path("D:\\output"));

        //7.提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0: 1);
    }
}

测试一下

Hadoop序列化

序列化概述

什么是序列化？
- 序列化就是把内存中的对象转换成字节序列以便于存储到磁盘和网络传输。
- 反序列化就是将收到字节序列或者是磁盘的持久化数据转换成内存中的对象
为什么要序列化？
- 一般来说，“活的”对象只生存在内存里，关机断电就没有了。而且“活的”对象只能由本地的进程使用，不能被发送到网络上的另外一台计算机。然而序列化可以存储“活的”对象，可以将“活的”对象发送到远程计算机。
为什么不使用JAVA的序列化
- Java 的序列化是一个重量级序列化框架（Serializable），一个对象被序列化后，会附带很多额外的信息（各种校验信息，Header，继承体系等），不便于在网络中高效传输。所以，Hadoop 自己开发了一套序列化机制（Writable）。
Hadoop序列化特点：
- 紧凑：高效使用存储空间
- 快速：读写数据的额外开销小
- 互操作：支持多语言的交互