MapReduce定义

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

MapReduce的优缺点

优点

1. 易于编程，用户只关心业务逻辑，实现框架的接口
2. 良好的扩展性，可以动态的增加服务器，解决计算资源不够的问题
3. 高容错性，任何一台挂掉，可以将任务转移到其他节点
4. 适合海量数据计算（TB/PB级别），几千台服务器共同计算

缺点

1. 不擅长实时计算
2. 不擅长流式计算
3. 不擅长DAG有向无环图计算

MapReduce核心思想

image.png

MapReduce的进程

一个完整的MapReduce程序在分布式环境下有三类实例进程

1. MrAppMaster：负责整个程序的过程调度及状态协调
2. MapTask：负责Map阶段整个数据处理流程
3. ReduceTask：负责Reduce阶段整个数据处理流程

官方WordCount源码

采用反编译工具反编译源码，发现WordCount案例有Map类、Reduce类、驱动类。且数据类型是Hadoop封装的序列化类型

image.png

image.png

常用数据序列化类型

image.png

MapReduce编程规范

用户编写的程序分为三个部分Map、Reduce、Driver

Map阶段

1. 用户自定义的Mapper要继承父类
2. Mapper的输入数据是KV对的形式
3. Mapper中的业务逻辑写在map方法里
4. Mapper的输出数据是KV对的形式
5. map方法（MapTask进程）对每个KV对调用一次

Reduce阶段

1. 用户自定义的Reducer要继承父类
2. Reducer的输入要对应Mapper的输出，KV对
3. Reducer的业务逻辑写在reduce方法里
4. reduce方法（ReduceTask进程）对每一组相同K的KV对调用一次

Driver阶段

相当于YARN集群的客户端，用于提交我们整个程序到YARN集群，提交的是封装了MapReduce程序相关运行参数的job对象

模拟官网WordCount案例

测试数据

a b c
s d
c
c
r
a
d s
r t
h e

Mapper

/**
 * KEYIN, map阶段输入的key的类型 long
 * VALUEIN, map阶段输入的value的类型 text
 * KEYOUT, map阶段输出的key的类型 text
 * VALUEOUT, map阶段输出的value的类型 int
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private Text outKey = new Text();
    private IntWritable outValue = new IntWritable(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) {
            // 封装outKey
            outKey.set(word);
            // 写出
            context.write(outKey, outValue);
        }
    }
}

Reducer

/**
 * KEYIN, reduce阶段输入的key的类型 long
 * VALUEIN, reduce阶段输入的value的类型 text
 * KEYOUT, reduce阶段输出的key的类型 text
 * VALUEOUT, reduce阶段输出的value的类型 int
 */
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable outValue = new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        // 累加
        for (IntWritable value : values) {
            sum += value.get();
        }
        outValue.set(sum);

        context.write(key, outValue);
    }
}

Driver

public class WordCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 获取job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        // 设置jar路径
        job.setJarByClass(WordCountDriver.class);
        // 关联mapper、reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        // 设置map输出的KV类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        // 设置最终输出的KV类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 设置输入路径和输出路径
        FileInputFormat.setInputPaths(job, new Path("test.txt"));
        FileOutputFormat.setOutputPath(job, new Path("result.txt"));
        // 提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}