Hadoop大数据实战系列文章之Mapreduce 计算框架 - 大数据测试

如果将Hadoop比做一头大象，那么MapReduce就是那头大象的电脑。MapReduce是 Hadoop 核心编程模型。在 Hadoop 中，数据处理核心就是 MapReduce 程序设计模型。

本章内容：

1) MapReduce 编程模型

2) MapReduce 执行流程

3) MapReduce 数据本地化

4) MapReduce 工作原理

5) MapReduce 错误处理机制

1. MapReduce 编程模型

Map和Reduce的概念是从函数式变成语言中借来的，整个MapReduce计算过程分为 Map 阶段和Reduce阶段，也称为映射和缩减阶段，这两个独立的阶段实际上是两个独立的过程，即 Map 过程和 Reduce 过程，在 Map 中进行数据的读取和预处理，之后将预处理的结果发送到 Reduce 中进行合并。

我们通过一个代码案例，让大家快速熟悉如何通过代码，快速实现一个我们自己的MapReduce。

案例：分布式计算出一篇文章中的各个单词出现的次数，也就是 WordCount。

1) 创建 map.py 文件，写入以下代码：

 #!/usr/bin/env python import sys word_list = [] for line in sys.stdin: word_list = line.strip().split(' ') if len(word_list) <= 0: continue for word in word_list: w = word.strip() if len(w) <= 0: continue print 't'.join([w, "1"])

该代码主要工作是从文章数据源逐行读取，文章中的单词之间以空格分割，

word_list = line.strip().split(' ')这块代码是将当前读取的一整行数据按照空格分割，将分割后的结果存入 word_list 数组中，然后通过 for word in word_list 遍历数组，取出每个单词，后面追加“1”标识当前 word 出现 1 次。

2) 创建 reduce.py，写入以下代码：

#!/usr/bin/env python import sys cur_word = None sum_of_word = 0 for line in sys.stdin: ss = line.strip().split('t') if len(ss) != 2: continue word = ss[0].strip() count = ss[1].strip() if cur_word == None: cur_word = word if cur_word != word: print 't'.join([cur_word, str(sum_of_word)]) sum_of_word = 0 cur_word = word sum_of_word += int(count) print 't'.join([cur_word, str(sum_of_word)]) sum_of_word = 0

该代码针对 map 阶段的数组进行汇总处理，map 到 reduce 过程中默认存在shuffle partition 分组机制，保证同一个 word 的记录，会连续传输到 reduce 中，所以在 reduce阶段只需要对连续相同的 word 后面的技术进行累加求和即可。

3) 本地模拟测试脚本：

$ cat big.txt | python map.py | sort -k1 | python reduce.py cat 1 run 3 see 2 spot 2 the 1

6) 脚本执行流程：

see spot run run spot run see the cat see spot run see the cat run spot run see,1 spot,1 run,1 run,1 spot,1 run,1 see,1 the,1 cat,1 see,1 see,1 spot,1 spot,1 run,1 run,1 run,1 the,1 cat,1 see,1 spot,1 run,1 the,1 cat,1 cat 1 run 3 see 2 spot 2 the 1

2. MapReduce 执行流程

上面的例子属于 MapReduce 计算框架的一般流程，经过整理总结：

1) 输入和拆分：

不属于 map 和 reduce 的主要过程，但属于整个计算框架消耗时间的一部分，该部分会为正式的 map 准备数据。

分片（split）操作：

split 只是将源文件的内容分片形成一系列的 InputSplit，每个 InputSpilt中存储着对应分片的数据信息（例如，文件块信息、起始位置、数据长度、所在节点列表…），并不是将源文件分割成多个小文件，每个 InputSplit 都由一个 mapper 进行后续处理。

每个分片大小参数是很重要的，splitSize 是组成分片规则很重要的一个参数，该参数由三个值来确定：

 minSize：splitSize 的最小值，由 mapred-site.xml 配置文件中

mapred.min.split.size 参数确定。

 maxSize：splitSize 的最大值，由 mapred-site.xml 配置文件中

