3. Hadoop：MapReduce 编程及 shuffle

一、MapReduce 的运行及 WordCount 程序

1. MapReduce 运行过程

MapReduce 充分借鉴了分而治之的思想来处理海量数据，当一台机器对庞大的数据力有未逮时，便可以通过搭建 MapReduce 集群来进行数据的处理。在集群中，多台机器并行计算，执行同一逻辑任务。集群中的机器分为三类：Master、Mapper 和 Reducer。Master 是整个集群的主机，负责任务的调度，为 Mapper 和 Reducer 分配对应的 map 和 reduce 任务。map 任务被称为映射，就是对数据的每一个元素进行指定的操作。reduce 任务被称为归约，它会对 map 任务的输出进行处理从而得出最终的结果。

简单地举个例子，现在需要统计图书馆中的所有书。你数1号书架，我数2号书架。这就是“Map”。我们人越多，数书就更快。
现在我们到一起，把所有人的统计数加在一起。这就是“Reduce”。

MapReduce.jpg

在编程时，MapReduce 计算框架将数据处理的过程简化为 Map 和 Reduce 两步。假设我们有一个巨大的数据集，里面有海量规模的元素，元素的个数为 M，每个元素都需要进行同一个函数处理。于是将 M 分成许多小份，然后每一份分给一个 Mapper 来做，Mapper 执行完函数，将结果传给 Reducer。Reducer 之后统计汇总各个 Mapper 传过来的结果，得到最后的任务的答案。用户只需用 Map 和 Reduce 描述清楚需要处理的问题，即可使用 MapReduce 编程框架编写 map() 和 reduce() 函数实现分布式计算。

在 MapReduce 编程中，数据操作的最小单位是一个键值对。用户在使用 MapReduce 编程模型时，首先需要将数据抽象为键值对的形式来作为 map() 函数的输入，map() 的结果将交给 MapReduce 框架进行 shfffle 处理，shuffle 的结果将会作为 reduce() 函数的输入，再得到最终的结果。

MapReduce 中数据的变化过程如下所示：
{Key1, Value1} -> {Key2, List<Value2>} -> {Key3, Value3}
对于所有的 MapReduce 程序来说，运行的基本流程如下所示：
input -> map() -> shuffle -> reduce() ->output

以 WordCount 程序为例，输入的文本文件会被一行一行读取为{key, value}格式，key 是当前行在文件中的偏移量，value 是该行的内容，这些键值对就是 map() 函数的输入，如{0， hello world}代表文件刚开始的一行为 "hello world"。
map() 函数需要对每一行的内容按照空格分割，出现某个单词时即记录一次，得到多个键值对作为结果，如{"hello", 1}、{"world", 1}。
shuffle 是 MapReduce 框架的工作，它会对 map() 的输出结果进行处理，将相同的 key 的 value 组合成一个列表，如{"hello", [1, 1, 1]}。
reduce() 函数只需要将 value 列表中的值相加就可以得到这个单词在文本中出现的频率，最终得到形如{"hello", 3}的结果。

2. WordCount 的 Mapper 及 Reducer

了解 MapReduce 框架下 WordCount 的流程后，可以得出 Mapper 和 Reducer 分别要处理的工作。

Mapper 需要读取每行的文本，并以空格分割得到每个单词，每得到一个单词就记录一次。定义 Mapper 类时，定义其输入键值对的类型为<LongWritable, Text>，LongWritable 是 Hadoop 自定义类型，对应 Java 中的 Long，用于表示当前行在文件中的偏移量；Text 对应 Java 中的 String，表示当前行的内容。Mapper 输出的键值对类型为<Text, IntWritable>，记录某个单词出现一次。
Map 类及 map() 函数如下：

public static class WordMapper 
    extends Mapper<LongWritable, Text, Text, IntWritable> {
        
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
        
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
            throws IOException, InterruptedException {
        String line = value.toString();
        String[] words = line.split(" ");
        for (String wordStr : words) {
            word.set(wordStr);
            context.write(word, one);
        }
    }
}

