hadoop-mapreduce
This commit is contained in:
		| @@ -3,20 +3,21 @@ | |||||||
| <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="#三MapReduce-编程模型详述">三、MapReduce 编程模型详述</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/> | <a href="#五词频统计案例进阶">五、词频统计案例进阶</a><br/> | ||||||
| </nav> | </nav> | ||||||
|  |  | ||||||
|  |  | ||||||
|  |  | ||||||
|  |  | ||||||
|  |  | ||||||
|  |  | ||||||
| ## 一、MapReduce 概述 | ## 一、MapReduce 概述 | ||||||
|  |  | ||||||
| Hadoop MapReduce是一个分布式计算框架,用于编写应用程序,以可靠,容错的方式在大型集群上并行处理大量数据(多为TB级别数据集)。 | Hadoop MapReduce是一个分布式计算框架,用于编写批处理应用程序。编写好的程序可以提交到Hadoop集群上用于并行处理大规模的数据集。 | ||||||
|  |  | ||||||
| MapReduce 作业通常将输入数据集拆分为独立的块,这些块由**map任务**以完全并行的方式处理。框架对**map任务**的输出进行排序,然后输入到**reduce任务**。通常,作业的输入和输出都存储在文件系统中。该框架负责调度任务,监视任务并重新执行失败的任务。 | 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框架专门用于`<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) | ||||||
| @@ -32,29 +33,28 @@ MapReduce框架专门用于`<key,value>`对,也就是说,框架将作业 | |||||||
|  |  | ||||||
| 1. **input** : 读取文本文件; | 1. **input** : 读取文本文件; | ||||||
|  |  | ||||||
| 2. **splitting** : 将文件按照行进行拆分,此时得到的K1为行数,V1表示对应行的文本内容; | 2. **splitting** : 将文件按照行进行拆分,此时得到的`K1`行数,`V1`表示对应行的文本内容; | ||||||
|  |  | ||||||
| 3. **mapping** : 并行将每一行按照空格进行拆分,拆分得到的List(K2,V2),其中K2代表每一个单词,由于是做词频统计,所以其V2为1,代表出现1次; | 3. **mapping** : 并行将每一行按照空格进行拆分,拆分得到的`List(K2,V2)`,其中`K2`代表每一个单词,由于是做词频统计,所以`V2`的值为1,代表出现1次; | ||||||
| 4. **shuffling**:由于Mapping操作可能是在不同的机器上并行处理的,所以需要通过shuffling将相同的数据分到同一个节点上去合并,这样才能统计出最终的结果,此时得到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名字的来源。 | ||||||
|  |  | ||||||
|  |  | ||||||
|  |  | ||||||
| ## 三、MapReduce 编程模型详述 | ## 三、combiner & partitioner | ||||||
|  |  | ||||||
| <div align="center"> <img width="600px" src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/Detailed-Hadoop-MapReduce-Data-Flow-14.png"/> </div> | <div align="center"> <img width="600px" src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/Detailed-Hadoop-MapReduce-Data-Flow-14.png"/> </div> | ||||||
|  |  | ||||||
| ### 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` 为一个抽象类,其中定义了两个抽象方法,而实际的操作则由其实现类来进行。其源码如下: | `InputFormat` 为一个抽象类,其中只定义了两个抽象方法,具体的操作则由其实现类来进行。其源码如下: | ||||||
|  |  | ||||||
| **getSplits**:将输入文件拆分为多个InputSplit; | - **getSplits**:将输入文件拆分为多个InputSplit; | ||||||
|  | - **createRecordReader**: 定义RecordReader的创建方法; | ||||||
| **createRecordReader**: 定义RecordReader的创建方法; |  | ||||||
|  |  | ||||||
| ```java | ```java | ||||||
| public abstract class InputFormat<K, V> { | public abstract class InputFormat<K, V> { | ||||||
| @@ -74,11 +74,11 @@ public abstract class InputFormat<K, V> { | |||||||
|  |  | ||||||
| ### 3.2 combiner | ### 3.2 combiner | ||||||
|  |  | ||||||
| combiner是map运算后的可选操作,其实际上是一个本地化的reduce操作,它主要是在map计算出中间文件后做一个简单的合并重复key值的操作。 | `combiner`是`map`运算后的可选操作,它实际上是一个本地化的`reduce`操作,它主要是在`map`计算出中间文件后做一个简单的合并重复`key`值的操作。这里以词频统计为例: | ||||||
|  |  | ||||||
| 例如我们对文件里的单词频率做统计,map计算时候如果碰到一个hadoop的单词就会记录为1,但是这篇文章里hadoop可能会出现n多次,那么map输出文件冗余就会很多,因此在reduce计算前对相同的key做一个合并操作,那么文件会变小。这样就提高了宽带的传输效率,因为hadoop计算的宽带资源往往是计算的瓶颈也是最为宝贵的资源。 | `map`在遇到一个hadoop的单词时就会记录为1,但是这篇文章里hadoop可能会出现n多次,那么`map`输出文件冗余就会很多,因此在`reduce`计算前对相同的key做一个合并操作,那么需要传输的数据量就会减少,传输效率就可以得到提升。 | ||||||
|  |  | ||||||
| 但并非所有场景都适合使用combiner,使用它的原则是combiner的输入不会影响到reduce计算的最终输入,例如:如果计算只是求总数,最大值,最小值可以使用combiner,但是做平均值计算使用combiner的话,最终的reduce计算结果就会出错。 | 但并非所有场景都适合使用`combiner`,使用它的原则是`combiner`的输出不会影响到`reduce`计算的最终输入,例如:如果计算只是求总数,最大值,最小值时都可以使用`combiner`,但是做平均值计算则不能使用`combiner`。 | ||||||
|  |  | ||||||
| 不使用combiner的情况: | 不使用combiner的情况: | ||||||
|  |  | ||||||
| @@ -94,17 +94,7 @@ combiner是map运算后的可选操作,其实际上是一个本地化的reduce | |||||||
|  |  | ||||||
| ### 3.3 partitioner | ### 3.3 partitioner | ||||||
|  |  | ||||||
| partitioner可以理解成分类器,按照key的不同分别将map的输出分给不同的reduce,可以自定义实现。 | `partitioner`可以理解成分类器,将`map`的输出按照key值的不同分别分给对应的`reducer`,支持自定义实现,在后文案例中会有演示。 | ||||||
|  |  | ||||||
|  |  | ||||||
|  |  | ||||||
| ### 3.4 sort & combiner |  | ||||||
|  |  | ||||||
| <div align="center"> <img width="600px" src="https://github.com/heibaiying/BigData-Notes/blob/master/pictures/mapreduce-sort.png"/> </div> |  | ||||||
|  |  | ||||||
| 经过partitioner处理后,每个key-value对都得到分配到的reduecer信息,然后把记录先写入内存(In-memory buffer)。当写入内存的数据越来越多时,当buffer达到一定阀值(默认80M),就开始执行spill(溢写)步骤,即分成小文件写入磁盘。在写之前,先对memory中每个partition进行排序(in-memory sort)。如果数据量大的话,这个步骤会产生很多个spilled文件,如果我们定义了combine,那么在排序之前还会进行combine,最后一个步骤就是merge,把 spill 步骤产生的所有spilled files,merge成一个大的已排序文件。merge是相同的partition之间进行。 |  | ||||||
|  |  | ||||||
| Merge是怎样的?如“aaa”从某个map task读取过来时值是5,从另外一个map 读取值是8,因为它们有相同的key,所以得merge成group。什么是group。对于“aaa”就是像这样的:{“aaa”, [5, 8, 10, …]},数组中的值就是从不同溢写文件中读取出来的,然后再把这些值加起来。请注意,因为merge是将多个溢写文件合并到一个文件,所以可能也有相同的key存在,在这个过程中如果client设置过Combiner,也会使用Combiner来合并相同的key。 |  | ||||||
|  |  | ||||||
|  |  | ||||||
|  |  | ||||||
| @@ -238,7 +228,7 @@ 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** : mappin输入的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类型; | ||||||
| @@ -267,7 +257,7 @@ public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritab | |||||||
| } | } | ||||||
| ``` | ``` | ||||||
|  |  | ||||||
| 这里的key是每个单词,这里的values是一个可迭代的数据类型,因为shuffling输出的数据实际上是下图中所示的这样的,即`key,(1,1,1,1,1,1,1,.....)`。 | 这里的key是每个单词,values是一个可迭代的数据类型,因为shuffling输出的数据实际上是下图中所示的这样的,即`key,(1,1,1,1,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> | ||||||
|  |  | ||||||
| @@ -344,7 +334,7 @@ public class WordCountApp { | |||||||
| } | } | ||||||
| ``` | ``` | ||||||
|  |  | ||||||
| 这里说明一下:`setMapOutputKeyClass`和`setOutputValueClass`控制reducer函数的输出类型。map函数的输出类型默认情况下和reducer函数式相同的,如果不同,则必须通过`setMapOutputKeyClass`和`setMapOutputValueClass`进行设置。 | 这里说明一下:`setMapOutputKeyClass`和`setOutputValueClass`用于设置reducer函数的输出类型。map函数的输出类型默认情况下和reducer函数式相同的,如果不同,则必须通过`setMapOutputKeyClass`和`setMapOutputValueClass`进行设置。 | ||||||
|  |  | ||||||
| ### 4.5 提交到服务器运行 | ### 4.5 提交到服务器运行 | ||||||
|  |  | ||||||
| @@ -384,7 +374,7 @@ hadoop fs -cat /wordcount/output/WordCountApp/part-r-00000 | |||||||
|  |  | ||||||
| ### 1. combiner的代码实现 | ### 1. combiner的代码实现 | ||||||
|  |  | ||||||
| combiner的代码实现比较简单,只要在组装作业时,添加下面一行代码即可 | combiner的代码实现比较简单,只要在组装作业时,添加下面一行代码即可: | ||||||
|  |  | ||||||
| ```java | ```java | ||||||
| // 设置Combiner | // 设置Combiner | ||||||
| @@ -403,7 +393,7 @@ job.setCombinerClass(WordCountReducer.class); | |||||||
|  |  | ||||||
| <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个),这一点从图中日志的`reduce input records`参数也可以看出来。在这个用例中combiner的效果就非常明显。 | 这里我们只有一个输入文件并且小于128M,所以只有一个Map进行处理,可以看到经过combiner后,records由3519降低为6(样本中单词种类就只有6个),在这个用例中combiner就能极大地降低需要传输的数据量。 | ||||||
|  |  | ||||||
| ## 5.2 Partitioner | ## 5.2 Partitioner | ||||||
|  |  | ||||||
|   | |||||||
										
											Binary file not shown.
										
									
								
							| Before Width: | Height: | Size: 84 KiB After Width: | Height: | Size: 80 KiB | 
		Reference in New Issue
	
	Block a user