mapreduce.jobtracker.split.metainfo.maxsize 参数确定。

 blockSize：HDFS 中文件存储的快大小，由 hdfs-site.xml 配置文件中

dfs.block.size 参数确定。

splitSize 的确定规则：splitSize=max{minSize，min{maxSize，blockSize}}

数据格式化（Format）操作：

将划分好的 InputSplit 格式化成键值对形式的数据。其中key为偏移量，value 是每一行的内容。

值得注意的是，在 map 任务执行过程中，会不停的执行数据格式化操作，每生成一个键值对就会将其传入 map，进行处理。所以 map和数据格式化操作并不存在前后时间差，而是同时进行的。

2) Map 映射：

是 Hadoop 并行性质发挥的地方。根据用户指定的 map过程，MapReduce 尝试在数据所在机器上执行该 map 程序。在 HDFS 中，文件数据是被复制多份的，所以计算将会选择拥有此数据的最空闲的节点。

在这一部分，map 内部具体实现过程，可以由用户自定义。

3) Shuffle 派发：

Shuffle 过程是指 Mapper 产生的直接输出结果，经过一系列的处理，成为最终的Reducer直接输入数据为止的整个过程。这是mapreduce的核心过程。该过程可以分为两个阶段：

Mapper 端的 Shuffle：由 Mapper 产生的结果并不会直接写入到磁盘中，而是先存储在内存中，当内存中的数据量达到设定的阀值时，一次性写入到本地磁盘中。并同时进行sort（排序）、combine（合并）、partition（分片）等操作。其中，sort 是把 Mapper 产生的结果按照key值进行排序；combine是把key值相同的记录进行合并；partition是把数据均衡的分配给 Reducer。

Reducer 端的Shuffle：由于Mapper 和Reducer往往不在同一个节点上运行，所以Reducer 需要从多个节点上下载 Mapper 的结果数据，并对这些数据进行处理，然后才能被 Reducer 处理。

4) Reduce 缩减：

Reducer接收形式的数据流，形成形式的输出，具体的过程可以由用户自定义，最终结果直接写入 hdfs。每个 reduce 进程会对应一个输出文件，名称以 part-开头。

3. MapReduce 数据本地化（Data-Local ）

首先，HDFS 和 MapReduce 是 Hadoop 的核心设计。对于 HDFS，是存储基础，在数据层面上提供了海量数据存储的支持。而 MapReduce，是在数据的上一层，通过编写MapReduce 程序对海量数据进行计算处理。

在前面 HDFS 章节中，知道了 NameNode 是文件系统的名字节点进程，DataNode是文件系统的数据节点进程。

MapReduce计算框架中负责计算任务调度的JobTracker对应HDFS的NameNode的角色，只不过一个负责计算任务调度，一个负责存储任务调度。

MapReduce计算框架中负责真正计算任务的TaskTracker对应到HDFS的DataNode的角色，一个负责计算，一个负责管理存储数据。

考虑到“本地化原则”，一般地，将 NameNode 和 JobTracker 部署到同一台机器上，各个 DataNode 和 TaskNode 也同样部署到同一台机器上。

这样做的目的是将 map 任务分配给含有该 map 处理的数据块的 TaskTracker上，同时将程序 JAR 包复制到该 TaskTracker 上来运行，这叫“运算移动，数据不移动”。而分配reduce 任务时并不考虑数据本地化。

4. MapReduce 工作原理

我们通过 Client、JobTrask 和 TaskTracker 的角度来分析 MapReduce 的工作原理：

首先在客户端（Client）启动一个作业（Job），向 JobTracker 请求一个 Job ID。将运行作业所需要的资源文件复制到 HDFS 上，包括 MapReduce 程序打包的 JAR文件、配置文件和客户端计算所得的输入划分信息。这些文件都存放在JobTracker专门为该作业创建的文件夹中，文件夹名为该作业的 Job ID。 JAR 文件默认会有 10 个副本

