MapReduce

摘要

MapReduce既是一种编程模型，也是一种与之关联的、用于处理和产生大数据集的实现。用户要特化一个map程序去处理key/value对，并产生中间key/value对的集合，以及一个reduce程序去合并有着相同key的所有中间key/value对。许多实际的任务都可以用这种模型来表示

用这种函数式风格写出的程序自动就拥有了在一个大的PC机集群上并行执行的能力。运行时系统会负责细节：切分输入数据，在一组机器上调度执行程序，处理机器错误，以及管理所需的机器间通信。这允许不具备任何并行和分布式系统经验的程序员也能轻松地利用一个大型分布式系统的资源

编程模型

计算过程就是输入一组key/value对，再生成输出一组key/value对MapReduce库的使用者用两个函数来表示这个过程：map和reduce

map由使用者编写，使用一个输入key/value对，生成一组中间key/value对。MapReduce库将有着相同中间key I的中间value都组合在一起，再传给reduce函数

reduce也由使用者编写，它接受一个中间key和一组对应的value。它将这些value合并为一个可能更小的value集合。通常每个reduce调用只产生0或1个输出value。中间value是通过一个迭代器提供给reduce函数的。这允许我们操作那些因为大到找不到连续存放的内存而使用链表的value集合

实现

执行过程

图1展示了MapReduce操作的整体流程。当用户程序调用MapReduce函数时，会发生下面一系列动作（图1中的标号与下面列表顺序相同）：

1. 用户程序中的MapReduce库首先将输入文件切分为M块，每块的大小从16MB到64MB（用户可通过一个可选参数控制此大小）。然后MapReduce库会在一个集群的若干台机器上启动程序的多个副本
2. 程序的各个副本中有一个是特殊的——主节点，其它的则是工作节点。主节点将M个map任务和R个reduce任务分配给空闲的工作节点，每个节点一项任务
3. 被分配map任务的工作节点读取对应的输入区块内容。它从输入数据中解析出key/value对，然后将每个对传递给用户定义的map函数。由map函数产生的中间key/value对都缓存在内存中
4. 缓存的数据对会被周期性的由划分函数分成R块，并写入本地磁盘中。这些缓存对在本地磁盘中的位置会被传回给主节点，主节点负责将这些位置再传给reduce工作节点
5. 当一个reduce工作节点得到了主节点的这些位置通知后，它使用RPC调用去读map工作节点的本地磁盘中的缓存数据。当reduce工作节点读取完了所有的中间数据，它会将这些数据按中间key排序，这样相同key的数据就被排列在一起了。同一个reduce任务经常会分到有着不同key的数据，因此这个排序很有必要。如果中间数据数量过多，不能全部载入内存，则会使用外部排序
6. reduce工作节点遍历排序好的中间数据，并将遇到的每个中间key和与它关联的一组中间value传递给用户的reduce函数。reduce函数的输出会写到由reduce划分过程划分出来的最终输出文件的末尾
7. 当所有的map和reduce任务都完成后，主节点唤醒用户程序。此时，用户程序中的MapReduce调用返回到用户代码中

成功完成后，MapReduce执行的输出都在R个输出文件中（每个reduce任务产生一个，文件名由用户指定）。通常用户不需要合并这R个输出文件——他们经常会把这些文件当作另一个MapReduce调用的输入，或是用于另一个可以处理分成多个文件输入的分布式应用

主节点数据结构

主节点维持多种数据结构。它会存储每个map和reduce任务的状态（空闲、处理中、完成），和每台工作机器的ID（对应非空闲的任务）

主节点是将map任务产生的中间文件的位置传递给reduce任务的通道。因此，主节点要存储每个已完成的map任务产生的R个中间文件的位置和大小。位置和大小信息的更新情况会在map任务完成时接收到。这些信息会被逐步发送到正在处理中的reduce任务节点处。

容错性

1.工作节点错误

