《MapReduce: Simplified Data Processing on Large Clusters》论文翻译（MapReduce-OSDI04）

在成功执行完毕后，MapReduce的输出可在通过R个输出文件访问（每个reduce任务一个文件，文件名由用户指定）。通常情况下，用户不需要将这R个输出文件合并到一个文件中，用户经常将这些文件作为另一次MapReduce调用的输入，或者在另一个能够从多个分区的文件输入的分布式程序中使用这些文件。

3.2 master数据结构

master中保存着多种数据类型。对每个map和reduce任务，master会存储其状态（状态包括等待中（idle）、执行中（in-progress）和完成（conpleted））和非等待中的任务对应的worker的标识符。

master是将中间文件区域的位置从map任务传递到reduce任务的管道。因此，对于每个已完成的map任务，master会存储其输出的R个中间文件区域的位置。当map任务完成后，master会收到其对中间文件区域位置和大小信息的更新。这些信息会被增量地推送到有执行中的reduce任务的worker中。

3.3 容错

因为MapReduce库是为使用成百上千台机器处理大规模数据提供帮助而设计的，所以必须能够优雅地对机器故障进行容错。

3.3.1 worker故障

master会定期ping每个worker。如果在一定时间内没有收到worker的响应，master会将该worker标记为故障。被故障的worker处理的已完成的map任务会被重设为其初始的“等待中”的状态，因此其符合被调度到其他worker的条件。同样，在故障的worker上任何执行中的map或reduce任务也会被重设为“等待中”的状态，符合重新调度的条件。

当worker故障发生时，该worker完成的map任务也需要被重新执行，因为map任务的输出被存储在故障的机器的本地磁盘上，无法被访问。故障worker完成的reduce任务则不需要被重新执行，因为他们的输出被存储在全局的文件系统中

当一个起初被worker A执行的map任务因A发生故障而随后被worker B执行时，所有正在执行reduce任务的worker会被告知这个map任务被重新执行。任何没从worker A中读取完数据的reduce任务将会从worker B中读取数据。

MapReduce可以弹性处理大规模worker故障。例如，在MapReduce操作中，由于在正在运行的集群中的网络维护工作导致了80台机器在几分钟内同时变得不可访问。MapReduce的master会简单地重新执行不可访问的worker的机器上已完成的工作，并继续执行后续任务，最终完成整个MapReduce操作。

3.3.2 master故障

我们让master简单地周期性地为之前提到的master中的数据结构设置检查点。如果master 任务挂掉，一份新的master的拷贝会从最后一次检查点的状态重启。尽管只有一个master，发生故障的可能性也很小。因此，目前我们的实现方式为：如果master故障，则终止MapReduce计算。client可以检测到该状态，如果有需要可以重试MapReduce操作。

3.3.3 故障出现时的语义

对于相同的输入数据，当用户提供的map和reduce操作是确定性函数时（译注：确定性函数指在任何时候，当函数输入相同时，总会得到相同的输出。），分布式的MapReduce输出的数据和一个没发生故障的顺序执行的程序输出的数据是一样的。

我们通过原子性地提交map任务和reduce任务输出的方式来实现这一性质。每个执行中的任务将其输出写入到私有的临时文件中。每个reduce任务会创建一个这样的临时文件，每个map任务会创建R个这样的临时文件（每有一个reduce任务就创建一个）。当有一个map任务完成时，该worker会向master发送一条带有R个临时文件名的消息。如果master收到了一个已经完成过的map任务的完成消息，master会忽略该消息。否则，master会在其数据结构中记录这R个文件的文件名。

当有一个reduce任务完成时，该worker会自动地将其临时输出文件重命名为一个永久的文件名。如果同一个reduce任务被在多台机器中执行，会出现多个重命名调用将文件重命名同一个永久文件名的情况。我们依赖下层文件系统提供了原子性重命名操作，来保证最终的文件系统中仅包含来自一次reduce任务输出的数据。

我们绝大多数map和reduce操作是确定性的。因此，分布式的MapReduce语义等同于顺序执行的语义。这使得编程人员可以很容易地理解程序行为。当map和（或）reduce为非确定性函数时，我们提供了较弱但仍合理的语义。当非确定性的操作出现时，一次特定的reduce任务的输出R1​等同于这个非确定性操作顺序执行的输出R1​。但是，不同次reduce任务的输出R2​可能对应这个非确定性操作顺序不同次执行的输出R2​。