（mapred.submit.replication属性控制）；输入划分信息告诉了JobTracker应该为这个作业启动多少个 map 任务等信息。

JobTracker 接收到作业后，将其放在一个作业队列里，等待作业调度器对其进行调度当作业调度器根据自己的调度算法调度到该作业时，会根据输入划分信息为每个划分创建一个 map 任务，并将 map 任务分配给 TaskTracker 执行。对于 map 和 reduce 任务，TaskTracker 根据主机核的数量和内存的大小有固定数量的 map 槽和 reduce 槽。这里需要强调的是：map 任务不是随随便便地分配给某个 TaskTracker 的，这里就涉及到上面提

到的数据本地化（Data-Local）。

TaskTracker每隔一段时间会给JobTracker发送一个心跳，告诉JobTracker它依然在运行，同时心跳中还携带着很多的信息，比如当前 map 任务完成的进度等信息。当JobTracker收到作业的最后一个任务完成信息时，便把该作业设置成“成功”。当JobClient查询状态时，它将得知任务已完成，便显示一条消息给用户。

如果具体从 map 端和 reduce 端分析，可以参考上面的图片，具体如下：

Map 端流程：

1) 每个输入分片会让一个 map 任务来处理，map 输出的结果会暂且放在一个环形内存缓冲区中（该缓冲区的大小默认为 100M，由 io.sort.mb 属性控制），当该缓冲区快要溢出时（默认为缓冲区大小的80%，由io.sort.spill.percent属性控制），会在本地文件系统中创建一个溢出文件，将该缓冲区中的数据写入这个文件。

2) 在写入磁盘之前，线程首先根据reduce任务的数目将数据划分为相同数目的分区，也就是一个reduce任务对应一个分区的数据。这样做是为了避免有些reduce任务分配到大量数据，而有些reduce任务却分到很少数据，甚至没有分到数据的尴尬局面。其实分区就是对数据进行hash的过程。然后对每个分区中的数据进行排序，如果此时设置了 Combiner，将排序后的结果进行 Combine 操作，这样做的目的是让尽可能少的数据写入到磁盘。

3) 当 map 任务输出最后一个记录时，可能会有很多的溢出文件，这时需要将这些文件合并。合并的过程中会不断地进行排序和 Combine 操作，目的有两个：

 尽量减少每次写入磁盘的数据量；

 尽量减少下一复制阶段网络传输的数据量。

最后合并成了一个已分区且已排序的文件。为了减少网络传输的数据量，这里可以将数据压缩，只要将 mapred.compress.map.out 设置为 true 就可以了。

4) 将分区中的数据拷贝给相对应的 reduce 任务。分区中的数据怎么知道它对应的reduce 是哪个呢？其实 map 任务一直和其父 TaskTracker 保持联系，而TaskTracker又一直和JobTracker保持心跳。所以JobTracker中保存了整个集群中的宏观信息。只要reduce任务向JobTracker获取对应的map输出位置就可以了。

Reduce端流程：

1) Reduce 会接收到不同 map 任务传来的数据，并且每个 map 传来的数据都是有序的。如果 reduce 端接受的数据量相当小，则直接存储在内存中（缓冲区大小由mapred.job.shuffle.input.buffer.percent属性控制，表示用作此用途的堆空间的百分比），如果数据量超过了该缓冲区大小的一定比例（由

mapred.job.shuffle.merge.percent 决定），则对数据合并后溢写到磁盘中。

2) 随着溢写文件的增多，后台线程会将它们合并成一个更大的有序的文件，这样做是为了给后面的合并节省时间。其实不管在 map 端还是 reduce 端，MapReduce都是反复地执行排序，合并操作，所以排序是 hadoop 的灵魂。

3) 合并的过程中会产生许多的中间文件（写入磁盘了），但MapReduce会让写入磁盘的数据尽可能地少，并且最后一次合并的结果并没有写入磁盘，而是直接输入到reduce 函数。

在 Map 处理数据后，到Reduce得到数据之前，这个流程在MapReduce中可以看做是一个 Shuffle 的过程。

