mit6824-01-MapReduce详解

MapReduce简述

MapReduce是一个在多台机器上并行计算大规模数据的软件架构。主要通过两个操作来实现：Map 和 Reduce：用于大规模数据集（大于1TB）的并行运算。

概念"Map（映射）“和"Reduce（归约）”，是它们的主要思想，都是从函数式编程语言里借来的，还有从矢量编程语言里借来的特性。它极大地方便了编程人员在不会分布式并行编程的情况下，将自己的程序运行在分布式系统上。 当前的软件实现是指定一个Map（映射）函数，用来把一组键值对映射成一组新的键值对，指定并发的Reduce（归约）函数，用来保证所有映射的键值对中的每一个共享相同的键组。

编程模型

计算采用一组输入 K/V 对，并产生一组输出 K/V 对。MapReduce 库将计算表示为由用户编写的两个函数：

1. map()：根据输入生成一组中间 K/V 对。 MapReduce 库将所有具有相同 Key ki的 K/V 对传递给 Reduce 函数；reduce()：
2. 将 ki对应的所有值合并成 Value Set，并能通过迭代器访问；

map-reduce经典举例即统计字母出现的次数，多个进程各自通过map函数统计获取到的数据片段的字母的出现次数；后续再通过reduce函数，汇总聚合map阶段下每个进程对各自负责的数据片段统计的字母出现次数。伪代码如下所示：

map(String key, String Value):// key: document name// value: document contentsfor each word w in value:EmitIntermediate(w, "1");reduce(String key, Iterator values):// key: a word// value: a list of countsint result = 0;for each v in values:result += ParseInt(v);Emit(AsString(result));// map 函数发出每个单词及其出现次数。 reduce 函数统计特定单词发出的所有次数。

执行流程

执行流程

MapReduce操作总执行流程如下：

• Master进程，被称为coordinator协调器，负责orchestrate编排wokers，把map jobs分配给它们
• reduce、map被称为task任务

1. MapReduce 库首先将输入文件 split 成 M 块，然后，它会在集群上启动该程序的多个副本；
2. 其中一个副本是 Master（在 lab1 中称为 Coordinator），其余的是由 Master 分配工作的 Worker。有 M 个 map 任务和 R 个 reduce 任务要分配。 Master 为每个空闲的 Worker 分配一个任务；
3. map Worker 从输入数据中解析出 K/V 对，并将每一对传递给 map 函数，生成的中间 K/V 对并缓存在内存中；
4. 缓冲的 K/V 对定期被写入本地磁盘，由分区函数划分为 R 个区域。Master 获取这些缓冲对的位置并负责将这些位置转发给 reduce Worker；
5. 当 Master 通知 reduce Worker 这些位置时，它使用 RPC 从 map Worker 的本地磁盘读取缓冲数据。当 reduce Worker 读取所有中间数据时，它会根据中间键对其进行排序，以便将所有出现的相同键组合在一起；
6. reduce Worker 迭代排序的中间数据，对于遇到的每个唯一中间键，它将键和相应的中间值集传递给用户的 reduce 函数。 reduce 函数输出 append 到对应分区的最终文件中；
7. 当所有的 map 任务和 reduce 任务都完成后，Master 唤醒用户程序，从 MapReduce 调用中返回；

排序保证

MapReduce 保证在给定的分区内，中间 K/V 对以 Key 递增顺序进行排列。从而保证每个分区的输出文件也是有序的——这在输出文件需要支持按键随机访问查找时很有用，同时可以对不同文件使用归并排序。具体实现时机：

Map Worker 中最后进行一次排序。

Combiner函数

在某些情况下，Map 函数产生的中间 key 值的重复数据会占很大的比重（例如词频统计，将产生成千上万的 <the, 1> 记录）。用户可以自定义一个可选的 Combiner 函数，Combiner 函数首先在本地将这些记录进行一次合并，然后将合并的结果再通过网络发送出去。

Combiner 函数的代码通常和 Reduce 函数的代码相同，启动这个功能的好处是可以减少通过网络发送到 Reduce 函数的数据量。

Master数据结构

Master 存储所有任务的状态（ idle 、 in-progress 或 completed ）和分配给所有工作机器执行的任务，以及由 map 任务生成的 R 个中间文件区域的位置和大小。当 map 任务完成时，对此位置和大小信息的更新被递增地推送给 reduce Worker。

容错性

Worker故障

Master 周期性地 ping 每个 Worker，如果 Worker 超时未回应，则将其标记为 Failed。
Worker 故障容错遵循以下原则：

• map 任务被 Worker 完成后将重置为 idle 状态，以便调度给其他 Worker；

• Failed Worker 未完成的任何任务将重置为 idle 状态；
• Failed Worker 已完成的 map 任务将重新执行，因为它们的输出存储在故障机器的本地磁盘上，无法访问；
• Failed Worker 已完成的 reduce 任务无需重新执行，因为它们的输出存储在全局文件系统中；
• 当一个 map 任务因 WorkerA 失败转而由 WorkerB 执行，所有 reduce Worker 会收到重新执行的通知，任何尚未从 Worker A 读取数据的 reduce 任务将从 Worker B 读取数据；

Master故障

对于Master故障，我查到的资料显示：

Master 故障：中止整个 MapReduce 运算，重新执行。一般很少出现 Master 故障。

Google当初设计MapReduce时设计协调器不允许失败。如果协调器真的失败了，整个job（包含具体的多个map、reduce步骤task）需要重新运行。在这篇论文中，没有谈论到协调器失败后他们的应对方式。

