目录

7.3 MapReduce工作流程

 7.3.1 工作流程概述

 7.3.2 MapReduce各个执行阶段

 7.3.3 Shuffle过程详解

1. Shuffle过程简介（过程分为Map端的操作和Reduce端的操作）

2、Map端的Shuffle过程：

3、在Reduce端的Shuffle过程

7.4  实例分析：WordCount

7.5 MapReduce的具体应用

小结

---

7.3 MapReduce工作流程

大规模数据集的处理包括：分布式存储和分布式计算两个核心环节。

（1）谷歌使用GFS实现分布式数据存储，用MapReduce实现分布式计算；

（2）Hadoop使用HDFS实现分布式数据存储，用Hadoop MapReduce实现分布式计算。

 7.3.1 工作流程概述

 7.3.2 MapReduce各个执行阶段

HDFS 以固定大小的block 为基本单位存储数据，而对于MapReduce 而言，其处理单位是split。split 是一个逻辑概念，它只包含一些元数据信息，比如数据起始位置、数据长度、数据所在节点等。划分方法完全由用户自己决定。

 7.3.3 Shuffle过程详解

Shuffle是MapReduce整个工作流程的核心环节。

1. Shuffle过程简介（过程分为Map端的操作和Reduce端的操作）

Shuffle是对Map任务输出结果进行分区、排序、合并、归并等处理并交给Reduce的过程。

2、Map端的Shuffle过程：

（1）输入数据和执行map任务

输入数据一般保存在分布式文件系统的数据块中，比如HDFS中。Map任务接收<key，value>作为输入，然后转换为多个<key，value>输出。

（2）写入缓存

每个Map任务分配一个缓存，Map的输出先放入缓存，累计后，一次性批量写入磁盘，减小对磁盘的I/O开销。多次写入需要多次寻址，一次写入需一次寻址即可。

（3）溢写（分区、排序和合并）

MapReduce的缓存容量有限，默认100MB，一旦占满，就启动溢写（单独后台线程，不影响缓存写入），一次性写入磁盘，并清空缓存。但为让缓存一直有可用空间，可设置溢写比例，比如0.8，100内存，80M时就启动溢写过程，剩下的20M供Map继续写入。

（3）溢写（分区、排序和合并）

分区：缓存中的数据写入磁盘前，先分区。缓存中的数据是<key,value>键值对，通过Partitioner接口对这些键值对进行分区，默认的分区方式：采用hash函数对key进行哈希后再用Reduce任务的数量进行取模，表示为hash（key）mod R，这样就可以把Map的输出结果均匀的分配给Reduce任务区并行处理。也允许用户通过重载Partitioner接口来定义分区方式。

（3）溢写（分区、排序和合并）

排序（默认）：每个分区内的键值对，后台线程会根据key，进行内存排序（Sort）。

合并（可选）：如用户事先没有定义Combiner函数，就不用进行合并操作。如果定义了，会执行合并操作，从而减少了需要溢写到磁盘的数据量。“合并”是指具有相同key的<key，value>的value加起来。如：<a,1> , <a,1> => <a,2> 发生在Map端，有别与Reduce。Combiner的输出是Reduce的输入，Combiner不能改变Reduce任务最终的计算结果。

（4）文件归并

Map任务全部结束前，系统会对所有溢写文件进行归并（Merge），生成一个大的溢写文件（键值对都经过分区和排序）。 

“归并”指对于具有相同key的键值对归并成一个新的键值对。如<a,1> , <a,1> => <a, <1,1>>。若干个相同key键值对<k1,v1>,<k1,v2>…<k1,vn>会被归并为一个新的键值对<k1,<v1,v2,v3,…vn>>。

进行归并时，如磁盘生成的溢写文件数量超过参数min.num.spills.for.combine的值时（默认是3，用户可以修改这个值），就可再次运行Combiner，对数据进行合并，减少磁盘的数据量。如果写磁盘中只有一两个溢写文件，就不会运行Combiner，因为执行合并操作本身也有代价。

经过（1）、（2）、（3）、（4），Map的Shuffle过程全部完成，最终形成一个大文件（被分区的），不同分区被送到不同的Reduce任务进行并行处理，JobTracker会一直进行Map任务检测，当一个任务完成，会立刻通知Reduce及时领取数据，开始Reduce的Shuffle过程。

合并（Combine）和归并（Merge）的区别：

两个键值对<“a”,1>和<“a”,1>，如果合并，会得到<“a”,2>，如果归并，会得到<“a”,<1,1>>

3、在Reduce端的Shuffle过程

（1）“领取”数据：Map端的Shuffle结束后，所有Map的输出结果都保存在Map机器的本地磁盘上，文件都是被分区的，不同的分区会被发送到不同的Reduce任务进行并行处理。每个Reduce任务会不断地通过RPC向JobTracker询问Map任务是否已经完成；JobTracker检测到一个Map任务完成后，就会通知相关的Reduce任务来"领取"数据；Reduce收到通知，就会从Map任务所在机器把属于自己的分区数据领取到本地磁盘。一般是Reduce任务使用多个线程通过是多个Map机器领回数据。

（2）归并数据：Map端领取的数据会被存放在Reduce端的缓存中，如果缓存被占满，就会溢写到磁盘。缓存数据来自不同Map机器，会存在很多合并（Combiner）的键值对，当溢写启动时，相同key的键值对会被归并，如用户定义Combiner，则归并后的数据可以执行合并操作，减少写入磁盘数据量。一次溢写，生成一个溢写文件，溢写结束，磁盘上存在多个溢写文件。   Map端数据都被领回时，多个溢写文件会被归并成一个大文件，归并时会进行排序。如果数据量很少就不需要进行溢写，直接在内存中执行归并操作。

了解：把磁盘多个溢写文件归并成一个大文件可能需要执行多轮归并操作，每轮归并操作可以归并文件数量是由参数io.sort.factor的值来控制的（默认是10，可以修改）。假设磁盘中生成50个溢写文件，每轮可以归并10个溢写文件，则需要经过5轮归并，得到5个归并后的大文件。

（2）把数据输入给Reduce任务：Reduce任务会执行Reduce函数中定义的各种映射，输出最终结果，保存在分布式文件系统中（比如GFS或HDFS）

了解：磁盘多轮归并后得到若干个大文件，不会归并成一个新的大文件，而是直接输入给Reduce任务，可减少磁盘读写开销

7.4  实例分析：WordCount

程序	WordCount
输入	一个包含大量单词的文本文件
输出	文件中每个单词及其出现次数（频数），并按照单词字母顺序排序，每个单词和其频数占一行，单词和频数之间有间隔

实例：

输入	输出
Hello World Hello Hadoop Hello MapReduce	Hadoop 1 Hello 3 MapReduce 1 World 1

7.5 MapReduce的具体应用

MapReduce可以很好地应用于各种计算问题

小结