在经过 mapper 的运行后，我们得知 mapper 的输出是这样一个 key/value 对。到底当前的 key 应该交由哪个 reduce 去做呢，是需要现在决定的。 MapReduce 提供Partitioner 接口，它的作用就是根据 key 或value 及 reduce的数量来决定当前的这对输出数据最终应该交由哪个 reduce task 处理。默认对 key 做 hash 后再以 reduce task 数量取模。默认的取模方式只是为了平均 reduce 的处理能力，如果用户自己对 Partitioner有需求，可以订制并设置到 job 上。

5. MapReduce 错误处理机制

MapReduce任务执行过程中出现的故障可以分为两大类：硬件故障和任务执行失败引发的故障。

1) 硬件故障

在 Hadoop Cluster 中，只有一个 JobTracker，因此，JobTracker 本身是存在单点故障的。如何解决JobTracker的单点问题呢？我们可以采用主备部署方式，启动JobTracker主节点的同时，启动一个或多个 JobTracker 备用节点。当 JobTracker主节点出现问题时，通过某种选举算法，从备用的 JobTracker 节点中重新选出一个主节点。

机器故障除了 JobTracker 错误就是 TaskTracker 错误。TaskTracker 故障相对较为常见，MapReduce 通常是通过重新执行任务来解决该故障。

在 Hadoop 集群中，正常情况下，TaskTracker 会不断的与 JobTracker 通过心跳机制进行通信。如果某 TaskTracker 出现故障或者运行缓慢，它会停止或者很少向 JobTracker发送心跳。如果一个TaskTracker在一定时间内（默认是1分钟）没有与JobTracker通信，那么 JobTracker 会将此 TaskTracker 从等待任务调度的 TaskTracker 集合中移除。同时JobTracker 会要求此 TaskTracker 上的任务立刻返回。如果此 TaskTracker 任务仍然在

mapping 阶段的Map任务，那么JobTracker会要求其他的TaskTracker重新执行所有原本由故障 TaskTracker 执行的 Map 任务。如果任务是在 Reduce 阶段的 Reduce 任务，那么JobTracker会要求其他TaskTracker重新执行故障TaskTracker未完成的Reduce任务。

比如:一个 TaskTracker 已经完成被分配的三个 Reduce 任务中的两个，因为 Reduce 任务一旦完成就会将数据写到 HDFS 上，所以只有第三个未完成的 Reduce 需要重新执行。但是对于 Map 任务来说，即使TaskTracker 完成了部分Map，Reduce仍可能无法获取此节点上所有 Map 的所有输出。所以无论Map 任务完成与否，故障TaskTracker上的Map 任务都必须重新执行。

2) 任务执行失败引发的故障

在实际任务中，MapReduce作业还会遇到用户代码缺陷或进程崩溃引起的任务失败等情况。用户代码缺陷会导致它在执行过程中抛出异常。此时,任务 JVM 进程会自动退出，并向 TaskTracker 父进程发送错误消息，同时错误消息也会写入 log 文件，最后 TaskTracker将此次任务尝试标记失败。对于进程崩溃引起的任务失败，TaskTracker的监听程序会发现进程退出，此时TaskTracker也会将此次任务尝试标记为失败。对于死循环程序或执行时间

太长的程序，由于 TaskTracker 没有接收到进度更新，它也会将此次任务尝试标记为失败，并杀死程序对应的进程。

在以上情况中，TaskTracker将任务尝试标记为失败之后会将TaskTracker自身的任务计数器减 1，以便想 JobTracker 申请新的任务。TaskTracker 也会通过心跳机制告诉JobTracker本地的一个任务尝试失败。JobTracker接到任务失败的通知后，通过重置任务状态，将其加入到调度队列来重新分配该任务执行（JobTracker 会尝试避免将失败的任务再次分配给运行失败的 TaskTracker）。如果此任务尝试了 4 次（次数可以进行设置）仍没有完成，就不会再被重试，此时整个作业也就失败了