Hadoop【基础知识 02】【分布式计算框架MapReduce核心概念+编程模型+combiner&partitioner+词频统计案例解析与进阶+作业的生命周期】（图片来源于网络）

1. 概述

同 HDFS 一样，Hadoop MapReduce 也采用了 Master/Slave（M/S）架构，具体如图所示。它主要由以下几个组件组成：Client、JobTracker、TaskTracker 和 Task。 下面分别对这几个组件进行介绍。

Client

我们将编写的 MapReduce 程序通过 Client 提交到 JobTracker 端；同时也可通过 Client 提供的一些接口查看作业运行状态。在 Hadoop 内部用 “作业”（Job） 表示 MapReduce 程序。一个 MapReduce 程序可对应若干个作业，而每个作业会被分解成若干个 Map/Reduce 任务（Task）。

JobTracker

JobTracker 主要负责资源监控和作业调度。JobTracker 监控所有 TaskTracker 与作业的健康状况，一旦发现失败情况后，其会将相应的任务转移到其他节点；同时 JobTracker 会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器，而调度器会在资源出现空闲时，选择合适的任务使用这些资源。在 Hadoop 中，任务调度器是一个可插拔的模块，用户可以根据自己的需要设计相应的调度器。

TaskTracker

TaskTracker 会周期性地通过 Heartbeat 将本节点上资源的使用情况和任务的运行进度汇报给JobTracker， 同时接收 JobTracker 发送过来的命令并执行相应的操作（如启动新任务、 杀死任务等）。TaskTracker 使用“slot” 等量划分本节点上的资源量。“slot” 代表计算资源（CPU、内存等）。一个 Task 获取到一个 slot 后才有机会运行，而 Hadoop 调度器的作用就是将各个TaskTracker 上的空闲 slot 分配给 Task 使用。 slot 分为 Map slot 和 Reduce slot 两种，分别供MapTask 和 ReduceTask 使用。 TaskTracker 通过 slot 数目（可配置参数）限定 Task 的并发度。

Task

Task 分为 Map Task 和 Reduce Task 两种， 均由 TaskTracker 启动。 HDFS 以固定大小的 block为基本单位存储数据， 而对于 MapReduce 而言， 其处理单位是 split。split 与 block 的对应关系如图所示。 split 是一个逻辑概念， 它只包含一些元数据信息， 比如数据起始位置、数据长度、数据所在节点等。它的划分方法完全由用户自己决定。 但需要注意的是，split 的多少决定了 Map Task 的数目 ，因为每个 split 会交由一个 Map Task 处理。

Map Task 先将对应的 split 迭代解析成一个个key/value 对，依次调用用户自定义的 map() 函数进行处理，最终将临时结果存放到本地磁盘上，其中临时数据被分成若干个 partition，每个 partition 将被一个 Reduce Task 处理。

ReduceTask执行过程

该过程分为三个阶段

1. 从远程节点上读取 MapTask 中间结果（称为“Shuffle 阶段”）；
2. 按照 key 对 key/value 对进行排序（称为“Sort 阶段”）；
3. 依次读取，调用用户自定义的 reduce() 函数处理，并将最终结果存到 HDFS上（称为“ Reduce 阶段）

Hadoop MapReduce 是一个分布式计算框架，用于编写批处理应用程序。编写好的程序可以提交到Hadoop 集群上用于并行处理大规模的数据集。

MapReduce 作业通过将输入的数据集拆分为独立的块，这些块由 map 以并行的方式处理，框架对map 的输出进行排序，然后输入到 reduce 中。MapReduce 框架专门用于 键值对处理，它将作业的输入视为一组 对，并生成一组 对作为输出。输出和输出的 key 和 value 都必须实现Writable 接口。

(input) <k1, v1> -> map -> <k2, v2> -> combine -> <k2, v2> -> reduce -> <k3, v3> (output)

2. 编程模型简述

这里以词频统计为例进行说明，MapReduce 处理的流程如下【两张图的侧重点不同】：