Mapper 的结果经过 shuffle 之后传递给 Reducer，Reducer 得到<Text, List<IntWritable>> 类型的输入，只需将 List 中的值相加，就会得到该单词在文本中出现的总次数。需要注意的是，在定义 Reduce 类时，Reduce 类的输入类型与 Map 的输出类型相同，都是<Text, IntWritable>。但是在生成 reduce() 函数中，输入类型会转变为<Text, Iterable<IntWritable>>以便用户处理。
Reduce 类及 reduce() 函数如下：

public static class WordReducer
    extends Reducer<Text, IntWritable, Text, IntWritable> {
        
    private IntWritable outputValue = new IntWritable();
        
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values,
                Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable intWritable : values) {
            sum += intWritable.get();
        }
        outputValue.set(sum);
        context.write(key, outputValue);
    }
}

3. WordCount 完整程序

在 WordCount 程序中，WordMapper 类和 WordReducer 都被定义为内部静态类，它们继承自 Hadoop 中基础的 Mapper 和 Reducer 类，在继承时需要指定输入输出的键值对的类型。
在主函数中，我们需要定义一个 Job 并为其配置一系列参数，再指定 MapReduce 程序的输入输出路径，最后通过 job.waitForCompletion(true) 即可运行。

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;

public class MyWordCount {
    
    public static class WordMapper 
        extends Mapper<LongWritable, Text, Text, IntWritable> {
        
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
                throws IOException, InterruptedException {
            String line = value.toString();
            String[] words = line.split(" "); // 按空格分割一行文本
            for (String wordStr : words) {
                word.set(wordStr);
                context.write(word, one);
            }
        }
    }
    
    public static class WordReducer
        extends Reducer<Text, IntWritable, Text, IntWritable> {
        
        private IntWritable outputValue = new IntWritable();
        
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values,
                Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable intWritable : values) {
                sum += intWritable.get(); // 计算 List<Intwritable> 列表中的值的和
            }
            outputValue.set(sum);
            context.write(key, outputValue);
        }
    }

    public static void main(String[] args) 
            throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration, "mywordcount");
        job.setJarByClass(MyWordCount.class);
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setMapperClass(WordMapper.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setReducerClass(WordReducer.class);
        
        FileInputFormat.addInputPath(job, new Path("/datas/test/mapred/example1/test_input"));
        FileOutputFormat.setOutputPath(job, new Path("/datas/test/mapred/result4-16"));
        
        boolean isSuccess = job.waitForCompletion(true);
        System.exit(isSuccess ? 0 : 1);
    }

}

二、shuffle 过程解析

MapReduce 计算模型主要由三部分组成：Map、Shuffle、Reduce，在编程时，Map 和 Reduce 由用户设计，而将 Map 的输出进行进一步处理再交给 Reduce 的过程就是 Shuffle。
MapReduce 整个流程大致如下：

shuffle.jpg

Spill 的关键操作有三步：partition、sort、group，分别为分区、排序和分组，这三步操作都与 map 中输出数据的 key 有关。

Map 操作结束后进入 Shuffle，首先将 Map 端结果（如<"hello", 1>）写入内存中，该内存区默认大小为100MB，是一个环形的缓冲区。当内存使用到80%时，内存中的数据会被 spill 到本地磁盘中。这里的本地磁盘指的是 map task 所运行的 NodeManager 机器的本地磁盘。

partition 依据 reduce task 的任务数进行分区，决定 map 输出的数据将会被哪个 reduce task 处理。sort 操作会对分区中数据进行排序。最后分区中排序完的数据会被写到本地磁盘的一个文件中。
以上操作会反复进行，当 map 处理数据结束之后，需要将本地磁盘的文件合并成一个文件，此时各个分区中的数据已经进行了排序。

Reduce 的输入数据拷贝自 Map 的输出，map task 处理数据结束之后会通知 Master，Master 将会告知 reduce task，Reducer 会自动去读取需要处理的数据。当 Reducer 获得所有数据之后，对数据进行 group，将相同的 key 的 value 合并，得到如<"hello", [1, 1, 1]>的数据，这就是用户编写的 reduce() 函数的输入。

默认情况下，reduce task 的个数为1，这是所有的 map 输出都会被分配到同一个 reduce task 中，此时所有的输出都是有序的。如果需要多个 reduce task，可以通过job.setNumReduceTasks()进行设置。如果将 reduce task 的数目设置为0，则表示整个 MapReduce Job 只有 map task，这时仅仅使用并行计算功能对数据进行操作，不需要聚合。