（协调器不允许失败）这使得容错性很难做得更高，因为它维护一些工作状态（每个map、reduce函数执行的状态），在论文的库中，协调器不能失败。

后面会谈论一些技术手段，可以实现协调器容错，他们可以这么做却不打算，原因是他们认为比起协调器，运行map函数的上千个机器中崩溃一台的概率更高（也就是收益和成本不成正比，所以暂时没有实现协调器容错的打算）。

性能提升

定制分区函数

MapReduce 库的用户可以自定义分区函数来应对不同应用场景。例如，使用 hash(Hostname(urlkey)) % R 作为分区函数可以使来自同一主机的所有 URL 最终出现在同一个输出文件中。

局部性

就近原则：
Google发表该 paper 时，网络带宽是一个相当匮乏的资源。Master 在调度 Map 任务时会考虑输入文件的位置信息，尽量将一个 Map 任务调度在包含相关输入数据拷贝的机器上执行；如果找不到，Master 将尝试在保存输入数据拷贝的附近的机器上执行 Map 任务。

需要注意的是，新的讲座视频提到，随着后来 Google 的基础设施的扩展和升级，他们对这种存储位置优化的依赖程度降低了。

执行缓慢的worker(slow workers)

MapReduce操作所用总时间受短板效应影响：
比如GFS也在同一台机器上运行占用大量的机器周期或带宽，或硬件本身问题，导致worker执行map/reduce很慢。慢的worker被称为straggler，当剩下几个map/reduce任务没有执行时，协调者会另外分配相同的map/reduce任务到其他闲置worker上运行，达到backup task（备份任务）的效果（因为函数式，map/reduce以相同输入执行最后会产生相同输出，所以执行多少次都不会有问题）。

• 通过backup task，性能不会受限于最慢的几个worker，因为有更快的worker会领先它们完成task（map或reduce）。这是应对straggler的普遍做法，通过replicate tasks复制任务，获取更快完成task的输出结果，处理了tail latency尾部延迟问题。

常见问题总结回顾

• 在远程读取进程中，文件是否会传输到reducer？

会。map函数产生的中间结果存放在执行map函数的worker机器的磁盘上，而之后解调器分配文件给reducer执行reduce函数时，中间结果数据需要通过网络传输到reducer机器上。这里其实很少有网络通信，因为一个worker在一台机器上，而每台机器同时运行着worker进程和GFS进程。worker运行map产生中间结果存储在本地，而之后协调器给worker分配文件以执行reduce函数时，才需要通过网络获取中间结果数据，最后reduce处理完在写入GFS，写入GFS的动作也往往需要通络传输。

• 协调器是否负责对数据进行分区，并将数据分发到每个worker或机器上？

不是的。mapreduce运行用户程序，这些输入数据在GFS中。（也就是说协调器告知worker从GFS取哪些数据进行map，后续协调器又告知worker从哪些worker机器上获取中间结果数据进行reduce，最后又统一写入到GFS中）

• 排序是如何工作的？比如谁负责排序，如何排序？

中间结果数据传递到reduce函数之前，mapreduce库进行一些排序。比如所有的中间结果键a、b、c到一个worker。比如(a,1) (b,1) (c,1) (a,1) 数据，被排序成(a,1) (a,1) (b,1) (c,1) 后才传递给reduce函数。

• map是否会被执行两次？

• 如果coordinator没有收到worker反馈task任务完成，那么会coordinator重新分配worker要求执行task（可能分配到同一个worker，重点是task会被重新执行）
• 或许没反馈task执行done完成的worker是遇到网络分区等问题，并没有宕机，或者协调者不能与worker达成网络通信，但实际上worker仍然在运行map任务，它正在产生中间结果。

同一个map可以被运行两次。 被执行两次是能够接受的（幂等性问题），正是map和reduce属于函数式(functional)的原因之一。如果map/reduce是一个funcitonal program，那么使用相同输入运行时，产生的输出会是相同的（也就是保证幂等）。

• reduce能够被执行两次吗？

reduce和map出于相同的原因，从容错的角度上看，执行reduce函数和map函数并没有太大区别。需要注意的是，这时候可能有两个reducer同时有相同的输出文件需要写入GFS，它们首先在全局文件系统GFS中产生一个中间文件，然后进行atomic rename原子重命名，将文件重命名为实际的最终名称。因为在GFS中执行的重命名是原子操作，最后哪个reducer胜出并不重要，因为reduce是函数式的，它们最终输出的数据都是一样的。

• 一台机器应该可以执行多个map任务，如果它分配10个map任务，而在执行第7个map任务时失败了，master得知后，会安排将这7个已完成的map任务分布式地重新执行，可能分散到不同的map机器上，对吗？

是的。但是通常一台机器只运行一个map函数或一个reduce函数，而不是多个。

• 在worker完成map任务后，它是否会直接将文件写入其他机器可见的位置，或者只是将文件保存到自己的文件系统中？

map函数总是在本地磁盘产生结果，所以中间结果文件只会在本地文件系统中。

• 即使一次只做一个map任务，但是如果执行了多次map任务后，如果机器突然崩溃，那么会丢失之前负责的所有map任务所产生的中间结果文件，对吗？

中间结果文件放在文件系统中。所以当机器恢复时，中间结果文件还在那里，因为文件数据是被持久化保存的，而不是只会存在于内存中（换句话说，这里依赖了操作系统的文件系统本身的容错性）。并且map或reduce会直接访问包含intermediate results中间结果的机器。

参考链接

链接一
链接二
链接三