1. input : 读取文本文件；
2. splitting : 将文件按照行进行拆分，此时得到的 K1 行数， V1 表示对应行的文本内容；
3. mapping : 并行将每一行按照空格进行拆分，拆分得到的 List(K2,V2) ，其中 K2 代表每一个单词，由于是做词频统计，所以 V2 的值为 1，代表出现 1 次；
4. shuffling：由于 Mapping 操作可能是在不同的机器上并行处理的，所以需要通过 shuffling 将相同 key 值的数据分发到同一个节点上去合并，这样才能统计出最终的结果，此时得到 K2 为每一个单词， List(V2) 为可迭代集合， V2 就是 Mapping 中的 V2；
5. Reducing : 这里的案例是统计单词出现的总次数，所以 Reducing 对 List(V2) 进行归约求和操作，最终输出。MapReduce 编程模型中 splitting 和 shuffing 操作都是由框架实现的，需要我们自己编程实现的只有 mapping 和 reducing ，这也就是 MapReduce 这个称呼的来源。

3. combiner & partitioner

3.1 InputFormat & RecordReaders

InputFormat 将输出文件拆分为多个 InputSplit ，并由 RecordReaders 将 InputSplit 转换为标准的键值对，作为 map 的输出。这一步的意义在于只有先进行逻辑拆分并转为标准的键值对格式后，才能为多个 map 提供输入，以便进行并行处理。

3.2 Combiner

combiner 是 map 运算后的可选操作，它实际上是一个本地化的 reduce 操作，它主要是在 map 计算出中间文件后做一个简单的合并重复 key 值的操作。这里以词频统计为例：

map 在遇到一个 hadoop 的单词时就会记录为 1，但是这篇文章里 hadoop 可能会出现 n 多次，那么 map 输出文件冗余就会很多，因此在 reduce 计算前对相同的 key 做一个合并操作，那么需要传输的数据量就会减少，传输效率就可以得到提升。

但并非所有场景都适合使用 combiner ，使用它的原则是 combiner 的输出不会影响到 reduce 计算的最终输入，例如：求总数，最大值，最小值时都可以使用 combiner ，但是做平均值计算则不能使用 combiner 。

不使用 combiner 的情况：

使用 combiner 的情况：

可以看到使用 combiner 的时候，需要传输到 reducer 中的数据由 12keys，降低到 10keys。降低的幅度取决于你 keys 的重复率，下文词频统计案例会演示用 combiner 降低数百倍的传输量。

3.3 Partitioner

partitioner 可以理解成分类器，将 map 的输出按照 key 值的不同分别分给对应的 reducer ，支持自定义实现，下文案例会给出演示。

4. MapReduce词频统计案例

4.1 项目简介

这里给出一个经典的词频统计的案例：统计如下样本数据中每个单词出现的次数。

Spark HBase
Hive Flink Storm Hadoop HBase Spark
Flink
HBase Storm
HBase Hadoop Hive Flink
HBase Flink Hive Storm
Hive Flink Hadoop
HBase Hive
Hadoop Spark HBase Storm
HBase Hadoop Hive Flink
HBase Flink Hive Storm
Hive Flink Hadoop
HBase Hive

为方便大家开发，我在项目源码中放置了一个工具类 WordCountDataUtils ，用于模拟产生词频统计的样本，生成的文件支持输出到本地或者直接写到 HDFS 上。

项目完整源码下载地址： hadoop-word-count

4.2 项目依赖

想要进行 MapReduce 编程，需要导入 hadoop-client 依赖：

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-client</artifactId>
  <version>${hadoop.version}</version>
</dependency>

4.3 WordCountMapper

将每行数据按照指定分隔符进行拆分。这里需要注意在 MapReduce 中必须使用 Hadoop 定义的类型，因为 Hadoop 预定义的类型都是可序列化，可比较的，所有类型均实现了 WritableComparable 接口。

public class WordCountMapperV2 extends Mapper<LongWritable, Text, Text, IntWritable> {
   
   
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
   
   
        String[] words = value.toString().split("\t");
        for (String word : words) {
   
   
            context.write(new Text(word), new IntWritable(1));
        }
    }
}

WordCountMapper 对应下图的 Mapping 操作：

WordCountMapper 继承自 Mappe 类，这是一个泛型类，定义如下：

WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> 
    public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
   
   
 ......
}

