blog/BigData-Notes
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

mapreduce

Browse Source
This commit is contained in:
luoxiang
2019-06-02 17:20:08 +08:00
parent bf90aea602
commit d18cb4b731
1 changed files with 75 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

223
notes/Hadoop-MapReduce.md
View File

@@ -1,23 +1,28 @@
# 分布式计算框架——MapReduce # 分布式计算框架——MapReduce
<nav> <nav>
<a href="#一MapReduce-概述">一、MapReduce 概述</a><br/> <a href="#一MapReduce概述">一、MapReduce概述</a><br/>
<a href="#二MapReduce-编程模型简述">二、MapReduce 编程模型简述</a><br/> <a href="#二MapReduce编程模型简述">二、MapReduce编程模型简述</a><br/>
<a href="#三combiner--partitioner">三、combiner & partitioner</a><br/> <a href="#三combiner--partitioner">三、combiner & partitioner</a><br/>
<a href="#四MapReduce-词频统计案例">四、MapReduce 词频统计案例</a><br/> <a href="#四MapReduce词频统计案例">四、MapReduce词频统计案例</a><br/>
<a href="#五词频统计案例进阶">五、词频统计案例进阶</a><br/> &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#41-项目简介">4.1 项目简介</a><br/>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#42-项目依赖">4.2 项目依赖</a><br/>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#43-WordCountMapper">4.3 WordCountMapper</a><br/>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#44-WordCountReducer">4.4 WordCountReducer</a><br/>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#44-WordCountApp">4.4 WordCountApp</a><br/>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#45-提交到服务器运行">4.5 提交到服务器运行</a><br/>
<a href="#五词频统计案例进阶之Combiner">五、词频统计案例进阶之Combiner</a><br/>
<a href="#六词频统计案例进阶之Partitioner">六、词频统计案例进阶之Partitioner</a><br/>
</nav> </nav>
## 一、MapReduce概述 ## 一、MapReduce概述
Hadoop MapReduce是一个分布式计算框架，用于编写批处理应用程序。编写好的程序可以提交到Hadoop集群上用于并行处理大规模的数据集。 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) 接口。 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) (input) <k1, v1> -> map -> <k2, v2> -> combine -> <k2, v2> -> reduce -> <k3, v3> (output)
@@ -27,7 +32,7 @@ MapReduce作业通过将输入的数据集拆分为独立的块，这些块由`m
## 二、MapReduce编程模型简述 ## 二、MapReduce编程模型简述
这里以词频统计为例说明MapReduce的编程模型，下图为词频统计的流程图： 这里以词频统计为例进行说明，MapReduce处理的流程如下：
<div align="center"> <img width="600px" src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/mapreduceProcess.png"/> </div> <div align="center"> <img width="600px" src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/mapreduceProcess.png"/> </div>
@@ -39,7 +44,7 @@ MapReduce作业通过将输入的数据集拆分为独立的块，这些块由`m
4. **shuffling**：由于`Mapping`操作可能是在不同的机器上并行处理的，所以需要通过`shuffling`将相同`key`值的数据分发到同一个节点上去合并，这样才能统计出最终的结果，此时得到`K2`为每一个单词，`List(V2)`为可迭代集合，`V2`就是Mapping中的V2； 4. **shuffling**：由于`Mapping`操作可能是在不同的机器上并行处理的，所以需要通过`shuffling`将相同`key`值的数据分发到同一个节点上去合并，这样才能统计出最终的结果，此时得到`K2`为每一个单词，`List(V2)`为可迭代集合，`V2`就是Mapping中的V2；
5. **Reducing** : 这里的案例是统计单词出现的总次数，所以`Reducing`对`List(V2)`进行归约求和操作，最终输出。 5. **Reducing** : 这里的案例是统计单词出现的总次数，所以`Reducing`对`List(V2)`进行归约求和操作，最终输出。
MapReduce编程模型中`splitting` 和`shuffing`操作都是由框架实现的，需要我们自己编程实现的只有`mapping`和`reducing`，这也就是框架MapReduce名字的来源。 MapReduce编程模型中`splitting` 和`shuffing`操作都是由框架实现的，需要我们自己编程实现的只有`mapping`和`reducing`，这也就是MapReduce这个称呼的来源。
@@ -49,36 +54,17 @@ MapReduce编程模型中`splitting` 和`shuffing`操作都是由框架实现的
### 3.1 InputFormat & RecordReaders ### 3.1 InputFormat & RecordReaders
`InputFormat`将输出文件拆分为多个`InputSplit`，并由`RecordReaders`将`InputSplit`转换为标准的<key,value>键值对，作为map的输出。这一步的意义在于只有先进行逻辑拆分并转为标准的键值对格式后，才能为多个`map`提供输入，进行并行处理。 `InputFormat`将输出文件拆分为多个`InputSplit`，并由`RecordReaders`将`InputSplit`转换为标准的<key，value>键值对，作为map的输出。这一步的意义在于只有先进行逻辑拆分并转为标准的键值对格式后，才能为多个`map`提供输入，以便进行并行处理。
`InputFormat` 为一个抽象类，其中只定义了两个抽象方法，具体的操作则由其实现类来进行。其源码如下：
- **getSplits**：将输入文件拆分为多个InputSplit；
- **createRecordReader**: 定义RecordReader的创建方法；
```java
public abstract class InputFormat<K, V> {
public abstract
List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException;
public abstract
RecordReader<K,V> createRecordReader(InputSplit split,
TaskAttemptContext context
) throws IOException,
InterruptedException;
} ### 3.2 Combiner
```
### 3.2 combiner
`combiner`是`map`运算后的可选操作，它实际上是一个本地化的`reduce`操作，它主要是在`map`计算出中间文件后做一个简单的合并重复`key`值的操作。这里以词频统计为例： `combiner`是`map`运算后的可选操作，它实际上是一个本地化的`reduce`操作，它主要是在`map`计算出中间文件后做一个简单的合并重复`key`值的操作。这里以词频统计为例：
`map`在遇到一个hadoop的单词时就会记录为1，但是这篇文章里hadoop可能会出现n多次，那么`map`输出文件冗余就会很多，因此在`reduce`计算前对相同的key做一个合并操作，那么需要传输的数据量就会减少，传输效率就可以得到提升。 `map`在遇到一个hadoop的单词时就会记录为1，但是这篇文章里hadoop可能会出现n多次，那么`map`输出文件冗余就会很多，因此在`reduce`计算前对相同的key做一个合并操作，那么需要传输的数据量就会减少，传输效率就可以得到提升。
但并非所有场景都适合使用`combiner`，使用它的原则是`combiner`的输出不会影响到`reduce`计算的最终输入，例如：如果计算只是求总数，最大值，最小值时都可以使用`combiner`，但是做平均值计算则不能使用`combiner`。 但并非所有场景都适合使用`combiner`，使用它的原则是`combiner`的输出不会影响到`reduce`计算的最终输入，例如：求总数，最大值，最小值时都可以使用`combiner`，但是做平均值计算则不能使用`combiner`。
不使用combiner的情况： 不使用combiner的情况：
@@ -90,21 +76,19 @@ public abstract class InputFormat<K, V> {
可以看到使用combiner的时候，需要传输到reducer中的数据由12keys，降低到10keys。降低的幅度取决于你keys的重复率，后文词频统计案例可以直观演示用combiner降低数百倍的传输量。 可以看到使用combiner的时候，需要传输到reducer中的数据由12keys，降低到10keys。降低的幅度取决于你keys的重复率，下文词频统计案例会演示用combiner降低数百倍的传输量。
### 3.3 partitioner ### 3.3 Partitioner
`partitioner`可以理解成分类器，将`map`的输出按照key值的不同分别分给对应的`reducer`，支持自定义实现，在后文案例中会有演示。 `partitioner`可以理解成分类器，将`map`的输出按照key值的不同分别分给对应的`reducer`，支持自定义实现，下文案例会给出演示。
## 四、MapReduce词频统计案例 ## 四、MapReduce词频统计案例
> 源码下载地址：[hadoop-word-count](https://github.com/heibaiying/BigData-Notes/tree/master/code/Hadoop/hadoop-word-count)
### 4.1 项目简介 ### 4.1 项目简介
这里给出一个经典的案例:词频统计。统计如下样本数据中每个单词出现的次数。 这里给出一个经典的词频统计的案例：统计如下样本数据中每个单词出现的次数。
```properties ```properties
Spark HBase Spark HBase
@@ -122,89 +106,29 @@ Hive Flink Hadoop
HBase Hive HBase Hive
``` ```
为方便大家开发，我在项目源码中放置了一个工具类`WordCountDataUtils`，用于产生词频统计样本文件： 为方便大家开发，我在项目源码中放置了一个工具类`WordCountDataUtils`，用于模拟产生词频统计的样本，生成的文件支持输出到本地或者直接写到HDFS上。
> 项目完整源码下载地址：[hadoop-word-count](https://github.com/heibaiying/BigData-Notes/tree/master/code/Hadoop/hadoop-word-count)
```java
public class WordCountDataUtils {
public static final List<String> WORD_LIST = Arrays.asList("Spark", "Hadoop", "HBase", ### 4.2 项目依赖
"Storm", "Flink", "Hive");
想要进行MapReduce编程，需要导入`hadoop-client`依赖：
/** ```xml
* 模拟产生词频数据 <dependency>
* <groupId>org.apache.hadoop</groupId>
* @return 词频数据 <artifactId>hadoop-client</artifactId>
*/ <version>${hadoop.version}</version>
private static String generateData() { </dependency>
StringBuilder builder = new StringBuilder();
for (int i = 0; i < 1000; i++) {
Collections.shuffle(WORD_LIST);
Random random = new Random();
int endIndex = random.nextInt(WORD_LIST.size()) % (WORD_LIST.size()) + 1;
String line = StringUtils.join(WORD_LIST.toArray(), "\t", 0, endIndex);
builder.append(line).append("\n");
}
return builder.toString();
}
/**
* 模拟产生词频数据并输出到本地
*
* @param outputPath 输出文件路径
*/
private static void generateDataToLocal(String outputPath) {
try {
java.nio.file.Path path = Paths.get(outputPath);
if (Files.exists(path)) {
Files.delete(path);
}
Files.write(path, generateData().getBytes(), StandardOpenOption.CREATE);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 模拟产生词频数据并输出到HDFS
*
* @param hdfsUrl HDFS地址
* @param user hadoop用户名
* @param outputPathString 存储到HDFS上的路径
*/
private static void generateDataToHDFS(String hdfsUrl, String user, String outputPathString) {
FileSystem fileSystem = null;
try {
fileSystem = FileSystem.get(new URI(hdfsUrl), new Configuration(), user);
Path outputPath = new Path(outputPathString);
if (fileSystem.exists(outputPath)) {
fileSystem.delete(outputPath, true);
}
FSDataOutputStream out = fileSystem.create(outputPath);
out.write(generateData().getBytes());
out.flush();
out.close();
fileSystem.close();
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
//generateDataToLocal("input.txt");
generateDataToHDFS("hdfs://192.168.0.107:8020", "root", "/wordcount/input.txt");
}
}
``` ```
### 4.2 WordCountMapper ### 4.3 WordCountMapper
将每行数据按照指定分隔符进行拆分：
```java ```java
/**
* 将每行数据按照指定分隔符进行拆分
*/
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
@Override @Override
@@ -219,8 +143,13 @@ public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritabl
} }
``` ```
`WordCountMapper`对应下图的Mapping操作：
WordCountMapper对应下图的Mapping操作，这里WordCountMapper继承自Mapper类，这是一个泛型类，定义如下： <div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/hadoop-code-mapping.png"/> </div>
`WordCountMapper`继承自`Mappe`r类，这是一个泛型类，定义如下：
```java ```java
public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> { public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
@@ -228,21 +157,20 @@ public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
} }
``` ```
+ **KEYIN** : mapping输入的key的数据类型，即每行的偏移量(每行第一个字符在文本中的位置)，Long类型，对应Hadoop中的LongWritable类型； + **KEYIN** : `mapping`输入的key的数据类型，即每行的偏移量(每行第一个字符在整个文本中的位置)，`Long`类型，对应Hadoop中的`LongWritable`类型；
+ **VALUEIN** : mappin输入的value的数据类型，即每行数据；String类型，对应Hadoop中Text类型； + **VALUEIN** : `mapping`输入的value的数据类型，即每行数据；`String`类型，对应Hadoop中`Text`类型；
+ **KEYOUT** ：mapping输出的key的数据类型，即每个单词；String类型，对应Hadoop中Text类型； + **KEYOUT** ：`mapping`输出的key的数据类型，即每个单词；`String`类型，对应Hadoop中`Text`类型；
+ **VALUEOUT**：mapping输出的value的数据类型，即每个单词出现的次数；这里用int类型，对应Hadoop中IntWritable类型； + **VALUEOUT**：`mapping`输出的value的数据类型，即每个单词出现的次数；这里用`int`类型，对应Hadoop中`IntWritable`类型；
在MapReduce中必须使用Hadoop定义的类型，因为Hadoop预定义的类型都是可序列化，可比较的，所有类型均实现了`WritableComparable`接口。 在MapReduce中必须使用Hadoop定义的类型，因为Hadoop预定义的类型都是可序列化，可比较的，所有类型均实现了`WritableComparable`接口。
<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/hadoop-code-mapping.png"/> </div>
### 4.3 WordCountReducer
### 4.4 WordCountReducer
在Reduce中进程单词出现次数统计：
```java ```java
/**
* 进行词频统计
*/
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
@Override @Override
@@ -257,7 +185,7 @@ public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritab
} }
``` ```
这里的key是每个单词，values是一个可迭代的数据类型，因为shuffling输出的数据实际上是下图中所示的这样的，即`key，(1,1,1,1,1,1,1,.....)`。 如下图，`shuffling`的输出是reduce的输入。这里的key是每个单词，values是一个可迭代的数据类型，类似`(1,1,1,...)`。
<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/hadoop-code-reducer.png"/> </div> <div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/hadoop-code-reducer.png"/> </div>
@@ -334,19 +262,17 @@ public class WordCountApp {
} }
``` ```
这里说明一下：`setMapOutputKeyClass`和`setOutputValueClass`用于设置reducer函数的输出类型。map函数的输出类型默认情况下和reducer函数式相同的，如果不同，则必须通过`setMapOutputKeyClass`和`setMapOutputValueClass`进行设置。 需要注意的是：如果不设置`Mapper`操作的输出类型，则程序默认它和`Reducer`操作输出的类型相同。
### 4.5 提交到服务器运行 ### 4.5 提交到服务器运行
在实际开发中，可以在本机配置hadoop开发环境，直接运行`main`方法既可。这里主要介绍一下打包提交到服务器运行： 在实际开发中，可以在本机配置hadoop开发环境，直接在IDE中启动进行测试。这里主要介绍一下打包提交到服务器运行。由于本项目没有使用除Hadoop外的第三方依赖，直接打包即可：
由于本项目没有使用除Hadoop外的第三方依赖，直接打包即可：
```shell ```shell
# mvn clean package # mvn clean package
``` ```
使用以下命令运行作业： 使用以下命令提交作业：
```shell ```shell
hadoop jar /usr/appjar/hadoop-word-count-1.0.jar \ hadoop jar /usr/appjar/hadoop-word-count-1.0.jar \
@@ -368,40 +294,40 @@ hadoop fs -cat /wordcount/output/WordCountApp/part-r-00000
## 五、词频统计案例进阶 ## 五、词频统计案例进阶之Combiner
## 5.1 combiner ### 5.1 代码实现
### 1. combiner的代码实现 想要使用`combiner`功能只要在组装作业时，添加下面一行代码即可：
combiner的代码实现比较简单，只要在组装作业时，添加下面一行代码即可：
```java ```java
// 设置Combiner // 设置Combiner
job.setCombinerClass(WordCountReducer.class); job.setCombinerClass(WordCountReducer.class);
``` ```
### 2. 测试结果 ### 5.2 执行结果
加入combiner后统计结果是不会有变化的，但是我们可以从打印的日志看出combiner的效果： 加入`combiner`后统计结果是不会有变化的，但是可以从打印的日志看出`combiner`的效果：
没有加入combiner的打印日志： 没有加入`combiner`的打印日志：
<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/hadoop-no-combiner.png"/> </div> <div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/hadoop-no-combiner.png"/> </div>
加入combiner后的打印日志如下。 加入`combiner`后的打印日志如下：
<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/hadoop-combiner.png"/> </div> <div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/hadoop-combiner.png"/> </div>
这里我们只有一个输入文件并且小于128M，所以只有一个Map进行处理，可以看到经过combiner后，records由3519降低为6（样本中单词种类就只有6个），在这个用例中combiner就能极大地降低需要传输的数据量。 这里我们只有一个输入文件并且小于128M，所以只有一个Map进行处理。可以看到经过combiner后，records由`3519`降低为`6`(样本中单词种类就只有6种)，在这个用例中combiner就能极大地降低需要传输的数据量。
## 5.2 Partitioner
### 1. 默认Partitioner规则
这里假设有个需求：将不同单词的统计结果输出到不同文件。这种需求实际上比较常见，比如统计产品的销量时，需要将结果按照产品分类输出。 ## 六、词频统计案例进阶之Partitioner
要实现这个功能，就需要用到自定义Partitioner，这里我们先说一下默认的分区规则：在构建job时候，如果不指定，默认的使用的是`HashPartitioner`，其实现如下： ### 6.1 默认的Partitioner
这里假设有个需求：将不同单词的统计结果输出到不同文件。这种需求实际上比较常见，比如统计产品的销量时，需要将结果按照产品种类进行拆分。要实现这个功能，就需要用到自定义`Partitioner`。
这里先介绍下MapReduce默认的分类规则：在构建job时候，如果不指定，默认的使用的是`HashPartitioner`：对key值进行哈希散列并对`numReduceTasks`取余。其实现如下：
```java ```java
public class HashPartitioner<K, V> extends Partitioner<K, V> { public class HashPartitioner<K, V> extends Partitioner<K, V> {
@@ -414,16 +340,11 @@ public class HashPartitioner<K, V> extends Partitioner<K, V> {
} }
``` ```
对key进行哈希散列并对`numReduceTasks`取余，这里由于`numReduceTasks`默认值为1，所以我们之前的统计结果都输出到同一个文件中。 ### 6.2 自定义Partitioner
### 2. 自定义Partitioner 这里我们继承`Partitioner`自定义分类规则，这里按照单词进行分类：
这里我们继承`Partitioner`自定义分区规则，这里按照单词进行分区：
```java ```java
/**
* 自定义partitioner,按照单词分区
*/
public class CustomPartitioner extends Partitioner<Text, IntWritable> { public class CustomPartitioner extends Partitioner<Text, IntWritable> {
public int getPartition(Text text, IntWritable intWritable, int numPartitions) { public int getPartition(Text text, IntWritable intWritable, int numPartitions) {
@@ -432,7 +353,7 @@ public class CustomPartitioner extends Partitioner<Text, IntWritable> {
} }
``` ```
并在构建job时候指定使用我们自己的分区规则，并设置reduce的个数： 在构建`job`时候指定使用我们自己的分类规则，并设置`reduce`的个数：
```java ```java
// 设置自定义分区规则 // 设置自定义分区规则
@@ -443,9 +364,9 @@ job.setNumReduceTasks(WordCountDataUtils.WORD_LIST.size());
### 3. 测试结果 ### 6.3 执行结果
测试结果如下，分别生成6个文件，每个文件中为对应单词的统计结果。 执行结果如下，分别生成6个文件，每个文件中为对应单词的统计结果：
<div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/hadoop-wordcountcombinerpartition.png"/> </div> <div align="center"> <img src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/hadoop-wordcountcombinerpartition.png"/> </div>