主节点周期性的ping每个工作节点。如果工作节点在一定时间内没有回应，主节点就将它标记为已失败。这个工作节点完成的任何map任务都被重置为空闲状态，并可被调度到其它工作节点上。同样地，失败的工作节点上正在处理的任何map或reduce任务也被重置为空闲状态，允许被调度

失败节点上已完成的map任务需要重执行的原因是它们的输出存储在失败机器的本地磁盘上，因此无法访问到了。已完成的reduce任务不需要重执行，因为它们的输出存储在了一个全球文件系统上

当一个map任务先被A节点执行过，随后又被B节点重执行（A节点已失败），所有执行reduce任务的工作节点都能收到重执行的通知。任何没有读取完A节点数据的reduce任务都会从B节点读取数据

2.主节点错误

一种简单的方法是令主节点定期将上面描述的数据结构保存为恢复点。如果主节点任务失败，就可以从上一个恢复点状态启动一个新的程序副本。但是给定的条件是只有一个主节点，它也不太可能失败；因此我们当前的实现会在主节点失败时中止MapReduce计算 。客户可以检查到这一情况，并在他们需要时重启MapReduce操作

3.出现故障时的语义

当用户提供的map和reduce操作对于它们输入的值都是确定性的，我们的分布式实现产生的输出值就如同将整个程序分成一个不间断的串行执行过程一样。

为了实现这个性质，我们依赖于map和reduce任务输出结果的提交是原子的。每个处理中的任务都会将它的输出写入私有的临时文件中。一个reduce任务产生一个这样的文件，而一个map任务则产生R个这样的文件（每个reduce任务一个）。当map任务完成时，工作节点发送给主节点的消息中带有R个临时文件的名字。如果主节点收到了一个来自已完成节点的完成消息，它就会忽略这个消息。否则，主节点会将R个文件的名字记录在相应的数据结构中。

当reduce任务完成时，工作节点会执行原子性的更名操作，将临时输出文件更名为最终输出文件。如果相同的reduce任务在多个机器上执行，就会有多个更名调用应用在相同的最终输出文件上。我们依赖于由底层文件系统提供的原子更名操作，才能保证最终的文件系统中只包含由其中一个reduce执行产生的数据。

我们的绝大多数map和reduce操作都是确定性的，这种情况下我们的语义和一个串行执行过程是等同的，这也使程序员很容易推出他们程序的行为。当map和reduce操作有不确定性时，我们提供较弱但仍然合理的语义。当存在不确定的操作时，某个reduce任务R1的输出等价于一个不确定程序的串行执行输出。但某个reduce任务R2的输出可能符合这个不确定程序的另一个串行执行输出。

考虑map任务M和reduce任务R1、R2。令e(Ri)为Ri的已提交的执行结果（只执行一次）。此时弱语义生效，因为e(R1)可能读取了M的一次输出，而e(R2)则可能读取了M的另一次输出。

局部性

在计算环境中，网络带宽是一种比较稀缺的资源。我们利用下面的事实来节省带宽：输入数据（由GFS管理）就存储在组成集群的机器的本地磁盘上。GFS将每个文件分成64MB大小的区块，每块复制若干份（通常为3份）存储到不同的机器上。MapReduce主节点会把输入文件的位置信息考虑进去，并尝试将map任务分配到保存有相应输入数据的机器上。如果失败的话，它会试图将map任务调度到临近这些数据的机器上（同一网关的工作节点）。这样，大多数输入数据都是本地读取，并不消耗网络带宽。

任务粒度

将map阶段分成M份，将reduce阶段分成R份。理想情况下，M和R应该比工作节点机器的数量大很多。每个工作节点处理很多不同的任务，可以增强动态负责均衡能力，也能加速有工作节点失败时的恢复情况：失败节点已经完成的map任务有很多的时候也能传递给其它所有工作节点来完成。

R通常由用户指定，因为每个reduce任务都会产生一个独立的输出文件。倾向于这样选择M：即可以将每个单独的任务分成16-64MB大的输入数据（此时上面所说的局部性优化效果最好），同时令R为待使用的工作节点数量较小的整数倍。