MapReduce 阅读笔记

这篇文章是我阅读 MapReduce 论文：《MapReduce: Simplified Data Processing on Large Clusters》的笔记，这篇笔记概述了 MapReduce 是什么，它的工作流程，一些细节问题，以及我的个人理解与思考。

《MapReduce: Simplified Data Processing on Large Clusters》： https://research.google.com/archive/mapreduce-osdi04.pdf

MapReduce 是什么？

MapReduce 是 Google设计的一种用于大规模数据集的分布式模型，它具有支持并行计算、容错、易使用等特点。它的设计目标如下：

• 支持并行
• 用于分布式
• 能够进行错误处理（比如机器崩溃）
• 易于使用（程序员友好）
• 负载均衡

模型流程

MapReduce 模型主要分为 2 个部分：Map 和 Reduce。

在 Map 过程中，Map 函数会获取输入的数据，产生一个临时中间值，它是一个 K/V 对，然后MapReduce Library 会按 Key 值给键值对（K/V）分组然后传递给 Reduce 函数。而后，Reduce 接收到了这些 K/V 对，会将它们合并。

以论文中的字数统计程序为例：

现在我们来考虑，如果我们有许多文档，然后我们想要统计在这些文档中每个字出现的次数，现在用 MapReduce 来解决这个问题。Map 函数所做的工作，就是进行分词，产生一组形如下表的 K/V 键值对：

然后将这组键值对传递给 Reduce，由 Reduce 进行合并。

具体流程如下：

1. 由用户程序中调用的 MapReduce Library 将文件分成 M 块（M 要远大于 Map Worker 的数量，每块大小16MB~64MB），此时，进入 MapReduce 过程；
2. 由 Master 给空闲的 Worker 分配任务，共有 M 个 Map 任务，R 个 Reduce 任务；
3. Map Worker 读取文件，将文件处理为 K/V 键值对，K/V 键值对缓存于内存中（此时存在一个问题，如果断电怎么办？往下看后边有解释）；
4. 将缓存于内存的 K/V 键值对写入磁盘，分成 R 堆（分堆方法有很多种，论文中提到了使用 Hash 散列函数），然后将结果发送给 Master；
5. Master 将这些 K/V 键值对的存储地址告知 Reduce，Reduce Worker 通过 RPC（远程过程调用）进行读取，读取完毕之后会根据 Key 值进行排序（这样，相同 Key 值的就会在一起。但是存在一个问题，如果内存不够大，排序该怎么进行？可以使用外部排序）；
6. Reduce Worker 将已经排序的结果进行遍历，将每个 Key 值所对应的一组 Value，所组成的 <key, value[num]>传递给用户所编写的 reduce 函数进行处理；
7. 所有的 Map，Reduce 任务都完成后，告知用户程序，MapReduce 已经结束，返回用户程序。

容错处理（Fault-Tolerance）

MapReduce 中的容错处理是非常重要的，因为MapReduce 是运行于分布式环境中的，在分布式环境中经常会有机器出现错误，我们不能让个别机器的错误影响到整体。

Worker 崩溃

Master 通过定期给 Worker 发送心跳（heartbeat）来检测 Worker 是否还在正常工作，如果 Worker 无应答或者是应答有误，我们认定它已经宕机（fail）。如果正在工作的 Worker 宕机了，那么运行在它上面的 map 任务会进行初始化（初始状态为 idle，任务还有其他2种状态，in-progress处理中，completed 已完成），重新被分配到正常的 Worker 上。

如果说 Map Worker 已经完成了一些工作，我们仍然要对运行在它上面的所有任务重新进行分配，这是为什么呢？这里同时可以解决上面的那个问题。因为 Map Worker 处理后的中间结果存在于内存中，或者是 local disk 中，一旦它宕机，这些数据就获取不到了。

但是对于 Reduce Worker，它完成的任务不用重做，因为它处理后的结果是保存在全局存储中的。

如果，在 Map Worker A 宕机之后，它所做的任务被重新分配给了 Map Worker B，后边的 Reduce Worker 会被告知，A 已经宕机，要去 B 去读取数据。

Master 崩溃

如果说 MapReduce 的 Master 宕机了，又该如何处理呢？

MapReduce 中的 Master 会定期进行 checkpoint 备份，如果 Master 宕机，会根据之前的 checkpoint 进行恢复，但是恢复期间，MapReduce 任务会中断。

一些细节问题

• 考虑用户编写的 reduce 函数是确定的（deterministic，对于同样的输入执行的结果是一样的），如果有多个 Reduce Worker 都执行了一个 Reduce 任务该怎么办？

因为用户的 reduce 函数是 deterministic 的，所以即使有多个 Reduce Worker 都执行了同一个任务，但是它们执行的结果都是一样的，并不影响最后的结果。

• 如果用户编写的 reduce 函数是不确定（non-deterministic）的呢？

正是因为 reduce 函数是 non-deterministic 的，本来每次执行的结果也不确定，所以更不会产生影响。

• 我们所需要处理的输入文件是如何保存的？

Input 文件保存于 GFS 中，GFS 会将它们分块保存（每块16MB~64MB），GFS 会对每个文件有3个备份，备份在不同的机器上。

• Master 是如何分配任务的？

遵循就『近』原则，将任务分配给离任务所保存的位置最『近』的 Worker，这里对『近』的定义是网络层面上的，比如说在同一个交换机下的两个机器就是距离『近』的。

• MapReduce 是如何做到负载均衡的？

一开始将文件分块时，分为 M 块，远大于 Map Worker 的数量就有助于负载均衡。同时，这样做还有一个好处，就是当一个 Worker 宕机的时候，可以将任务迅速分配开来，分到多个 Worker 上去。如果 M 比较小，有可能当一个 Worker 宕机时，它的任务不够分配到剩下的 Worker 中，会有 Worker 闲置。

• 如何解决 straggler 问题（其他 Worker 都已经完成了自己的任务，但是有一个异常慢的机器，它还有任务没完成，拖慢了整体的速度）？

MapReduce 有一种机制应对这种情况：MapReduce 会对未完成的任务（in-progress） 定时执行备份执行操作（即，把这些正在某些 Worker 上执行但未完成的任务再次分配给其他 Worker 去执行），不论这个任务被哪个 Worker 完成都会被标记为已完成。

• 如果在 Map 任务中有一个 key 特别多，可能会拖慢整个网络的速度，该怎么办？（例如，在字数统计的例子中，the 这个词的数量特别多）

MapReduce 给用户提供了一个 Combiner 函数，这个函数可以将结果在发送到网络之前进行合并，例如发送键值对<”by”, 3>。

原文：http://blog.luoyuanhang.com/2017/04/19/mapreduce-notes/