• KEYIN : mapping 输入 key 的类型，即每行的偏移量 (每行第一个字符在整个文本中的位置)，Long 类型，对应 Hadoop 中的 LongWritable 类型；
• VALUEIN : mapping 输入 value 的类型，即每行数据； String 类型，对应 Hadoop 中 Text 类型；
• KEYOUT ： mapping 输出的 key 的类型，即每个单词； String 类型，对应 Hadoop 中 Text类型；
• VALUEOUT： mapping 输出 value 的类型，即每个单词出现的次数；这里用 int 类型，对应IntWritable 类型。

4.4 WordCountReducer

在 Reduce 中进行单词出现次数的统计：

public class WordCountReducerV2 extends Reducer<Text, IntWritable, Text, IntWritable> {
   
   
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
   
   
        int count = 0;
        for (IntWritable value : values) {
   
   
            count += value.get();
        }
        context.write(new Text(key), new IntWritable(count));
    }
}

如下图， shuffling 的输出是 reduce 的输入。这里的 key 是每个单词，values 是一个可迭代的数据
类型，类似 (1,1,1,...) 。

4.5 WordCountApp

组装 MapReduce 作业，并提交到服务器运行，代码如下：

public class WordCountJobV2 extends Configured implements Tool {
   
   

    public static final String HDFS_URL = "hdfs://aliyun:8020";
    private static final String HADOOP_USER_NAME = "root";

    public static void main(String[] args) throws Exception {
   
   
        //执行job作业任务类的对象是谁
        ToolRunner.run(new WordCountJobV2(), args);
    }

    @Override
    public int run(String[] args) throws Exception {
   
   
        int argsNum = 2;
        // 文件输入路径和输出路径由外部传参指定
        if (args.length < argsNum) {
   
   
            System.out.println("Input and output paths are necessary!");
            return 0;
        }
        System.setProperty("HADOOP_USER_NAME", HADOOP_USER_NAME);
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", HDFS_URL);

        // 创建job
        Job job = Job.getInstance(configuration);
        job.setJarByClass(WordCountJobV2.class);

        // 设置map和reduce
        job.setMapperClass(WordCountMapperV2.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setReducerClass(WordCountReducerV2.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 设置 inputFormat 和 outputFormat
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path(args[0]));

        // 如果输出目录已经存在，则必须先删除，否则重复运行程序时会抛出异常
        FileSystem fileSystem = FileSystem.get(new URI(HDFS_URL), configuration, HADOOP_USER_NAME);
        Path outputPath = new Path(args[1]);
        if (fileSystem.exists(outputPath)) {
   
   
            fileSystem.delete(outputPath, true);
        }
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, outputPath);

        // 将作业提交到群集并等待它完成，参数设置为 true 代表打印显示对应的进度
        boolean completion = job.waitForCompletion(true);

        // 关闭之前创建的 fileSystem
        fileSystem.close();

        // 根据作业结果,终止当前运行的 Java 虚拟机,退出程序
        System.exit(completion ? 0 : -1);
        return completion ? 0 : -1;
    }
}

需要注意的是：如果不设置 Mapper 操作的输出类型，则程序默认它和 Reducer 操作输出的类型相同。

4.6 提交到服务器运行

在实际开发中，可以在本机配置 hadoop 开发环境，直接在 IDE 中启动进行测试。这里主要介绍一下打包提交到服务器运行。由于本项目没有使用除 Hadoop 外的第三方依赖，直接打包即可：

# mvn clean package

使用以下命令提交作业：

hadoop jar /usr/appjar/hadoop-word-count-1.0.jar \
com.heibaiying.WordCountApp \
/wordcount/input.txt /wordcount/output/WordCountApp

作业完成后查看 HDFS 上生成目录：

# 查看目录
hadoop fs -ls /wordcount/output/WordCountApp
# 查看统计结果
hadoop fs -cat /wordcount/output/WordCountApp/part-r-00000

5 词频统计案例进阶之Combiner

5.1 代码实现

想要使用 combiner 功能只要在组装作业时，添加下面一行代码即可：

// 设置 Combiner
job.setCombinerClass(WordCountReducer.class);

5.2 执行结果

加入 combiner 后统计结果是不会有变化的，但是可以从打印的日志看出 combiner 的效果：

