MapReduce：大型集群上的简单数据处理
摘要
MapReduce是一个编程模型和一个处理和生成大数据集的相关实现。用户指定一个map函数处理一个key-value对来生成一组中间key-value对；指定一个reduce函数合并所有和同一中间key值相联系的中间value值。许多现实世界中的任务以这个模型展现，就像文中展示的那样。
以这种函数类型编写的程序在一群日常机器上自动并行化并执行。运行时系统关心划分输入数据的细节，在一组机器间调度程序的执行，处理机器失效，管理内部机器需要的通信。这使得那些没有任何并行和分布式系统经验的程序员可以容易地使用大型分布式系统的资源。
我们MapReduce的实现运行在一大群日常机器上并且高度可扩张：一个典型的MapReduce计算在成千上万台机器上处理数TB的数据。程序员发现系统很易用：成百上千的MapReduce程序已经被实现，每天都有多余1000个MapReduce作业在Google集群上执行。
1．介绍
在过去的5年里，作者以及Google里的其他程序员已经实现了数以百计的，特殊目的的计算。这些计算处理海量原始数据，比如，文档抓取（shijin：类似网络爬虫的程序）、web请求日志等；或者计算各种各样的派生数据，比如倒排索引、web文档的图结构的各种表示形势、每台主机上网络爬虫抓取的页面数量的汇总、列举一天中一组最频繁的查询等。大多数这种计算概念上直截了当。然而输入的数据量巨大，并且为了在合理的时间内完成，计算不得不分布到数百或数千台机器上。如何并行计算、分发数据、处理失效的问题凑在一起使原本简单的计算晦涩难懂，需要大量复杂的代码来处理这些问题。
作为对上述复杂性的应对，我们设计一个新的抽象模型，其使我们可以表达我们试图执行的简单运算，但是将并行、容错、数据分布和负载均衡等散乱的细节隐藏在了一个库里面。我们抽象模型的灵感来自Lisp和许多其他函数式语言的map和reduce原语。我们意识到大多数我们的计算都涉及这样的操作：在我们输入中的每个逻辑“记录”上应用map操作，以便计算出一组中间key/value对，然后在所有享有相同key值的value值应用reduce操作，以便合适地合并派生的数据。我们使用带有用户指定map和reduce操作的函数模型，就可以轻易地并行化大规模计算；并且可以使用“再次执行”（re-execution）作为基础的容错机制。
这项工作的主要贡献是一个简单强大的接口，该接口使自动地并行化和分布大规模计算成为了可能，接口连同该接口的实现，实现了在大群日常PC机上的高性能。
第二节描述基本的编程模型给出一些例子。第三节描述了为我们基于集群的计算环境定做的MapReduce接口的实现。第四节描述一些我们发现有用的编程模型优化。第五节对我们各种不同任务实现的性能进行了测量。第六节探索了MapReduce在Google内部的使用，包含我们在使用其作为我们生产索引系统的重写操作的基础时的经验。第七节讨论相关的和未来的工作。
2．编程模型
计算取出一组输入key/value对，产生一组输出key/value对。使用MapReduce库的用户用两个函数表达这个计算：Map和Reduce。
用户自己编写的Map接受一个输入对，然后产生一组中间key/value对。MapReduce库把所有和相同中间key值I关联的中间value值聚集在一起后传递给Reduce函数。
也是由用户编写的Reduce函数接受一个中间key值I和一组那个key值的value值。Reduce函数将这些value值合并在一起，形成一个可能更小的value值的集合。通常每次Reduce调用仅产生0或1个输出value值。中间value值通过一个迭代器提供给用户的Reduce函数，这样我们就可以处理因太大而无法适应内存的value值列表。
2.1 例子
考虑这么一个问题，在一个大的文档集合中对每个单词出现的次数进行计数，用户可能要类似下面伪代码的代码：
map(String key, String value):
// key: document name
// value: documen t contents
for each word w in value:
EmitIntermediate(w, “1”);
reduce(String key, Iterator values):
// key: a word
// values: a list of counts
int result = 0;
for each v in value s:
result += Parse Int(v);
Emit(AsString(result));
Map函数emit每个词加上一个相关的出现计数(在这个简单的例子里就是1)。Reduce函数把为一个特定的词emit的所有计数加起来。
另外，用户编写代码用输入和输出文件的名字以及可选的调节参数填充一个mapreduce说明对象，用户之后调用MapReduce函数，并把这个说明对象传递给它。用户的代码和MapReduce库链接在一起(用C++实现)。附录A包含了这个例子的全部程序文本。
2.2 类型
尽管在前面的伪代码按照字符串输入输出书写，但是在概念上，用户提供的map和reduce函数有相关的类型：
map (k1,v1) ->list(k2,v2)
reduce (k2,list(v2)) ->list(v2)
比如，输入的key值和value值与输出的key值和value值从不同的类型域得到。并且，中间key值和value值与输出key值和value值来自同一个类型域。（alex注：原文中这个domain的含义不是很清楚，我参考Hadoop、KFS等实现，map和reduce都使用了泛型，因此，我把domain翻译成类型域）。
我们的C++实现使用字符串类型作为用户自定义函数的输入输出，并将字符串与适当类型的转换工作交给了客户代码。
2.3 更多的例子
这里有一些有趣程序的简单例子，可以很容易地作为MapReduce计算来表示：
分布式的Grep：如果与提供的模式串匹配，Map函数emit一行，Reduce函数是一个恒等函数，即仅仅把提供的中间数据复制到输出。
URL访问频率计数：Map函数处理web页面请求的日志，然后输出(URL,1)。Reduce函数把相同URL的value值加在一起，emit一个(URL,总数)对。
倒转网络链接图：Map函数为每个链接输出(target,source)对，每个链接连接到在名字叫源的页面中发现的目标URL。Reduce函数把与给定目标URL相关的所有源URL连接成列表，emit(target,list(source))对。
每个主机的检索词向量：检索词向量将一个文档或者一组文档中出现的嘴重要的词汇总为一个（词，频）对列表。Map函数为每个输入文档emit(主机名,检索词向量)，其中主机名来自文档的URL。Reduce函数为一个给定主机接收所有每文档检索词向量，其将这些检索词向量加在一起，丢弃低频的检索词，然后emit一个最终的(主机名,检索词向量)对。
倒排索引：Map函数解析每个文档，emit一系列(词,文档号)列表，Reduce函数接受一个给定词的所有(词,文档号)对，排序相关的文档号，emit(词,list（文档号）)。所有的输出对集合形成一个简单的倒排索引，增加计算来跟踪词在文档中的位置很简单。
分布式排序：Map函数从每个记录提取key值，emit(key,record)对。Reduce函数原封不动地emit所有对。这个运算依赖4.1描述的分区设备和4.2节描述的排序属性。
3．实现
MapReduce有多种不同的可能实现。正确的选择取决于环境。例如，一种实现可能适用于小型共享内存的机器，另外一种则适用于大型NUMA多处理器，然而还有的适合大型的网络集群。
本章节描述一个针对Google内部广泛使用的运算环境的实现：大群日常机器用交换以太网连接在一起。在我们环境中：
1. 机器通常是x86双核处理器、运行Linux系统、每台机器2-4GB内存。
2. 使用日常网络硬件，通常在机器级别带宽为百兆每分或者千兆每分，但是平均远小于网络整体带宽的一半。（averaging considerably less in overall bisection bandwidth）
3. 集群包含数百或数千台机器，因此机器失效是常态。
4. 存储由直接附属在个体机器上的廉价IDE硬盘提供。一个内部开发的分布式文件系统用来管理存储在这些磁盘上的数据。文件系统使用副本来提供不可靠的硬件上的可用性和可靠性。
5. 用户向一个调度系统提交作业。每个作业包含一组任务，调度系统将这些任务映射到一组集群内部可用的机器上。
3.1 执行概览
Map调用通过将输入数据自动分为M个片段（a set of M splits）的方式被分布到多台机器上。输入片段能够被不同的机器并行处理。Reduce调用使用分区函数将中间key值空间分成R份（例如，hash(key) mod R），继而被分布到多台机器上执行。分区数量（R）和分区函数由用户指定。
图1展示了我们的实现中MapReduce操作的整体流程。当用户调用MapReduce函数时，下列动作发生（图一中的数字标签对应下面列表中的序号）：
1. 用户程序中的MapReduce库首先将输入文件分成M份，每份通常在16MB到64MB之间(可以通过可选参数由用户控制)。然后在机器集群中启动许多程序副本。
2. 程序副本中有一个是特殊-master。其它的是worker，由master分配工作。有M个map任务和R个reduce任务将被分配，master选择空闲的worker然后为每一个worker分配map任务或reduce任务。
3. 被分配了map任务的worker读取相关输入片段的内容，它从输入的数据片段中解析出key/value对，然后把key/value对传递给用户自定义的Map函数，由Map函数生成的中间key/value对缓存在内存中。
4. 缓存对被定期地写入本地磁盘，被分区函数分成R个域。缓存对在本地磁盘上的位置被传回master，master负责将这些位置转寄给reduce worker。
5. 当一个reduce worker被master告知位置信息后，它使用远程过程调用从map worker的本地磁盘读取缓存数据。当一个reduce worker读取了所有的中间数据后，它通过中间key值对缓冲数据排序，以便相同key值的出现组织在一起。由于通常许多不同的key值映射到同一reduce任务上，因此排序是需要的。如果中间数据量太大而无法适应内存，那么就使用外部排序。
6.Reduce worker迭代排序后的中间数据，对于每一个遇到的唯一的中间key值，Reduce worker将这个key值和与它相关的中间value值的集合传递给用户的Reduce函数。Reduce函数的输出被追加到这个reduce分区的一个最终输出文件。
7. 当所有的map和reduce任务完成之后，master唤醒用户程序。此时此刻，用户程序里的对MapReduce调用返回用户代码。
成功完成之后，mapreduce执行的输出可以在R个输出文件中得到（每个文件对应一个reduce任务，文件名由用户指定）。通常，用户不需要将这R个输出文件合并成一个文件-他们经常把这些文件作为输入传递给另外一个MapReduce调用，或者在另外一个分布式应用中使用它们，这种分布式应用能够处理分成多个文件的输入。
3.2 Master数据结构
Master保持一些数据结构，对每一个map和reduce任务，它保存其状态（空闲、进行中或已完成)，以及Worker机器(对于非空闲任务)的身份（identity）。
Master是一个管道，通过它中间文件域的位置信息从map任务传播到reduce任务。因此，对于每个已经完成的map任务，master存储了map任务产生的R个中间文件域的位置和大小。当map任务完成时，接收到了位置和大小信息的更新，这些信息被递进地推送给那些正在运行的reduce任务。
3.3 容错
因为MapReduce库是设计用来协助使用数百数千的机器处理超大规模数据的，这个库必须优雅地处理机器故障。
worker故障
master周期性地ping每个worker。如果在一个确定的时间段内没有收到worke的回应，master将这个worker标记为失效。任何由这个worker完成的map任务被重置回它们初始的空闲状态，因此变得可以被调度到其它worker。同样，在一个失效的worker上正在运行的map或reduce任务被重置为空闲状态，变得可被重新调度。
故障时已完成的map任务必须重新执行是因为它们的输出被存储在失效机器的本地磁盘上，因此不可访问了。已经完成的reduce任务不需要再次执行，因为它们的输出存储在全局文件系统。
当一个map任务首先被worker A执行，之后被worker B执行（因为A失效），所有执行reduce任务的worker会接到重新执行的通知。还没有从worker A读取数据的任何reduce任务将从worker B读取数据。
MapReduce对规模worker失效很有弹性。例如，在一次MapReduce操作执行期间，在正在运行的集群上进行的网络维护一次造成一组80台机器在几分钟内无法访问，MapReduce master只需简单德再次执行那些不可访问的worker完成的工作，然后继续执行，直终完成这个MapReduce操作。
master失败
让master定期对上面描述的master数据结构作检查点很简单。如果这个master任务失效了，一个新的备份可以从上一个检查点状态启动。然而，考虑到只有一个单独master，master失效是不太可能的，因此我们现在的实现是如果master失效，就中止MapReduce运算。客户可以检查这个情况，并且如果需要可以重试MapReduce操作。
在失效面前的语义
（semantics in the presence of failures）
当用户提供的map和reduce操作是输入值的确定性函数，我们的分布式实现产生相同的输出，就像没有错误、顺序执行整个程序产生的一样。
我们依赖对map和reduce任务的输出是原子提交来完成这个特性。每个工作中的任务把它的输出写到私有的临时文件中。一个reduce任务生成一个这样的文件，并且一个map任务生成R个这样的文件（一个reduce任务对应一个）。当一个map任务完成时，worker向master发送一个消息，消息中包含R个临时文件名。如果master收到一个已经完成的map任务的完成消息，它将忽略这个消息；否则，master将这R个文件的名字记录在一个master数据结构里。
当一个reduce任务完成时，reduce worker原子性地将临时文件重命名为最终的输出文件。如果同一个reduce任务在多台机器上执行，针对同一个最终输出文件将有多个重命名调用执行。我们依赖底层文件系统提供的原子重命名操作来保证最终的文件系统状态仅仅包含reduce任务一次执行产生的数据。
绝大多数map和人reduce操作是确定的，而且存在这样的一个事实：我们的语义等同于顺序的执行，在这种情况下，程序员可以很容易推断他们程序的行为。当map或/和reduce操作是不确定性的时候，我们提供较弱但是依然合理的语义。在非确定性操作面前，一个特定reduce任务R1的输出等价于一个非确定性程序顺序执行产生的R1的输出。然而，一个不同reduce任务R2的输出可能相当于一个不同的非确定行程序顺序执行产生的R2的输出。
考虑map任务M和reduce任务R1、R2。设e(Ri)是Ri提交的执行过程（只有一个这样的执行过程）。由于e(R1)可能读取了由M一次执行产生的输出，而e(R2)可能读取了由M的不同执行产生的输出，较弱的语义随之而来。
3.4 存储位置
在我们的计算环境中，网络带宽是一个相当稀缺的资源。我们通过充分利用输入数据(由GFS管理)存储组成集群的机器的本地磁盘上这样一个事实来节省网络带宽。GFS把每个文件分成64MB的块，并且在不同机器上存储每个块的一些拷贝(通常是3个拷贝)。考虑到输入文件的位置信息，MapReduce的master试图将一个map任务调度到包含相关输入数据拷贝的机器上；尝试失败的话，它将尝试调度map任务到靠近任务输入数据副本的机器上(例如，一个包含数据并在同一网关上的worker机器)。当在一个集群的大部分worker上运行大型MapReduce操作的时候，大部分输入数据从本地机器读取，并且不消耗带宽。
3.5 任务粒度
如上所述，我们把map阶段细分成M个片段、把reduce阶段细分成R个片段。理想情况下，M和R应当比worker机器的数量要多得多。在每台worker上执行许多不同的任务提高动态负载平衡，并且在worker故障时加速恢复：失效机器上完成的很多map任务可以分布到所有其他的worker机器。
但是在我们的实现中对M和R的取值有实际的限制，因为master必须制定O(M+R)调度策略，并且如上所述在内存中保存O(M*R)个状态（然而内存使用的常数因子还是比较小的：O(M*R)片状态大约包含每对map任务/reduce任务对1个字节）。
此外，R值通常是被用户制约的，因为每个reduce任务的输出最终在一个独立的输出文件中。实际上，我们倾向于选择M值以使每个独立任务有大约16M到64M的输入数据（这样如上所述的本地最优化最有效），我们把R值设置为我们想使用的worker机器数量的一个小的倍数。我们通常用，使用2000台worker机器，以M=200000，R=5000来执行MapReduce计算。
3.6 备用任务
延长一个MapReduce操作花费的总时间的常见因素之一是“落伍者”：一台机器花费了不同寻常的长时间才完成计算中最后几个map或reduce任务之一，出现“落伍者”的原因非常多。比如：一个有坏磁盘的机器可能经历频繁的纠错以致将其读性能从30M/s降低到1M/s。集群调度系统可能已经降其他任务调度到这台机器上，由于CPU、内存、本地磁盘和网络带宽的竞争导致执行MapReduce代码更慢。我们最近遇到的一个问题是机器初始化代码中的bug，导致处理器缓存失效：受影响机器上的计算减慢了超过百倍。
我们有一个通用的机制来减轻“落伍者”的问题。当一个MapReduce操作接近完成的时候，master调度剩余正在运行任务的备份执行、无论主执行还是备份执行完成，任务被标记为已完成。我们调优了这个机制，以便其通常增加操作不多于几个百分点的计算资源。作为示例，5.3节描述的排序程序在关掉备用任务机制时要多花44%的时间完成。
4．精细化
尽管简单地书写Map和Reduce函数提供的基本功能能够满足大多数需求，我们还是发现了一些有用的扩展。在本节做了描述。
4.1 分区函数
MapReduce的使用者指定他们需要的reduce任务/输出文件的数量（R）。我们在中间key值上使用分区函数将数据在这些任务间分开。一个默认的分区函数是使用哈希(比如，hash(key) mod R)提供的。它倾向于导致相当均衡的分区。然而，在某些情况下，通过key值的其它函数分割数据是有用的。比如，输出的key值是URLs，我们希望单个主机的所有条目以结束在同一个输出文件中。为了支持类似的情况，MapReduce库的用户可以提供一个专门的分区函数。例如，使用“hash(Hostname(urlkey)) mod R”作为分区函数就可以把所有来自同一个主机的URL结束在同一个输出文件中。
4.2 顺序保证
我们保证在给定的分区中，中间key/value对是以key值递增的顺序处理的。这个顺序保证每个分成生成一个有序的输出文件很容易，这在下列情况时很有用：输出文件的格式需要支持按key值高效地随机访问查找，或者输出的用户有序的数据很方便。
4.3 合并函数
某些情况下，每个map任务产生的中间key值有显著的重复，并且用户指定的Reduce函数满足结合律和交换律。这个情况很好的例子是2.1节中的词数统计示例。由于词频倾向于满足zipf分布，每个map任务将产生数百数千形如<the,1>的记录。所有这些记录将通过网络被发送到一个单独的reduce任务，然后被Reduce函数累加起来产生一个数。我们允许用户指定一个可选的合并函数，其在通过网络发送数据之前对数据进行部分合并。
合并函数在每台执行map任务的机器上执行。通常使用相同的代码实现合并函数和reduce函数。合并函数和reduce函数唯一的区别就是MapReduce库如何处理函数的输出。Reduce函数的输出被写入在最终的输出文件，合并函数的输出被写到中间文件里，该文件被发送给reduce任务。
部分合并显著加速了MapReduce操作中的某些类。附录A包含一个使用合并函数的例子。
4.4 输入输出类型
MapReduce库支持读取一些不同的格式的输入数据。比如，文本模式输入将每一行视为一个key/value对。key是文件中的偏移，value是那一行的内容。另外一种通常支持的格式以key值排序存储了一系列key/value对。每种输入类型的实现都懂得如何把自己分割成有意义的范围，以便作为独立的map处理(例如，文本模式的范围分割确保分割只在行边界发生)。虽然大多数用户仅仅使用少量预定义输入类型之一，但是用户可以通过提供一个简单的Reader接口的实现支持一个新的输入类型。
需要提供数据的reader不必从文件中读取，比如，我们可以容易地定义一个从数据库里读记录的reader，或者从映射在内存中的数据结构读。
类似的，我们为生产不同格式的数据提供一组输出类型，用户代码可以容易地为新的输出类型添加支持。
4.5 副作用
某些情况下，MapReduce的使用者发现从map和/或reduce操作中产生作为附加输出的辅助文件比较方便。我们依赖程序writer把这种“副作用”变成原子的和幂等的（alex注：幂等的指一个总是产生相同结果的数学运算）。通常应用程序首先写到一个临时文件，在全部生成之后，原子地将这个文件重新命名。
我们不为单个任务产生的多个输出文件的原子两步提交提供支持。因此，产生多个输出文件、并且具有跨文件一致性要求的任务，必须是确定性的。实际这个限制还没有成为问题。
4.6 跳过受损记录
有时候，用户代码中的bug导致map或者reduce函数在某些记录上确定性地崩溃。这样的bug阻止MapReduce操作完成。应对的通常过程是修复bug，但是有时不可行；可能这个bug在第三方库里，其源码是得不到的。而且有时忽略一些记录是可以接受的，比如在一个大数据集上进行统计分析。我们提供了一种可选的执行模式，这种模式下，为了保证继续进行，MapReduce库探测哪些记录导致确定性的崩溃，并且跳过这些记录。
每个worker进程都安装了一个捕获段例外和总线错误的信号句柄。在调用用户的Map或Reduce操作之前，MapReduce库在一个全局变量中保存了参数的序号。如果用户代码产生了一个信号，信号句柄向MapReduce的master发送包含序列号的“奄奄一息”的UDP包。当master在特定记录上看到多余一次失败时，当master发布相关Map或者Reduce任务的下次重新执行时，它就指出这条记录应该跳过。
4.7 本地执行
调试Map和Reduce函数的是非常棘手的，因为实际的执行发生在分布式系统中，通常是在数千台机器上，由master动态地制定工作分配策略。为了促进调试、性能剖析和小规模测试，我们开发了一套可选的MapReduce库的实现， MapReduce操作在本地计算机上顺序地执行所有工作。为用户提供了控制以便把计算限制到特定的map任务上。用户通过特殊的标志来调用他们的程序，然后可以容易地使用他们觉得有用的调试和测试工具（比如gdb）。
4.8 状态信息
Master运行着内部的HTTP服务器并且输出一组供人消费的状态页面。状态页面显示包括计算的进展，比如已经完成了多少任务、有多少任务正在进行、输入的字节数、中间数据的字节数、输出的字节数、进行的百分比等等。页面还包含指向每个任务产生的stderr和stdout文件的链接。用户可以使用这些数据预测计算需要执行多长时间、是否需要增加向计算增加更多的资源。这些页面也可以用来搞清楚什么时候计算比预期的要慢。
另外，最顶层的状态页面显示了哪些worker失效了，以及他们失效的时候正在运行的map和reduce任务是哪些。这些信息在尝试诊断用户代码中bug的时候很有用。
4.9 计数器
MapReduce库提供了一个计数器促进统计各种事件发生次数。比如，用户代码可能想统计处理的总词数、或者已经索引的德文文档数等等。
为了使用这个附加功能，用户代码创建一个名叫计数器的对象，然后在在Map和Reduce函数中适当地增加计数。例如：
Counter* uppercase;
uppercase = GetCounter(“uppercase”);
map(String name, String contents):
for each word w in c ontents:
if (IsCapitalized(w)):
uppercase->Increment();
EmitIntermediate(w ,“1″);
这些来自单个worker机器的计数值周期性传播到master（附加在ping的应答包中）。master把来自成功执行的map和reduce任务的计数器值进行累加，当MapReduce操作完成后，返回给用户代码。当前的计数器值也会显示在master的状态页面上，这样人们可以观看当前计算的进度。当累加计数器值的时候，master排除同一map或者reduce任务重复执行的影响，避免重复计数（重复执行额可以起因于我们使用备用任务以及失效后任务的重新执行）。
有些计数器的值由MapReduce库自动维护，比如已经处理的输入key/value对的数量、已经产生的输出key/value对的数量。
用户发现计数器附加功能对MapReduce操作行为的完整性检查有用。比如，在一些MapReduce操作中，用户代码可能需要确保产生的输出对的数量精确的等于处理的输入对的数量，或者处理的German文档的比例在处理文档的整体数量中在可以容忍的比例。
5．性能
本节我们用在一大群机器上运行的两个计算来测量MapReduce的性能。一个计算搜寻大约1TB的数据查找特定的模式匹配，另一个计算排序大约1TB的数据。
这两个程序是由MapReduce用户书写的实际程序子集的代表-一类程序是将从一种表现形式打乱为另外一种表现形式；另一类是从大数据集中抽取少量用户感兴趣的数据。
5.1 集群配置
所有程序都运行在一个大约由1800台机器组成的集群上。每台机器有两个2G主频、支持超线程的Intel Xeon处理器，4GB的内存，两个160GB的IDE硬盘和一个千兆以太网。这些机器被安排在一个两层树形交换网络中，在root节点大约支持100-200GBPS的合计带宽。所有机器使用相同主机设备，因此任意对之间的往返时间小于1毫秒。
4GB的内存中，大约1-1.5G被运行在集群上的其他任务预订。程序在周末下午执行，这时CPU、磁盘和网络大多数空闲。
5.2 查找
这个grep程序从头到尾扫描1010个100字节的记录，查找相当稀少的3个字符的模式（这个模式出现在92337个记录中）。输入数据被分割成大约64M的块（M=15000，shijin：1010*100/(64*1024*1024)），整个输出被放在一个文件中（R=1）。
图2 显示了运算随时间的进展。Y轴表示输入数据的扫描速度。这个速度随着更多的机器被分配到MapReduce计算中而增加，当1764台worker被分配，速度达到超过30GB/s的峰值。当Map任务结束时，速度开始降低并在计算到80秒时达到0。整个计算从开始到结束大约花了150秒。这包括大约一分钟的启动开销。开销起因于程序传播到所有worker机器、与GFS交互打开1000个输入文件集合的延迟、获取文件优化所需信息的时间。
5.3 排序
排序程序对1010个100字节的记录（大约1TB的数据）排序。这个程序模仿TeraSort基准测试程序[10]。
排序程序由不到50行代码组成。一个三行Map函数从一个文本行中提取出10字节的排序key值，并且将key值和原始文本行作为中间key/value对emit。我们使用了一个内置的恒等函数作为Reduce操作。这个函数将中间key/value对作为输出输出key/value对不加修改地传送。最终已排序的输出写入到一组双备份(2-way replicated)的GFS文件（也就是说，作为程序输出，2TB的数据被写入）。
像以前一样，输入数据被分割成64MB的片（M=15000）。我们把已排序的输分到4000个文件（R=4000）。分区函数使用key值的原始字节将其分离到R个片段其中之一。
我们这个基准测试的分区函数具有key值分布的内置knowledge。在一个普通排序程序中，我们会增加一个预处理的MapReduce操作，用于key值的样本并使用key值样本的分布计算最终排序过程的分割点。
图三（a）显示了这个排序程序的正常执行过程。左上的图显示了输入数据的读取速度。速度达到大约13GB/s的峰值，并且自从所有map任务完成后，在200秒消逝前速度下降的相当快。值得注意的是，输入速度小于grep。这是因为排序map任务花了大约一半的时间和I/O带宽把中间输出写到本地硬盘。Grep的相应中间输出大小可以忽略不计。
左边中间的图显示了数据通过网络从map任务发送到reduce任务的速度。这个重排从第一个ma任务完成就开始了。图中的第一个峰值是第一批大约1700个reduce任务（整个MapReduce大约分配了1700台机器，每台机器一次至多执行1个reduce任务）。计算进行到大约300秒后，第一批reduce任务中的其中一些结束，我们开始为剩余的reduce任务重排数据。所有的重排大约在计算进行到600秒时结束。
左下图显示已排序的数据由reduce任务写入最终输出文件的速度。在第一个重排阶段结束和写阶段开始之间有一个延迟，这是因为机器正忙于排序中间数据。写操作以2-4GB/s的速度持续了一段时间。所有的写操作在计算进行到850秒时结束。计入启动的开销，整个运花费了891秒。这和当前已报道的最好结果类似，该记录是TeraSort基准测试[18]的1057秒。
一些要注意的事情：输入速度高于重排速度高于输出速度是因为我们的本地化优化策略-大部分数据从本地硬盘读取，绕考了我们的相关带宽约束网络。重排速度比输出速度高是因为输出过程写了两份已排序数据（我们基于可靠性和可用性的原因生成了输出的两个副本）。我们写了两个副本是因为这是底层文件系统为可靠性和可用性提供的机制。如果底层文件系统使用纠删码编码[14]而不是备份的方式，写数据需要的网络带宽将会减少。
5.4 备用任务的影响
图三（b）显示了关闭备用任务的一个排序程序的执行情况。执行的流程和图3（a）显示的类似，除了在没有任何写活动出现的地方有一个很长的尾巴。960秒后，只有5个reduce任务没有完成。然而这些最后的少量落伍者300秒之后才结束。整个计算花费了1283秒，在消逝的时间中增加了44%（？）。
5.5 机器故障
在图三（c）显示这样一个排序程序的执行，其中我们在计算进行几分钟后故意杀掉了1746个worker进程中的200个。底层集群调度器立刻在这些机器上重启新的worker进程（因为只是进程被杀死了，机器仍在正确运行）。
Worker进程死亡显示为“负”的输入速度，因为之前一些完成的map工作消失了（由于相应的map worker进程被杀掉了）并且需要重做。这个map工作的重新执行相当快。整个计算，包含启动开销，在933秒内完成（只比正常执行时间增加了5%）。
6．经验
我们在2003年1月写了MapReduce库的第一个版本，然后在2003年8月做了显著增强，包括本地优化、任务执行在worker机器之间的动态负载均衡等等。从那以后，我们惊喜于MapReduce库能如此广泛地应用于于我们从事工作中的各种问题。它已经用于Google内部很广的区域范围，包括：
·大规模机器学习问题问题，
·Google News和Froogle产品的集群问题
·用于生产受欢迎查询（比如Google Zeitgeist）的报告的数据提取。
·为新的实验或者产品提取网页属性（例如，为本地搜索从大型网页库中抽取地理位置信息）。
·大规模的图形计算。
图四显示了随着时间推移，我们的基础源码管理系统中经验证的独立的MapReduce程序数量的显著增加。从2003年早期的0个增长截止2004年9月底的几乎900独立的实例。MapReduce如此成功是因为，使用MapReduce库能够在半个小时过程内写出一个能够在上千台机器上高效运行的简单程序，这极大地加快了开发和原形设计的周期。另外，它允许没有分布式和/或并行系统经验的程序员很容易地开发大量资源。
在每个作业结束的时候，MapReduce库对作业使用的计算资源的相关统计数据进行记录。在表1中，我们列出了2004年8月份运行在Google上的MapReduce任务的子集的统计数据。
6.1 大规模索引
目前为止我们对MapReduce最重要的使用之一就是重写了产生Google网络搜索服务所使用的数据结构的生产索引系统。索引系统将通过爬虫系统检索的海量文档作为输入，这些文档存储为一组GFS文件。这些文档的原始内容的大小超过了20TB。索引程序作为一系列五到十个MapReduce操作运行。使用MapReduce（而不是索引系统的优先版本中自组分布式传送）提供了一些好处：
·索引代码更简单、小巧、容易理解，因为处理容错、分布式和并行的代码隐藏在MapReduce库内部。比如，当使用MapReduce表达时，一个计算过程的大小从大约3800行C++代码减少到大约700行代码。
·MapReduce库的性能足够好，因此我们可以把概念上不相关的计算分开，而不是混在一起以期避免数据传送。这使修改索引程序相当简单。比如，在旧的索引系统耗费几个月的一次改变在新系统中只需几天实现。
·索引程序变得更易操作。因为由机器失效、慢速机器以及网络临时阻塞引起的大部分问题自动地由MapReduce库解决，不需要操作人员干预。另外，通过在索引集群中增加机器可以容易地提高索引程序的性能。
7．相关工作
许多系统提供了严格的编程模型，并且使用这些限制来自动地并行化计算。例如，一个associative函数可以在logN的时间内在N个处理器上使用并行前缀计算[6,9,13]来计算N个元素的数组的所有前缀。MapReduce可以看作是基于我们在真实世界大型计算的经验，对其中一些模型的简化和净化。更重要的是，我们提供了可扩展为上千台处理器的容错实现。相比而言，大部并行处理系统只是在小规模上实现，并且将处理机器故障的细节留给了程序员。
大型同步编程[17]和一些MPI原语[11]提供了更高级别的抽象，可以使程序员容易地编写并行程序。MapReduce和这些系统的关键区别在于，MapReduce开发了一个受限的编程模型自动地并行化用户程序，并提供透明的容错处理。
我们本地优化的灵感来源于活跃硬盘[12,15]，其中计算被推送到靠近本地磁盘的计算元素，继而减少了通过网络或IO子系统传输的数据量。我们运行在直接连接少量硬盘的日常机器上，而不是直接运行在在磁盘处理器上，但是大体的方法是类似的。
我们的备用任务机制类似于Charlotte System[3]应用的迫切调度机制。简单迫切调度机制的其中一个缺点是如果一个给定任务反复失效，整个计算无法完成。我们通过跳过坏记录的方式在我们机制中修正了这个问题的一些实例。
MapReduce的实现依赖于一个内部的集群管理系统，该系统负责在一大群共享机器上分布和运行用户任务。尽管不是本文的侧重点，这个集群管理系统在理念上和其它系统是一样的，如Condor[16]。
MapReduce库的一部分-排序附加功能和类似于对NOW-Sort[1]的操作。源机器（map workers）把待排序的数据分区后，发送到R个reduce worker中的一个。每个reduce worker在本地对数据进行排序（尽可能在内存中）。当然，NOW-Sort没有用户自定义的Map和Reduce函数，这俩函数使我们的库具有广泛的适用性。
River[2]提供了一个编程模型，其中程序通过分布式队列发送数据进行互相通信。和MapReduce类似，River系统试图即使在由异构硬件和系统扰动引入的不均匀面前也提供良好的平均情况性能。River通过精心调度硬盘和网络传输来均衡完成时间以达到这个目的。通过限制编程模型，MapReduce框架能够把问题分成大量精细任务。这些任务在可用的worker上动态调度，因此速度快的worker执行更多的任务。受限的编程模型也使我们可以调度接近作业末尾的任务冗余执行，减少在不均匀面前的完成时间（比如有较慢或者受阻的worker）。
BAD-FS[5]有一个和MapReduce不同的编程模型，不像MapReduce那样，它是针对广域网上的作业执行。然而，有两个基本相似点。（1）两个系统都使用冗余执行的方式从失效导致的数据丢失中恢复。（2）两个都使用本地化调度减少通过拥堵网络连接发送的数据量。
TACC[7] 是一个设计用来高可用性网络服务的构造的系统。与MapReduce类似，它也依赖冗余执行作为实现容错的机制。
8．总结
MapReduce编程模型在Google内部成功用于多个目的。我们把这种成功归因为几个方面：首先，由于它隐藏了并行、容错、本地优化、负载均衡的细节，即便对于没有并行和分布式系统经验的程序员而言，模型也易于使用；其次，大量不同种类的问题都可以作为MapReduce计算简单的表达。比如，MapReduce用于生成Google的网络搜索服务产品所需要的数据、用来排序的数据、用来数据挖掘的数据、用于机器学习的数据，以及许多其它系统；第三，我们开发了一个扩展到由数千台机器组成的大型集群上的MapReduce实现。这个实现使得使用这些机器资源很有效，因此也适合用在Google内部遇到的许多大型计算问题中。
我们也从这项工作中学到了一些事。首先，限制这个编程模型使得并行和分布式计算非常容易，对计算进行容错也非常容易；其次，网络带宽是稀缺资源。大量的系统优化因此是旨在减少通过网络传输的数据量：本地优化允许我们从本地磁盘读取数据，将中间数据的单独一份拷贝写入本地磁盘节约了网络带宽；第三，冗余执行可以用来降低慢速机器的影响（alex注：即硬件配置的不平衡），并且可以用来处理机器失效和数据丢失。
附录：词频程序
本节包含这样一个程序：程序统计在一组由命令行指定的输入文件中每个唯一的词出现的次数。
#include "mapreduce/mapreduce.h"
//用户的map函数
Class WordCounter : public Mapper{
public:
virtual void Map(const MapInput& input) {
const string& text = input.value();
const int n = text.size();
for(int I = 0;I < n;) {
//跳过导致空格的i
while((i < n) & & isspace(text[i]))
i++;
//找到词尾
int start = i;
while((i < n) && !isspace(text[i]))
i++;
if(start < i)
Emit(text.substr(start,i-start),"1");
}
}
};
REGISTER_MAPPER(WordCounter);
//用户的reduce函数
class Adder : public Reducer {
virtual void Reduce(ReduceInput* input) {
//遍历所有key值相同的条目并累加value值
int64 value = 0;
while(!input->done()) {
value += StringToInt(input->value());
input->NextValue();
}
//Emit sum for input->key()
Emit(IntToString(value));
}
};
REGISTER_REDUCER(Adder);
int main(int argc,char** argv){
ParseCommandLineFlags(argc,argv);
MapReduceSpecification spec;
//将输入文件列表存入"spec"
for(int i = 1;I < argc;i++) {
MapReduceInput* input = spec.add_input();
input->set_format("text");
input->set_filepattern(argv[i]);
input->set_mapper_class("WordCounter");
}
//指定输出文件:
///gfs/test/freq-00000-of-00100
///gfs/test/freq-00001-of-00100
//...
MapReduceOutput* out = spec.output();
out->set_filebase("/gfs/test/freq");
out->set_num_tasks(100);
out->set_format("text");
out->set_reducer_class("Adder");
//可选:在map任务内部进行并行加和以节省网络带宽
out->set_combiner_class("Adder");
//调节参数：每个任务使用至多2000台机器和100M的内存
spec.set_machines(2000);
spec.set_map_megabytes(100);
spec.set_reduce_megabytes(100);
//运行
MapReduceResult result;
if(!MapReduce(spec,&result))
abort();
//Done:result结构包含计数器，花费的时间以及用到的机器数目等
Return 0;
}