考虑这样一种情况，有map任务M和reduce任务R1​和R2​。e(Ri​)表示被提交的任务Ri​的执行过程（有且仅有一个该执行过程）。因为e(R1​)与e(R2​)可能读取了任务M的不同次执行后的输出文件，因此会出现较弱的语义。（译注：即如果M因故障等原因被多次执行，因为M多次执行的输出不一致，所以R1​和R2​读取的输入可能不一致。）

3.4 位置分配

在我们的计算环境中，网络带宽是相对稀缺的资源。为了节约网络带宽，我们将输入数据（由GFS管理^[8]）存储在集群中机器的本地磁盘中。GFS将每个文件分割为若干个64MB的块，并为每个块存储在不同机器上若干个副本（通常为3个）。MapReduce的master会考虑输入文件的位置信息，并试图在持有输入文件的副本的机器上分配相应的map任务。如果分配失败，master会试图将map任务分配在离其输入文件的副本较近的机器上（例如，在与持有输入数据副本的机器在相同交换机下的机器上分配）。在集群中较大比例的机器上运行大型MapReduce操作时，大部分输入数据都是从本地读取，不消耗网络带宽。

3.5 任务粒度

如前文所述，我们将map阶段进一步划分为M份，将reduce阶段进一步划分为R份。在理想状态下，M和R应远大于worker的机器数。让每个worker执行多个不同的任务可以提高动态负载均衡能力，也可以在一个worker故障时提高恢复速度：该worker完成的多个map任务可以被分散到所有其他的worker机器上执行（译注：否则，考虑M小于worker机器数的情况，每个worker上只有一个任务，如果一个worker故障，那么该worker中完成的任务只能在另一台worker机器上重跑，无法充分利用并行的性能）。

在我们的MapReduce实现的实际情况中，对M和R的上限进行了限制。如前文所述，master必须做出O(M+R)个调度决策，并在内存中保存O(M×R)个状态。（内存占用的常数因子比较小：O(M×R)条状态由大约每个map/reduce任务仅一字节的数据组成。）

此外，R还经常受用户限制，因为每个reduce任务会生成一个单独的输出文件。在实际情况下，我们更倾向于自定义参数M，这样可以使每个单独的任务的输入数据大概在16MB到64MB（这样可以使前面提到的局部性优化最有效），同时，我们使R是期望使用的worker机器的较小的倍数。我们经常在2,000台机器上选择M=200,000、R=5,000的参数执行MapReduce计算。

3.6 任务副本

延长MapReduce操作总时间的常见原因之一为“离群问题”：一个机器花费了不寻常的长时间完成计算中最后的几个map任务或reduce任务。吃线离群问题的原因有很多。例如，一台磁盘情况不良的机器可能频繁修正磁盘错误，导致其读取速度从30MB/s降低到1MB/s。集群的调度系统可能已经将其他任务调度到了该机器上，导致其因CPU、内存、本地磁盘或网络带宽等因素执行MapReduce代码更慢。我们最近遇到的问题是在机器初始化代码中的一个bug，其导致了处理器缓存被禁用，受影响的机器上的计算慢了超过100倍。

我们有一个通用的机制来避免离群问题。当MapReduce操作将要完成时，master会通过调度对仍在执行中的任务创建副本并执行。当原任务和其副本之一执行完成时，该任务会被标记为已完成。我们对这个机制进行了一些调优，使它通常情况下对计算资源的占用仅提高几个百分点。我们发现这个机制显著地减少了完成大型MapReduce操作的时间。例如，章节5.3中的排序程序在禁用任务副本机制时，完成时间延长了44%。

4. 改进

尽管编写map和reduce函数提供的基本功能已经能够满足大多数场景下的需求，有一些扩展功能还是会提供很大帮助。我们将在本节中讨论这些扩展。

4.1 分区函数

MapReduce的用户可以自定义其需要的reduce的任务或输出文件数（R）。分区函数（partitioning function）通过中间键值对的键将数据为每个任务分区。我们提供了使用哈希函数（例如，hash(key)modR）的默认分区函数。使用该函数往往会得到比较平衡的分区。然而，在有些情况下，通过某些其他的函数按照键分区很有用。例如，有时输出的键为URL，我们希望所有来自同一个主机的条目最终会被输出到相同的文件中。为了支持类似情况，MapReduce库的用户可以提供一个自定义的分区函数。例如，使用”hash(hostname(urlkey))modR“作为分区函数可以使来自同一个主机的所有URL最终输出到同一个文件中。

