388 lines
16 KiB
Markdown
388 lines
16 KiB
Markdown
# 分布式计算框架——MapReduce
|
||
|
||
<nav>
|
||
<a href="#一MapReduce概述">一、MapReduce概述</a><br/>
|
||
<a href="#二MapReduce编程模型简述">二、MapReduce编程模型简述</a><br/>
|
||
<a href="#三combiner--partitioner">三、combiner & partitioner</a><br/>
|
||
<a href="#四MapReduce词频统计案例">四、MapReduce词频统计案例</a><br/>
|
||
<a href="#41-项目简介">4.1 项目简介</a><br/>
|
||
<a href="#42-项目依赖">4.2 项目依赖</a><br/>
|
||
<a href="#43-WordCountMapper">4.3 WordCountMapper</a><br/>
|
||
<a href="#44-WordCountReducer">4.4 WordCountReducer</a><br/>
|
||
<a href="#44-WordCountApp">4.4 WordCountApp</a><br/>
|
||
<a href="#45-提交到服务器运行">4.5 提交到服务器运行</a><br/>
|
||
<a href="#五词频统计案例进阶之Combiner">五、词频统计案例进阶之Combiner</a><br/>
|
||
<a href="#六词频统计案例进阶之Partitioner">六、词频统计案例进阶之Partitioner</a><br/>
|
||
</nav>
|
||
|
||
|
||
|
||
|
||
## 一、MapReduce概述
|
||
|
||
Hadoop MapReduce 是一个分布式计算框架，用于编写批处理应用程序。编写好的程序可以提交到 Hadoop 集群上用于并行处理大规模的数据集。
|
||
|
||
MapReduce 作业通过将输入的数据集拆分为独立的块，这些块由 `map` 以并行的方式处理，框架对 `map` 的输出进行排序，然后输入到 `reduce` 中。MapReduce 框架专门用于 `<key，value>` 键值对处理，它将作业的输入视为一组 `<key，value>` 对，并生成一组 `<key，value>` 对作为输出。输入和输出的 `key` 和 `value` 都必须实现[Writable](http://hadoop.apache.org/docs/stable/api/org/apache/hadoop/io/Writable.html) 接口。
|
||
|
||
```
|
||
(input) <k1, v1> -> map -> <k2, v2> -> combine -> <k2, v2> -> reduce -> <k3, v3> (output)
|
||
```
|
||
|
||
|
||
|
||
## 二、MapReduce编程模型简述
|
||
|
||
这里以词频统计为例进行说明，MapReduce 处理的流程如下：
|
||
|
||
<div align="center"> <img width="600px" src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/mapreduceProcess.png"/> </div>
|
||
|
||
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 这个称呼的来源。
|
||
|
||
|
||
|
||
## 三、combiner & partitioner
|
||
|
||
<div align="center"> <img width="600px" src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/Detailed-Hadoop-MapReduce-Data-Flow-14.png"/> </div>
|
||
|
||
### 3.1 InputFormat & RecordReaders
|
||
|
||
`InputFormat` 将输出文件拆分为多个 `InputSplit`，并由 `RecordReaders` 将 `InputSplit` 转换为标准的<key，value>键值对，作为 map 的输出。这一步的意义在于只有先进行逻辑拆分并转为标准的键值对格式后，才能为多个 `map` 提供输入，以便进行并行处理。
|
||
|
||
|
||
|
||
### 3.2 Combiner
|
||
|
||
`combiner` 是 `map` 运算后的可选操作，它实际上是一个本地化的 `reduce` 操作，它主要是在 `map` 计算出中间文件后做一个简单的合并重复 `key` 值的操作。这里以词频统计为例：
|
||
|
||
`map` 在遇到一个 hadoop 的单词时就会记录为 1，但是这篇文章里 hadoop 可能会出现 n 多次，那么 `map` 输出文件冗余就会很多，因此在 `reduce` 计算前对相同的 key 做一个合并操作，那么需要传输的数据量就会减少，传输效率就可以得到提升。
|
||
|
||
但并非所有场景都适合使用 `combiner`，使用它的原则是 `combiner` 的输出不会影响到 `reduce` 计算的最终输入，例如：求总数，最大值，最小值时都可以使用 `combiner`，但是做平均值计算则不能使用 `combiner`。
|
||
|
||
不使用 combiner 的情况：
|
||
|
||
<div align="center"> <img width="600px" src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/mapreduce-without-combiners.png"/> </div>
|
||
|
||
使用 combiner 的情况：
|
||
|
||
<div align="center"> <img width="600px" src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/mapreduce-with-combiners.png"/> </div>
|
||
|
||
|
||
|
||
可以看到使用 combiner 的时候，需要传输到 reducer 中的数据由 12keys，降低到 10keys。降低的幅度取决于你 keys 的重复率，下文词频统计案例会演示用 combiner 降低数百倍的传输量。
|
||
|
||
### 3.3 Partitioner
|
||
|
||
`partitioner` 可以理解成分类器，将 `map` 的输出按照 key 值的不同分别分给对应的 `reducer`，支持自定义实现，下文案例会给出演示。
|
||
|
||
|
||
|
||
## 四、MapReduce词频统计案例
|
||
|
||
### 4.1 项目简介
|
||
|
||
这里给出一个经典的词频统计的案例：统计如下样本数据中每个单词出现的次数。
|
||
|
||
```properties
|
||
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](https://github.com/heibaiying/BigData-Notes/tree/master/code/Hadoop/hadoop-word-count)
|
||
|
||
|
||
|
||
### 4.2 项目依赖
|
||
|
||
想要进行 MapReduce 编程，需要导入 `hadoop-client` 依赖：
|
||
|
||
```xml
|
||
<dependency>
|
||
<groupId>org.apache.hadoop</groupId>
|
||
<artifactId>hadoop-client</artifactId>
|
||
<version>${hadoop.version}</version>
|
||
</dependency>
|
||
```
|
||
|
||
### 4.3 WordCountMapper
|
||
|
||
将每行数据按照指定分隔符进行拆分。这里需要注意在 MapReduce 中必须使用 Hadoop 定义的类型，因为 Hadoop 预定义的类型都是可序列化，可比较的，所有类型均实现了 `WritableComparable` 接口。
|
||
|
||
```java
|
||
public class WordCountMapper 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 操作：
|
||
|
||
<div align="center"> <img src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/hadoop-code-mapping.png"/> </div>
|
||
|
||
|
||
|
||
`WordCountMapper` 继承自 `Mapper` 类，这是一个泛型类，定义如下：
|
||
|
||
```java
|
||
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 中进行单词出现次数的统计：
|
||
|
||
```java
|
||
public class WordCountReducer 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(key, new IntWritable(count));
|
||
}
|
||
}
|
||
```
|
||
|
||
如下图，`shuffling` 的输出是 reduce 的输入。这里的 key 是每个单词，values 是一个可迭代的数据类型，类似 `(1,1,1,...)`。
|
||
|
||
<div align="center"> <img src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/hadoop-code-reducer.png"/> </div>
|
||
|
||
### 4.4 WordCountApp
|
||
|
||
组装 MapReduce 作业，并提交到服务器运行，代码如下：
|
||
|
||
```java
|
||
|
||
/**
|
||
* 组装作业 并提交到集群运行
|
||
*/
|
||
public class WordCountApp {
|
||
|
||
|
||
// 这里为了直观显示参数 使用了硬编码，实际开发中可以通过外部传参
|
||
private static final String HDFS_URL = "hdfs://192.168.0.107:8020";
|
||
private static final String HADOOP_USER_NAME = "root";
|
||
|
||
public static void main(String[] args) throws Exception {
|
||
|
||
// 文件输入路径和输出路径由外部传参指定
|
||
if (args.length < 2) {
|
||
System.out.println("Input and output paths are necessary!");
|
||
return;
|
||
}
|
||
|
||
// 需要指明 hadoop 用户名，否则在 HDFS 上创建目录时可能会抛出权限不足的异常
|
||
System.setProperty("HADOOP_USER_NAME", HADOOP_USER_NAME);
|
||
|
||
Configuration configuration = new Configuration();
|
||
// 指明 HDFS 的地址
|
||
configuration.set("fs.defaultFS", HDFS_URL);
|
||
|
||
// 创建一个 Job
|
||
Job job = Job.getInstance(configuration);
|
||
|
||
// 设置运行的主类
|
||
job.setJarByClass(WordCountApp.class);
|
||
|
||
// 设置 Mapper 和 Reducer
|
||
job.setMapperClass(WordCountMapper.class);
|
||
job.setReducerClass(WordCountReducer.class);
|
||
|
||
// 设置 Mapper 输出 key 和 value 的类型
|
||
job.setMapOutputKeyClass(Text.class);
|
||
job.setMapOutputValueClass(IntWritable.class);
|
||
|
||
// 设置 Reducer 输出 key 和 value 的类型
|
||
job.setOutputKeyClass(Text.class);
|
||
job.setOutputValueClass(IntWritable.class);
|
||
|
||
// 如果输出目录已经存在，则必须先删除，否则重复运行程序时会抛出异常
|
||
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);
|
||
}
|
||
|
||
// 设置作业输入文件和输出文件的路径
|
||
FileInputFormat.setInputPaths(job, new Path(args[0]));
|
||
FileOutputFormat.setOutputPath(job, outputPath);
|
||
|
||
// 将作业提交到群集并等待它完成，参数设置为 true 代表打印显示对应的进度
|
||
boolean result = job.waitForCompletion(true);
|
||
|
||
// 关闭之前创建的 fileSystem
|
||
fileSystem.close();
|
||
|
||
// 根据作业结果,终止当前运行的 Java 虚拟机,退出程序
|
||
System.exit(result ? 0 : -1);
|
||
|
||
}
|
||
}
|
||
```
|
||
|
||
需要注意的是：如果不设置 `Mapper` 操作的输出类型，则程序默认它和 `Reducer` 操作输出的类型相同。
|
||
|
||
### 4.5 提交到服务器运行
|
||
|
||
在实际开发中，可以在本机配置 hadoop 开发环境，直接在 IDE 中启动进行测试。这里主要介绍一下打包提交到服务器运行。由于本项目没有使用除 Hadoop 外的第三方依赖，直接打包即可：
|
||
|
||
```shell
|
||
# mvn clean package
|
||
```
|
||
|
||
使用以下命令提交作业：
|
||
|
||
```shell
|
||
hadoop jar /usr/appjar/hadoop-word-count-1.0.jar \
|
||
com.heibaiying.WordCountApp \
|
||
/wordcount/input.txt /wordcount/output/WordCountApp
|
||
```
|
||
|
||
作业完成后查看 HDFS 上生成目录：
|
||
|
||
```shell
|
||
# 查看目录
|
||
hadoop fs -ls /wordcount/output/WordCountApp
|
||
|
||
# 查看统计结果
|
||
hadoop fs -cat /wordcount/output/WordCountApp/part-r-00000
|
||
```
|
||
|
||
<div align="center"> <img src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/hadoop-wordcountapp.png"/> </div>
|
||
|
||
|
||
|
||
## 五、词频统计案例进阶之Combiner
|
||
|
||
### 5.1 代码实现
|
||
|
||
想要使用 `combiner` 功能只要在组装作业时，添加下面一行代码即可：
|
||
|
||
```java
|
||
// 设置 Combiner
|
||
job.setCombinerClass(WordCountReducer.class);
|
||
```
|
||
|
||
### 5.2 执行结果
|
||
|
||
加入 `combiner` 后统计结果是不会有变化的，但是可以从打印的日志看出 `combiner` 的效果：
|
||
|
||
没有加入 `combiner` 的打印日志：
|
||
|
||
<div align="center"> <img src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/hadoop-no-combiner.png"/> </div>
|
||
|
||
加入 `combiner` 后的打印日志如下：
|
||
|
||
<div align="center"> <img src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/hadoop-combiner.png"/> </div>
|
||
|
||
这里我们只有一个输入文件并且小于 128M，所以只有一个 Map 进行处理。可以看到经过 combiner 后，records 由 `3519` 降低为 `6`(样本中单词种类就只有 6 种)，在这个用例中 combiner 就能极大地降低需要传输的数据量。
|
||
|
||
|
||
|
||
## 六、词频统计案例进阶之Partitioner
|
||
|
||
### 6.1 默认的Partitioner
|
||
|
||
这里假设有个需求：将不同单词的统计结果输出到不同文件。这种需求实际上比较常见，比如统计产品的销量时，需要将结果按照产品种类进行拆分。要实现这个功能，就需要用到自定义 `Partitioner`。
|
||
|
||
这里先介绍下 MapReduce 默认的分类规则：在构建 job 时候，如果不指定，默认的使用的是 `HashPartitioner`：对 key 值进行哈希散列并对 `numReduceTasks` 取余。其实现如下：
|
||
|
||
```java
|
||
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` 自定义分类规则，这里按照单词进行分类：
|
||
|
||
```java
|
||
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` 的个数：
|
||
|
||
```java
|
||
// 设置自定义分区规则
|
||
job.setPartitionerClass(CustomPartitioner.class);
|
||
// 设置 reduce 个数
|
||
job.setNumReduceTasks(WordCountDataUtils.WORD_LIST.size());
|
||
```
|
||
|
||
|
||
|
||
### 6.3 执行结果
|
||
|
||
执行结果如下，分别生成 6 个文件，每个文件中为对应单词的统计结果：
|
||
|
||
<div align="center"> <img src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/hadoop-wordcountcombinerpartition.png"/> </div>
|
||
|
||
|
||
|
||
|
||
|
||
## 参考资料
|
||
|
||
1. [分布式计算框架 MapReduce](https://zhuanlan.zhihu.com/p/28682581)
|
||
2. [Apache Hadoop 2.9.2 > MapReduce Tutorial](http://hadoop.apache.org/docs/stable/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html)
|
||
3. [MapReduce - Combiners]( https://www.tutorialscampus.com/tutorials/map-reduce/combiners.htm)
|
||
|
||
|
||
|
||
|
||
|
||
<div align="center"> <img src="https://gitee.com/heibaiying/BigData-Notes/raw/master/pictures/weixin-desc.png"/> </div>
|