没有加入 combiner 的打印日志：

加入 combiner 后的打印日志如下：

这里我们只有一个输入文件并且小于 128M，所以只有一个 Map 进行处理。可以看到经过 combiner 后，records 由 3519 降低为 6 (样本中单词种类就只有 6 种)，在这个用例中 combiner 就能极大地降低需要传输的数据量。

6 词频统计案例进阶之Partitioner

6.1 默认的Partitioner

这里假设有个需求：将不同单词的统计结果输出到不同文件。这种需求实际上比较常见，比如统计产品的销量时，需要将结果按照产品种类进行拆分。要实现这个功能，就需要用到自定义 Partitioner 。

这里先介绍下 MapReduce 默认的分类规则：在构建 job 时候，如果不指定，默认的使用的是HashPartitioner ：对 key 值进行哈希散列并对 numReduceTasks 取余。其实现如下：

public class HashPartitioner<K, V> extends Partitioner<K, V> {
   
   
    public int getPartition(K key, V value, int numReduceTasks) {
   
   
        return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }
}

6.2 自定义Partitioner

这里我们继承 Partitioner 自定义分类规则，这里按照单词进行分类：

    public class CustomPartitioner extends Partitioner<Text, IntWritable> {
   
   
    public int getPartition(Text text, IntWritable intWritable, int numPartitions) {
   
   
        return WordCountDataUtils.WORD_LIST.indexOf(text.toString());
    }
}

在构建 job 时候指定使用我们自己的分类规则，并设置 reduce 的个数：

// 设置自定义分区规则
job.setPartitionerClass(CustomPartitioner.class);
// 设置 reduce 个数
job.setNumReduceTasks(WordCountDataUtils.WORD_LIST.size());

6.3 执行结果

执行结果如下，分别生成 6 个文件，每个文件中为对应单词的统计结果：

7 作业的生命周期

1. 作业提交与初始化

提交作业后， 首先由 JobClient 实例将作业相关信息， 比如将程序 jar 包、作业配置文件、 分片元信息文件等上传到分布式文件系统（ 一般为 HDFS）上，其中，分片元信息文件记录了每个输入分片的逻辑位置信息。 然后 JobClient 通过 RPC 通知 JobTracker。

JobTracker 收到新作业提交请求后， 由作业调度模块对作业进行初始化：为作业创建一个JobInProgress 对象以跟踪作业运行状况， 而 JobInProgress 则会为每个 Task 创建一个TaskInProgress 对象以跟踪每个任务的运行状态，TaskInProgress 可能需要管理多个“Task 运行尝试”（ 称为“ Task Attempt”）。

1. 任务调度与监控

前面提到，任务调度和监控的功能均由 JobTracker 完成。TaskTracker 周期性地通过Heartbeat 向 JobTracker 汇报本节点的资源使用情况， 一旦出现空闲资源， JobTracker会按照一定的策略选择一个合适的任务使用该空闲资源， 这由任务调度器完成。 任务调度器是一个可插拔的独立模块， 且为双层架构， 即首先选择作业， 然后从该作业中选择任务， 其中，选择任务时需要重点考虑数据本地性。 此外，JobTracker 跟踪作业的整个运行过程，并为作业的成功运行提供全方位的保障。 首先， 当 TaskTracker 或者 Task 失败时， 转移计算任务 ； 其次， 当某个 Task 执行进度远落后于同一作业的其他 Task 时，为之启动一个相同Task， 并选取计算快的 Task 结果作为最终结果。

1. 任务运行环境准备

运行环境准备包括 JVM 启动和资源隔离， 均由 TaskTracker 实现。 TaskTracker 为每个Task 启动一个独立的 JVM 以避免不同 Task 在运行过程中相互影响 ； 同时，TaskTracker 使用了操作系统进程实现资源隔离以防止Task滥用资源。

1. 任务执行

TaskTracker 为 Task 准备好运行环境后， 便会启动 Task。 在运行过程中， 每个 Task 的最新进度首先由 Task 通过 RPC 汇报给 TaskTracker， 再由 TaskTracker 汇报给 JobTracker。

1. 作业完成。

待所有 Task 执行完毕后， 整个作业执行成功。