4.2 有序性保证

我们保证在一个给定的分区中，中间键值对是按照键的升序进行处理的。这种有序性保证使每个分区生成有序的输出变得非常简单。这对于输出文件格式需要支持按照键进行高效的随机访问等情况时十分有用。

4.3 合并函数

在一些情况下，map任务会产出很多键相同的中间键值对，且用户定义的reduce函数服从交换律和结合律。章节2.1中的单词计数就是一个很好的例子。因为词频往往服从Zipf分布（齐夫定律），每个map任务会产出成百上千条<the,1>的记录。所有的这些计数记录会被通过网络发送到同一个reduce任务，并随后被reduce函数加在一起得到一个总数。我们允许用户自定义一个可选的合并函数（combiner function），在数据通过网络发送前对这部分数据进行合并。

合并函数会在每个执行map任务的机器上执行。通常，实现合并函数和reduce函数的代码是相同的。合并函数和reduce函数唯一的区别是MapReduce库处理函数输出的方式。reduce函数的输出会被写入最终输出文件。合并函数的输出会被写入中间文件，随后中间文件会被发送给reduce任务。

部分数据的合并显著地提高了某些类型的MapReduce操作的速度。附录A包含了一个使用了合并函数的例子。

4.4 输入输出类型

MapReduce库提供了以多种格式读取输入数据的支持。例如，“text”模式将每一行作为一个键值对输入：其中键为行号，值为行的内容。另一种支持的常见的格式被存储为按键排序的键值对序列。每个输入类型的实现都知道如何将输入划分为有意义的区间，以便分开交给map任务处理（例如，“text”模式的区间划分保证仅在行分隔符处划分区间）。用户可以通过提供类似reader接口的实现的方式来增加对新的输入类型的支持，虽然大部分用户仅使用了预支持输入类型中的一小部分。

reader并非必须从文件读取数据。例如，我们可以很容易地实现一个从数据库或从内存中映射的数据结构中读取记录的reader。

类似地，我们也提供了一系列可以将数据输出位不同格式的输出类型，且也可以很容易地通过哦用户代码添加新的输出类型支持。

4.5 附属输出

在一些情况下，MapReduce的用户可以很方便地通过map和（或）reduce操作生成附属输出文件作为额外的输出。我们依赖应用程序的writer来使这种操作具有原子性（atomic）与幂等性（idempotent）。通常，应用程序将数据写入到一个临时文件，并在该文件完全生成完成后原子性地将该文件重命名。

我们没有对一个任务生产多个输出文件提供原子性的两段提交协议（two-phase commits，2PC）支持。因此，产生多个输出文件且有跨文件一致性需求的任务应该具有“确定性（译注，如章节3.3.3）”。但在实际环境中，这一限制并不是什么问题。

4.6 跳过损坏的记录

有时，用户代码中的bug会导致map或reduce函数在处理某些记录时会发成确定性地崩溃。这种bug导致MapReduce操作无法完成。这种情况下，通常的处理方式是修复这个bug，但有时这样并不可行，因为bug可能在无法访问源码的第三方库中。同时，有些时候忽略一些记录时可以接受的，例如在大规模数据集中进行统计分析时。为此，我们提供了一种可选的执行模式，该模式下MapReduce库可以检测会导致确定性崩溃的记录并跳过这些记录，以让处理进程能够继续执行。

每个worker进程会安装一个捕捉段违规（segmentation violation）和总线错误（bus error）的处理器。再调用用户的map或reduce操作之前，MapReduce库会在全局变量中存储参数的编号。如果用户代码产生了一个信号，信号处理器会向master发送一个含有该编号的“last gasp（奄奄一息）”UDP包。当master在同一条记录上收到超过一个故障时，master会在下一次重新执行相关map任务或reduce任务时指示跳过该记录。

4.7 本地执行

调试map或reduce函数中的bug是非常棘手的，因为它们实际运行在分布式系统中，且其经常运行在几千台机器上并由master动态地决定任务的分配。为了帮助开发者debug、分析和小规模测试，我们开发了一个MapReduce库的替代实现，其可以在一台本地机器上顺序的执行所有MapReduce操作。用户可以仅执行计算任务中的几个特定的map任务。用户仅需使用一个特殊的标识符调用程序，就可以轻松地使用任何调试工具或测试工具（如gdb）。

4.8 状态信息

master会运行一个内部的HTTP服务器，并将一系列的状态信息页面暴露给用户。这些页面会展示计算的进度，如多少个任务已经完成、多少个任务正在执行、输入的字节数、中间数据的字节数、输出的字节数、处理速度等。这些页面中还包含展示每个任务输出的标准错误和标准输出文件的页面链接。用户可通过这些数据预测计算需要消耗多长时间、需不需要为计算任务增加额外的资源。这些页面也可以用来发现计算是否比预期慢很多。

另外，顶级的状态页面展示了哪些worker执行失败了与它们失败时运行的map任务和reduce任务。这些信息对诊断用户代码中的bug十分有帮助。

4.9 计数器

MapReduce库提供了用来计数不同事件发生次数的计数器。例如，用户代码可能希望对处理的词数或者索引的德语文档计数等。

若使用计数器，用户代码需要创建一个命名的计数器对象并在map和（或）reduce函数中适当地增加计数器计数。例如：

Counter* uppercase;
uppercase = GetCounter("uppercase");

map(String name, String contents):
  for each word w in contents:
    if (IsCapitalized(w)):
      uppercase->Increment();
    EmitIntermediate(w, "1");

每个worker机器会周期性地将计数器的值传给master（通过ping的响应报文携带）。master将成功执行的map和reduce任务中的计数器的值加在一起，并在MapReduce操作完成时将其返回给用户代码。计数器当前的值同样在master的分析页面中显示，这样用户就可以查看实时的计算进度。master在对计数器求和时，会对多次执行的相同的map或reduce任务中的计数器值去重，以避免重复计数。（任务副本和因故障被重新执行的任务都会导致重复执行。）

有些计数器的值被MapReduce库自动维护，如处理过的输入键值对的数量或生成的输出键值对的数量。

对用户而言，计数器对检查MapReduce操作是否完成非常有帮助。例如，在有些MapReduce操作中，用户代码可能希望保证输出键值对的数量和输入键值对的数量正好相等，或者想保证处理过的德语文档在总数中占的比例是否在允许的范围内。

5. 性能

在本章中，我们将测量大规模集群中的两个MapReduce计算的性能。其中一个计算任务是在大约1TB的数据中按照一个模板（pattern）搜索。另一个计算任务时排序大约1TB的数据。

这两个程序都代表了用户编写的真实程序中占比很大的两类子集。其中一类程序是将数据从一种表示变换到另一种表示，另一类程序是从大规模数据集中提取少量感兴趣的数据。

5.1 集群配置

所有的程序都在一个由大约1800台机器的集群中执行。每台机器有两个开启了超线程的$GHz的Intel至强Xeon处理器、4GB内存、2个160GB的IDE硬盘和1Gbps的以太网连接。这些机器组成了双层树状的交换机网络，根节点总带宽约100~200Gbps。所有机器都在同一个中心托管，因此任何两个机器间往返时延（RTT）小于1ms。

在4GB内存中，有大约1~1.5GB内存被集群为了运行其他任务保留。这些程序是在一个周末的下午执行的，那时CPU、磁盘和网络几乎都处于空闲状态。

5.2 grep

grep程序会扫描1010条100B的记录，以搜索匹配一个相对较少的三个字母的模板（92,337条记录命中）。输入数据被分割为约64MB的分片（M=15000），所有的输出被放置在一个文件中（R=1）。

图2展示了计算进度随时间的变化。Y轴展示了输入数据被扫描的速率。随着分配给MapReduce计算的机器越来越多，其速度也逐渐提高。当有1764个worker被分配到该任务时，速率峰值超过了30GB/s。当map任务完成时，速率开始逐渐下降并在整个计算时间的大概第80s时下降到0。整个计算从开始到结束大概消耗了150s。这包括了大概一分钟的启动时间开销。这一开销的原因是程序需要传播到所有worker机器与打开1000个输入文件并获取局部优化所需的信息时与GFS交互的时延。