标签：map 翻译 机器 worker 论文 reduce MapReduce 任务

MapReduce: Simplified Data Processing on Large Clusters（MapReduce: 简化大型集群下的数据处理）

作者：Jeffrey Dean and Sanjay Ghemawat

Abstract（摘要）

MapReduce是一个关于实施大型数据集处理和生成的编程模型。
用户指定一个用于处理k/v对，生成中间态k/v集合的映射（map）函数，以及一个用于合并所有具有相同中间态key的中间态value值的归约（reduce）函数。
正如本篇论文所展示的那样，很多现实世界中的任务都可以通过该模型（MapReduce）表达。

以这种函数式风格编写的程序可以在一个大型的商品级(注：意思是很廉价、性能也很一般)机器集群中自动、并行的执行。

该系统在运行时会关注如下细节：输入数据的分割，在一系列机器间跨机器的调度程序的执行，机器故障的处理以及管理集群内机器间的必要通信。

使用MapReduce，使得没有任何并行计算、分布式系统经验的程序员们都可以轻松利用大型分布式系统中的资源。

我们已实现的MapReduce运行在一个大型商品级机器集群上，而且具有高度的可拓展性：一个典型的MapReduce计算可以在数千台机器上处理TB级别的数据。

程序员们发现系统很容易使用：已经有数以百计的MapReduce程序被实现，并且每天都有一千以上的MapReduce任务运行在谷歌的（计算机）集群中。

1. Introduction（介绍）

在过去的五年时间里，包括作者在内的许多谷歌工作人员实现了数以百计的、用于特殊目的的计算程序来处理大量的原始数据，例如爬虫获取到的文档、网络请求日志等等。

其目的是为了计算出各种类型的衍生数据，例如倒排索引、多种关于web文档的图结构表示、被每个主机所爬取的页面数摘要、给定的某天中被最频繁查询的集合等等。

大多数这样的计算在概念上都很简单，然而输入的数据却通常是巨大的。而且为了能在一个合理的时间范围内完成，计算操作需要被分配到数百甚至数千台机器上运行。

关于如何并行计算，如何分派数据以及如何处理故障等问题被混杂在了一起，使得原本简单的计算逻辑被用于处理这些问题的大量复杂代码所模糊。

为了应对这些复杂性，我们设计了一个全新的抽象，该抽象允许我们表达我们想要执行的简单计算，但是将关于并行化、容错、数据分发和负载均衡等机制中复杂、繁琐的细节隐藏在了库中。
我们的这一抽象其设计灵感是来源于Lisp和很多其它函数式语言中的map和reduce原语。

我们意识到我们的绝大多数计算都涉及到为每一个输入的逻辑记录应用(applying)一个map映射操作，目的是对输入集计算从而将其转化为一个中间态的k/v对集合；

然后为了恰当地合并衍生数据，再对所有拥有相同key值的k/v对中的value值应用一个reduce规约操作。

通过一个由用户指定具体逻辑的map和reduce操作的函数式模型，使得我们能轻易地并行化大规模的计算，并且将重复执行（自动重试）机制作为容错的主要手段。

这项工作的主要贡献在于提供了一个简单且强大的接口，该接口能够使大规模计算自动地并行化和分布式的执行。

结合该接口的实现，得以在大型的商品级PC集群中获得很高的性能。

第二章介绍基本的编程模型并给出了几个示例。

第三章介绍一个针对集群计算环境的MapReduce接口实现。

第四章介绍几个我们发现的，关于该编程模型的有效改进。

第五章则是关于我们对各种任务所实施的性能测试。

第六章探讨MapReduce在谷歌内部的应用，其中包括了我们以MapReduce为基础去重建索引生成系统的经验。

第七章讨论一些相关的话题以及日后要做的工作。

2. Programming Model（编程模型）

这一计算获得并输入一个k/v键值对集合，然后生成并输出一个k/v键值对集合。
MapReduce库的用户通过Map和Reduce这两个函数来表达该计算逻辑。

Map函数是由用户编写的，其获得一个输入的k/v对并生成一个中间态的k/v对。
MapReduce库对所有的k/v对进行分组，使得所有有着相同中间态key值的k/v对的value值组合在一起，然后将它们传递给Reduce函数。

Reduce函数也是由用户编写的，其接收一个中间态的key值和与该键对应的一组value值的集合。
它会将这些value值进行统一的合并以形成一个可能更小的value值集合。
通常，每次reduce调用只会生成零个或一个输出值。这个中间态的value集合通过一个迭代器提供给用户的reduce函数。
这允许我们得以处理那些无法被完整放入内存的，过大的列表集合。

2.1 Example（示例）

思考一个关于在一个大型文档集合中计算每一个单词出现次数的程序。
用户可能会写下形如以下伪代码的程序：

map(String key, String value):
    // key: document name
    // value: document 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 values:
        result += ParseInt(v);
    Emit(AsString(result));

这个map映射函数生成每一个单词以及其出现的次数（在这个简单的例子中次数恰好是1）。
reduce函数则累加每一个生成的特定单词其所有的出现计数。

此外，用户编写代码以指定的输入、输出文件的名字和可选的调优参数来填充一个规范的mapreduce对象。
用户随后调用MapReduce函数，传递这个符合规范的对象。用户的代码与MapReduce库（c++实现）进行链接。
附录A包含了本示例的完整程序文本。

2.2 Types（类型）

尽管前面的伪代码是依据字符串类型的输入、输出编写的，但从概念上说，用户提供的map和reduce函数在类型上是有关联的：

map (k1,v1) --> list(k2,v2)
reduce (k2,list(v2)) --> list (v2)

举个例子，输入的key和value和输出的key和value分属不同的域。
此外，中间态的key和value和输出的key和value则属于相同的域。

我们在c++的实现中传递字符串，以作为用户自定义函数的输入和输出，并将字符串（类型）与合适的类型间进行转化的逻辑留给用户代码实现。

2.3 More Examples（更多的例子）

这里有几个很容易用MapReduce计算来表达的有趣程序的简单示例。

分布式Grep:
map函数如果匹配某个给定规则则输出对应的那一行。
reduce函数是一个恒等函数，其只是将所输入的中间数据原封不动的复制到输出（译者小熊餐馆注：恒等函数：f(x) = x, 即输入=输出）。

URL访问频率计数:
map函数处理网页请求的处理日志，并且输出<URL,1>的键值对。
reduce函数累加所有具有相同URL键值对的value值，并且输出一个<URL,总访问数>的键值对。

反向web链接图:
map函数从每一个源页面（source）中找出每一个目标页URL（target）的链接，输出（target，source）格式的kv对。
reduce函数将所有具有相同target目标页的所有源页面（source）结合在一起组成一个列表，输出这样一个kv对（target,list(source)）。

每台主机的检索词向量:
汇总从一个或一系列文档中出现的最重要的单词作为检索词向量（term-vector），生成以<word(单词),frequency(出现频次)>格式的kv对列表。
map函数针对每一个输入的文档，输出一个<hostname(主机名),term vector(检索词向量)>的kv对（主机名是从文档的URL中提取出来的）。
reduce函数接收一个给定host下基于每个文档的所有term-vectors(检索词向量)。
将这些检索词向量进行累加，抛弃掉一些出现频率较低的检索词项然后返回最终的<hostname(主机名),term vector(检索词向量)>的kv对。

倒排索引:
map函数解析每一个文档，然后输出一连串<word(单词)，documentID(文档ID)>格式的kv对。
reduce函数接收一个给定单词对应的所有kv对，针对文档ID进行排序然后返回一个<word(单词),_list_documentID(文档ID列表)>格式的kv对。
所有输出的kv对集合构成了一个简单的倒排索引。基于此，我们能简单的增加记录每一个单词(在这些文档中)的位置的计算功能。

分布式排序:
map函数提取每一个记录中的key值，然后返回一个<key,record>格式的kv对。
reduce函数对所有的kv对不做修改直接返回。
该计算依赖于后续4.1章节中所述的分区机制和4.2章节中所述的有序性机制。

3.Implementation(实现)

MapReduce接口有很多种实现方式。具体选择哪一种取决于具体的环境。

例如，某一种实现方式可能适合较小共享内存的机器，而另一种实现方式则适用于大型的多核处理器机器，还有的实现方式则更适用于基于网络的大型机器集群。

本章介绍的一个(MapReduce)实现是针对谷歌内部广泛使用的计算环境的：通过交换式以太网互相连接起来的大型商品级PC集群。

在我们的环境中:

(1) 机器通常是有着x86架构的双处理器的、运行linux操作系统的平台，每台机器有2-4GB的内存。

(2) 使用商品级的网络硬件 - 通常每台机器的带宽为100M/s或者1GB/s，但其平均(实际使用的)带宽远小于整个网络带宽的一半。

(3) 整个集群是由几百或几千台机器所组成的，因此机器故障是频繁出现的。

(4) 存储是由直接连接到独立机器上的IDE硬盘提供的。存储在这些磁盘上的数据由一个内部自研的分布式文件系统来管理。这一文件系统采用复制机制，旨在不可靠的硬件之上实现可用性和可靠性。

(5) 用户提交作业(job)给一个调度系统。每个作业都由一系列的任务(task)组成，任务被调度器映射到内部集群中的一组可用机器上去执行。

3.1 Execution Overview(执行概述)

通过将输入的数据自动分割为M份，map调用得以分布在多个机器上调用执行。拆分后的输入数据可以被不同的机器并行的处理。
通过一个分区函数将中间态的key值空间划分为R份(例如: hash(key) mod R, 对key做hash后再对R求模)，Reduce调用也得以分布式的执行。
分区的个数(R)和分区函数都由用户来指定。

图1展示了我们所实现的MapReduce操作的总体流程。当用户程序调用MapReduce函数时，会发生以下的一系列动作（图1中的数字标号与以下列表中的数字是对应的）:

1. MapReduce库将输入的文件拆分为M份，每份大小通常为16MB至64MB（大小可通过参数控制）
2. 在集群中启动多个程序的副本，其中一个是master。剩下的都是worker,worker由master分配任务
3. 现在有M个map任务和R个reduce任务需要分配。master选择空闲的worker，并为每一个被选中的worker分配一个map或reduce任务
4. Map Worker:分配了map任务的worker，从输入的数据中解析key/value对，并将每一个kv对作为参数传递给用户自定义的map函数。map函数产生的中间态key/value对然后缓存在内存中。
5. 缓存中的kv对会被周期性地写入本地磁盘，通过分区函数将其划分成R个区域内。这些缓冲在本地磁盘上的位置会被发送给master，master将这些位置信息转发给reduce worker。
6. 当reduce worker获得位置信息后，它会通过RPC从map worker的本地磁盘读取缓冲数据。当reduce worker读完所有的中间数据后，它会根据key值进行排序，从而将key值相同的数组放在同一组。排序是必需的，因为通常会有很多不同的key映射到同一个reduce任务。如果中间数据量太大，无法放入内存中，则可使用外部排序。
7. reduce worker遍历已排序的中间数据，对于每个Key，它将该Key和与之对应的中间值集合传递给用户的reduce函数。Reduce函数产生的结果，会被写入到这个Reduce的结果输出文件中。
8. 当所有map和reduce任务都完成后，master唤醒用户程序。此时，用户程序中对MapReduce调用，就会获得结果。

成功完成后，MapReduce执行结果会被存放在R个输出文件中(每个reduce任务对应一个输出文件，文件名由用户指定)。通常，用户不需要将这R个输出文件合并为一个文件 — 他们通常将这些文件作为输入传递给另一个MapReduce调用，或者由另一个能处理多个被分割输入文件的分布式应用进行处理。

3.2 Master Data Structures(Master数据结构)

master中维护了一些数据结构。对于每个map和reduce任务，master存储了对应的任务状态(闲置，运行中，或已完成)，以及worker机器的标识(对于非空闲的任务)。

master是一个管道，将中间态文件的位置信息从map任务传递给reduce任务。
因此，对于每个已完成的map任务，master存储了map生成的R个中间态文件区域的位置和大小。当map任务完成时，master将接受到位置和大小信息的更新。这些信息会以增量的方式推送给运行中的reduce任务。

3.3 Fault Tolerance(容错)

MapReduce库的设计目标是使用成百上千台机器处理海量数据，因此必须能够处理机器故障。

Worker Failure

主进程周期性地ping每个工作进程。如果在一段时间内没有收到worker的响应，master会将该worker标记为失败。所有由该worker负责的map或reduce任务都会被重置为初始状态，这些任务会被调度去其他worker重新执行。

已完成的map任务在故障时需要重新执行，因为map任务的输出存储在故障机器的本地磁盘上，无法被访问。而已完成的reduce任务不需要重复执行，因为它们的输出存储在全局文件系统中。

一个map任务由A执行，但是A执行失败，转而由B执行，此时所有执行reduce任务的worker都会被通知map任务被重新执行了。之前从A读取数据，现在转从B读取数据。

MapReduce能够抵御大规模的工作节点故障。在一次执行任务时，集群因网络维护导致80台机器同时在几分钟内无法访问。master只是重新执行了那些故障机器负责的工作，之后继续执行，最后完成了任务。

Master Failure(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会记录下来R文件的信息，如果master重复收到已完成map任务的完成消息，它会忽略这个消息。

当一个reduce任务完成时，reduce worker会自动将临时的输出文件重命名为最终输出文件。如果在多台机器上执行同一个reduce任务，则会对同一个最终输出文件执行多次rename调用。我们依赖底层文件系统所提供的原子性重命名操作，以确保最终文件系统中只包含一次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 Locality(就近原则)

在我们的计算环境中，网络带宽是一个相对稀缺的资源。
我们利用输入的数据(由GFS管理)存储在组成集群的机器的本地磁盘上来节约网络带宽。
GFS将每个文件分割为64MB的块，并在不同的机器上存储每个块的若干副本(通常为3份)。
MapReduce master会考虑输入文件的位置信息，并尝试将map任务调度到包含相应输入数据副本的机器上。
如果做不到这一点，它会尝试将map任务安排在该任务输入数据的副本附近(例如，在与包含数据的机器在同一个网络交换机上的工作机器上)。
当集群中的相当一部分worker都在执行大型MapReduce操作时，绝大多数的输入数据都在本地读取从而不会消耗网络带宽。

3.5 Task Granularity(任务粒度)

如前所述，我们将map任务拆分为M份，将reduce任务拆分为R份。

理想情况下，M和R的值都应该远大于worker机器的数量。

让每一个worker执行很多任务可以提高动态负载均衡的效率，同时也能加快当某个worker故障时的恢复速度：故障worker的map任务可以分散到其它worker上重新执行。

在我们的实现中，对M和R的实际大小做了限制，因为master的调度必须维持在O(M+R)的时间复杂度，同时要在内存中保持O(M*R)个记录跟踪任务状态。(然而，内存使用的常量因素很小：这O(M*R)份的状态里，每个map/reduce任务对由大约1字节的数据组成)。

除此之外，用户通常会限制R的大小，因为每个reduce任务的输出最终都会保存在一个单独的输出文件中。
实践中，我们倾向于设置M的大小，这样每个任务大约有16 MB到64 MB的输入数据(使得上文所述的Locality优化效果最好), 同时我们设置R的大小为我们预期使用worker机器数量的小倍数。我们经常使用2000台工作机器执行M = 20000和R = 5000的MapReduce计算。

3.6 Backup Tasks(备份执行)

导致MapReduce总耗时变长的一个常见原因“掉队者”：一台机器花费了异常长的时间去完成最后的几个map或reduce任务。“掉队者”出现的原因有很多。例如，有坏磁盘的机器会频繁地遇到可纠正的错误，这将使其读取性能从30 MB/s降至1 MB/s。集群调度系统可能在这台机器上调度了其它任务，导致MapReduce执行变慢，因为要竞争CPU、内存、本地磁盘或网络带宽。我们最近遇到的一个问题是机器初始化代码中的一个bug，它导致处理器缓存被禁用:受影响机器上的计算速度下降了100倍以上。

我们有一个通用的机制来缓解"掉队者"问题。当一个MapReduce任务接近完成时，master会调度剩余任务进行后备执行(backup executions)。即同一个任务交由多个节点执行，无论主节点还是后备节点执行完成，任务都会被标记为已完成。我们还对这个机制进行了优化，使的这一操作所使用的计算资源增加不超过几个百分点。这一操作大大减少了大型MapReduce操作的时间。例如，如果禁用后备任务机制，5.3节所述的sort程序需要多花44%的时间才能完成。

4. Refinements(改进)

尽管已提供的编写简单Map和Reduce函数的功能能满足大多数需求，但我们还发现了一些有价值的拓展。本章节将对此进行介绍。

4.1 Partitioning Function(分区函数)

MapReduce用户指定他们想要的reduce任务/输出文件的数量(R)。
在这些任务中，使用一个基于中间态key的分区函数对数据进行分区。
(我们)提供了一个使用哈希取模的默认分区函数(例如：hash(key) mod R)。
这往往会得到一个非常均衡的分区结果。
然而在有些情况下，使用其它的一些基于key的分区函数对数据进行分区是很有用的。
举个例子，有时(map任务)输出的key是URL，且我们希望同一个主机上的所有条目最后都写入同一个输出文件中。
为了支持这种场景，MapReduce库的用户可以提供一个特殊的分区函数。
例如，使用hash(Hostname(urlkey)) mod R作为分区函数，就可以使得来自同一个主机的所有URL最终都写入同一个输出文件中。

4.2 Ordering Guarantees(有序性保证)

我们保证在给定的分区内，中间态的k/v对是以中间态key值递增的顺序处理的。这一有序性保证使得生成每个分区的有序输出文件变得容易，当输出文件的格式需要支持基于key的随机访问查找，或用户需要数据有序时会很方便。

4.3 Combiner Function(组合器函数)

在一些情况下，每个map任务生成的中间态key存在明显的重复，同时用户自定义的reduce函数具备可交换性和可结合性。
2.1章节中的单词计数的示例程序就是一个很好的例子。
由于单词出现的频率遵循齐夫分布，每个map任务都将产生成百上千条形如<the,1>形式的记录。
所有的这些计数将通过网络发送给单独的reduce任务，然后由reduce函数将它们相加得到一个数字。
我们允许用户指定一个可选的Combiner函数，在数据通过网络发送前对数据进行部分合并。
Combiner函数在每台执行map任务的机器上执行。
通常，combiner和reduce函数使用相同的代码。
reduce函数和combiner函数间唯一的不同在于MapReduce是如何处理函数的输出。
reduce函数的输出会写入最终的输出文件中。
而combiner函数的输出会被写入到中间态文件中，然后发送给reduce任务。

部分合并可以显著加快某些MapReduce操作的速度。
附录A中包含了一个使用combiner的例子。

4.4 Input and Output Types(输入和输出的类型)

MapReduce库支持读取多种不同格式的输入数据。
例如，"文本"模式下将每一行视为一个键值对：key是该行在文件中的偏移量，value是该行的内容。
另一种支持的常用格式存储基于key排序的kv键值对序列。
每个输入类型的实现都知道如何将输入的数据分割为有意义的区间，以便作为单独的map任务处理(例如，文本模式的范围分割确保范围分割只发生在行边界)

用户可以通过提供简单reader接口的实现来添加对新输入类型的支持，尽管大多数用户只使用少量预定义输入类型中的一种。

reader不一定需要提供从文件读取的数据，举个例子，很容易定义一个reader，从数据库或映射在内存中的数据结构中读取记录。类似的，我们也支持多种不同格式的输出数据，并且用户代码很容易添加对新输出类型的支持。

4.5 Side-effects(副作用)

在某些情况下，MapReduce的用户可以很方便地使用map/reduce操作生成辅助文件作为额外的输出。
我们依赖应用程序的作者确保这些副作用具有原子性和幂等性。
通常，应用程序会写入一个临时文件，并在该文件完全生成后自动重命名该文件。 我们不支持由单个任务生成多个输出文件的原子两阶段提交。因此，产生多个具有跨文件一致性要求的输出文件的任务应该是确定的。

在我们的实践中，这一限制并没有带来什么问题。

4.6 Skipping Bad Records(跳过错误的记录)

有时用户的代码中存在一些bug，造成了Map或Reduce函数在处理某些数据时一定会崩溃。这些bug会阻止MapReduce操作的完成。
通常的做法是修复这个bug，但有时这是行不通的；可能这个bug位于三方库中，且无法获得其源代码。
当然，有时忽略掉少量的数据是可接受的，比如对一个大型数据集上进行统计分析时。
我们提供了一个可选的执行模式，当MapReduce库检测到某些记录一定会导致崩溃时，跳过这些记录并继续向前推进。

每个worker进程都安装了一个信号处理器，用于捕获段异常(segmentation violations)和总线错误(bus errors)。
在调用用户的Map或Reduce操作前，MapReduce库会将参数的序列号存储在一个全局变量中。
如果用户代码产生了一个信号，则信号处理器将会向MapReduce的master发送一个包含了(该参数)序列号的"最后喘息(last gasp)"UDP包。
当master一个特定的记录不止一次的导致故障时，master会指示对应的Map或Reduce任务在下一次重新执行时应该跳过该记录。

4.7 Local Execution(本地执行)

在实际计算发生在分布式系统中时，调试Map或Reduce函数会变得很棘手，通常由master动态的在几千台机器上决定工作的分配。
为了更利于调试、分析和小规模的测试，我们开发了一个(运行在本地机器上的)MapReduce库的可替代实现，该库能让所有的MapReduce工作在本地机器上顺序执行。
控制权被交给了用户,使得计算可以被限制在指定的Map任务中。
用户通过一个特殊的标志来调用他们的程序，然后可以轻松地使用任何他们觉得好用的调试或者测试工具(例如：gdb)。

4.8 Status Information(状态信息)

master机器运行了一个内置地Http服务器，并提供了一系列地状态信息页面供用户访问。
状态信息页面会展示计算的进度，例如有多少任务已经完成，多少任务正在执行中，输入数据的字节数，中间数据的字节数，输出数据的字节数，处理速度等等信息。
页面也包含了指向每个任务对应的标准误差(standard error)和标准输出文件的链接。
用户可以使用这些数据预测还要多长时间完成计算，以及是否需要为该计算投入更多资源。
这些页面也可用于找出为什么实际的计算比所预期的要慢的原因。

此外，高级状态页面展示了哪些worker机器发生了故障，以及哪些map和reduce任务在执行时发生了故障。
在尝试调试用户代码中的bug时这些信息会很有用。

4.9 Counters(计数器)

MapReduce库提供了一个计数器的功能，用于计数不同事件出现的次数。
例如，用户代码可能会想要统计已经处理过的单词总数或者被编入德文文档的索引数等等。

为了使用这一功能，用户代码中需要创建一个名为计数器的对象，然后在Map或Reduce函数中以恰当的方式对计数器进行累加操作。
例如：

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

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

独立worker机器中的counter值会周期性的传递给master(在ping响应包中附带)
master将来自已经成功完成的map和reduce任务中的counter值聚合在一起，并在MapReduce任务完成时返回给用户代码。
当前的counter值也会展示在master的状态页上，使得用户可以看到实时的计算进度。
在聚合counter值时，master消除了同一个map或reduce任务多次执行的影响，避免了重复计数。
(多次执行出现的原因是我们的备份任务或任务故障时的重复执行导致的)

有些counter值是由MapReduce自行维护的，例如已处理的输入k/v对的数量和已生成的输出k/v对的数量。

用户发现计数器功能能很好的用于检查MapReduce操作的行为是否正常。
例如，在某些MapReduce操作中，用户代码想要确保已生成的k/v对数量严格等于已处理的输入k/v对数量，或者确保已处理的德语文档数量在已处理的全部文档中的占比是否处于一个可接受的比例内。

5 Performance

在这一章节，我们通过在大型机器集群上运行的两个MapReduce计算来测量MapReduce的性能。
其中一个计算是在大约1TB的数据中检索特定的模式。
另一个计算是对大约1TB的数据进行排序。

上述两个程序代表了现实中大多数MapReduce用户所编写的程序，一类程序将数据从一种表示方式转化为另一种表示方式，而另一类程序则从一个大的数据集中提取出少量感兴趣的数据。

5.1 Cluster Configuration

集群配置

所有的程序都在一个由大约1800台机器组成的集群上被执行。
每台机器都配置有两颗开启了超线程功能的、2GHZ主频的Intel至强处理器，4GB的内存，两块160GB容量的IDE硬盘，以及一条千兆的以太网链路。
所有机器都被安置在一个双层的树形交换网络中，根节点处的总可用网络带宽大概为100-200GB每秒。
所有的机器都位于同一个主机托管设施(hosting facility)内，因此任意一对机器间的(网络交互的)往返时间都低于1毫秒。

在4GB的内存中，大约1-1.5GB的内存是为集群上要运行的其它任务而保留的。
任务是在周末的下午执行的，(因为)这个时间点CPU、硬盘和网络一般都是空闲的。

5.2 Grep

(Globally search a Regular Expression and Print 基于正则表达式的全局搜索并打印)

grep程序扫描通过扫描10^10个100字节大小的记录，搜索一个相对比较少见的3字符模式(这个模式只出现在92337条记录中)。
输入数据被分割为大约64MB大小的块(M = 15000)，并且完整的输出被放在了一个文件中(R = 1)。

图二展示了随时间推移的计算进度。
Y轴标识着扫描输入数据的速率。
随着越来越多的机器被分配给当前MapReduce计算，扫描输入数据的速率也越来越快，并且当分配了1764个worker机器时其峰值达到了30GB每秒。
当map任务完成后，扫描输入数据的速率开始下降并在计算执行到大约80秒的时候降至0。
整个计算从开始到结束大概耗时150秒。
这其中包括了一分钟左右的启动开销。
这一开销是由于需要将程序分发到所有的worker机器上，以及为了打开1000个输入文件集合而与GFS交互并获得局部性优化信息的延迟。

5.3 Sort(排序)

这个排序程序对10^10个100字节大小的记录进行排序(大约1TB的数据)。
这个程序是参照TeraSort基准测试程序而编写的。

排序程序包含了少于50行的用户代码。
一个三行的Map函数从一个文本行中提取出一个10字节大小的、用于排序的key并且发出该key，并将原始的文本行作为value而生成中间态的k/v键值对。
我们使用内置的恒等函数(Identity function)作为Reduce算子。
这个函数传入中间态的k/v键值对，并且不做任何修改的将之作为输出的k/v键值对。
最终完成排序的输出被写入了一个双向复制的GFS文件集合中(即程序总共写入、输出了2TB的数据)。

如上所述，输入的数据被分给为64MB的块(M = 15000)。
我们将排好序后的输出数据分割为4000个文件(R = 4000)。
分区函数基于key的初始字节值将其分割为R份。

我们的基准测试中内置的分区函数是了解key值具体分布的。
在一个常规的排序程序中，我们会预先插入一个MapReduce操作，该操作将会收集key值的一个样本并且基于key值样本的分布情况来计算最终排序时需要的分割点。

图3的a部分展示了一个排序程序的正常执行过程。
左上角的图表标识了输入数据读取的速率。
输入数据速率的峰值为13GB每秒，由于所有的map任务都在200秒内完成了因此其非常快速地降到了零。
请注意输入速率是小于上述地grep程序的。
这是因为排序的map任务有一半的耗时和I/O带宽用于将的中间态的输出写入它们机器的本地磁盘。
而相应的，grep任务的中间态输出则可以忽略不计。

左边排中间的图表标识了map任务通过网络将数据发送给reduce任务的速率。
这一转换在第一个map任务完成不久后便开始了。
图表中的第一个高峰对应着第一批的大约1700个reduce任务(整个MapReduce分配了1700台机器，并且每一台机器同一时间至多只能执行一个reduce任务)
大概执行了300秒的计算时，第一批的一些reduce任务陆续完成并且剩余的reduce任务继续转换数据。
所有的转换大概在计算执行了600秒时完成。

左下方的图表标识了reduce任务将排序好的数据写入最终的输出文件的速率。
在第一个转换(shuffling)阶段结束到开始写入之间存在一点延迟，其原因是机器此时正忙于对中间态的数据进行排序。
写入数据的以2-4GB每秒的速率持续了一段时间。所有的写入大约在计算执行至850秒左右时完成。
包括启动的开销在内，整个计算过程共耗时891秒。
这与TeraSort基准测试目前已报告的最快记录很相近。

有几点值得注意：

• 输入的速录比转换和输出的速率要高很多，其原因在于我们进行了局部性优化。大多数的数据是从本地的硬盘中读取的，从而避免使用我们相对有限的网络带宽。
• 转换速率比输出速率要高很多，其原因在于输出阶段写入了已排序数据的两个备份(出于可靠性和可用性的考虑，我们构建了两个输出数据的备份)。
  我们写入两个备份的原因在于这是我们底层文件系统所提供的可靠性和可用性的机制。
• 如果底层文件系统使用纠错码(Erasure Coding)来代替复制(来保证可靠性)，则需要写入数据时所需要的网络带宽将减少很多。

5.4 Effect of Backup Tasks

后备任务的影响

在图3的b部分，我们展示了禁用后备任务时排序程序的执行状况。
执行流与图3的a部分很相似，除了最后面有一个非常长的尾部，且其几乎没有任何写入发生(注意观察代表done的那条竖线)。
在960s后，除了5个reduce任务外其它任务都已经完成。
然而最后几个“落伍者”任务直到300秒后才相继完成。整个计算过程共花费了1283秒，(相比正常执行的情况)增加了44%的耗时。

5.5 Machine Failures

机器故障

在图3的c部分，我们展示了一个排序程序的执行流程，在其计算过程中我们故意在几分钟内杀死(killed)了1746台worker机器中的200台机器(的worker进程)。
底层的集群调度器立即在这些机器上重新启动新的worker进程(因为只是杀掉了worker进程，机器依然是正常工作的)。

worker进程被杀死时展示一个负的输入速率，因为之前已完成的任务失效了(因为对应的map worker被杀掉了)并且这些任务需要被重新执行。
map任务的重新执行相对来说是比较快的。
包括启动开销在内，整个计算过程共耗时933秒（相较于正常执行时的耗时，只增加了5%）

6 Experience

经验

我们于2003年2月编写了第一版的MapReduce库，并且在2003年的8月对其进行了重大改进，其中包括局部性优化、跨worker机器间任务执行的动态负载均衡等等。
从那时起，我们惊喜的看到MapReduce库被广泛的应用于我们工作中所遇到的各种问题上。
MapReduce库已在谷歌内的许多领域中被广泛的使用，其中包括：

• 大规模的机器学习问题
• Google新闻和Froogle产品的聚类问题(clustering problems)
• 基于常见查询所产出的报告提取数据(例如，Google Zeitgeist (注：Google开发的一款网络查询分析程序))
• 基于新实验和产品的网页提取相关属性(例如，从用于本地化搜索的大型网页语料库中提取地理位置)
• 大规模的图计算

图4展示了登记在我们主要的源码管理系统中的独立MapReduce程序数量随着时间的推移有着显著的增长，
从2003年年初的0个,再到2004年的9月有了接近900个独立的MapReduce程序实例了。
MapReduce如此成功的原因在于其使得编写一个简单的程序，并在半小时内于上千台机器上高效的运行成为了可能，这极大地加快了开发和原型设计的周期。
此外，MapReduce允许没有任何分布式或并行系统开发经验的程序员得以轻松的利用大量的(计算)资源。

在每个job完成时，MapReduce库会以日志的形式记录对应job所使用的计算资源的统计信息。
在表1中，我们展示了谷歌在2004年8月所运行的MapReduce job的一个子集的(所使用计算资源的)一些统计信息。

6.1 Large-Scale Indexing

大规模索引

迄今为止，我们对MapReduce最重要的一个应用就是完全重写了索引生成系统，其生成的数据结构被用于Google web的搜索服务。
索引系统将我们的爬虫系统所检索到的、被存储为一系列GFS文件的大量文档作为输入。
这些文档的原始内容的数据大小超过了20TB。
整个索引处理过程由5到10个连续的MapReduce操作组成。
使用MapReduce(而不是之前版本索引系统的点对点分布式传输)能带来几个好处:

• 索引相关的代码变得更简单、(代码量)更少和更容易理解，因为处理容错、分布式和并行化的代码被隐藏在了MapReduce库内部。
  例如，某一计算阶段的代码量在(改为)使用MapReduce表达后从(原来的)大约3800行c++代码降低至大约700行。

• MapReduce库的性能是足够好的，这使得我们可以将概念上无关的计算进行拆分，而不是将它们混合在一起，从而避免额外的数据传输。
  这使得可以简单的改变索引的处理过程。
  举个例子，在我们老的索引系统中曾进行的一次改动耗费了我们几个月的时间，而在新系统中去实现则只需要几天时间。

• 处理索引变得更容易操作，因为大多数机器故障、机器执行缓慢和网络间歇性断开(networking hiccups)等问题都由MapReduce库自动处理了，而不需操作员介入。
  此外，通过向索引处理的集群中添加新的机器可以轻松地提高索引处理的性能。

7 Related Work

相关工作

很多系统都提供了受限制的编程模型，并且使用这些约束来自动的将计算并行化。
举个例子，使用并行前缀计算时，一个结合函数可以在N个处理器上，以logN的时间计算出一个N元素数组的所有前缀。
MapReduce可以被认为是基于我们在现实世界中关于大型计算的经验所总结出的一些模型的一个简化和精炼。
更重要的是，我们提供了一个可拓展到几千个处理器规模的容错实现。
相比之下，大多数的并行处理系统的实现只能运用在更小的规模下，并且将处理机器故障的细节留给了程序员(去实现)。

整体同步程序(Bulk Synchronous Programming)和一些消息传递接口(MPI Message-Passing Interface)原语提供了更高级别的抽象，使得程序员可以更加简单的编写并行程序。
这些系统与MapReduce最关键的不同在于MapReduce利用一个受限的编程模型令用户程序自动的并行化并且了提供透明的(用户无需感知的)容错机制。

我们局部性优化机制的灵感源自active disks等技术，推进计算并使得所要处理的元素是靠近本地磁盘的，以减少通过网络I/O子系统发送的数据量。
我们的计算运行在直连少量磁盘的商用处理器上，而不是直接运行在有着磁盘控制器的处理器(disk controller processors)上，但大致的方法是类似的。

我们的后备任务机制类似于Charlotte系统中所应用的紧急调度(eager scheduling)机制。
简单的紧急调度机制的一个缺点就是如果一个给定的任务反复失败，则整个计算将无法完成。
我们通过跳过有问题记录的机制，一定程度上的修复了这一问题。

MapReduce的实现依赖于一个内部的集群管理系统，该系统负责在大量的共享机器中分发和运行用户的任务。
虽然这并不是本论文的重点，但该集群管理系统从本质上来说和Condor系统非常相似。

排序机制做为MapReduce库的一部分，在操作上与NOW-Sort类似。
源机器(map workers)将待排序的数据进行分区，并将其发送给R个reduce worker中的一个。
每一个reduce worker在本地对数据进行(尽可能的在内存中排序)。
当然，NOW-Sort不支持使得可用户自定义的Map和Reduce函数，相比之下我们的MapReduce库则有着更广的适用范围。

River提供了一个编程模型，该模型中进程间通过向分布式队列中发送数据来进行通信。
和MapReduce一样，即使由于异构的硬件或者系统扰动而导致了(计算资源的)不均衡，River系统也试图在这种场景下提供足够均衡的性能。
River通过仔细的对磁盘和网络传输进行调度，用以实现任务完成时间的平衡。
MapReduce则采用了不同的方法。
通过受限的编程模型，MapReduce框架能够将一个问题分割为大量细粒度的任务。
这些任务会在可用的worker机器上动态的调度，因此运行速度更快的worker能够处理更多的任务。
这一受限的编程模型也允许我们在job接近完成时进行冗余任务的调度，这可以极大地减少在非均衡场景下的任务完成时间(比如存在缓慢或者卡住不动的worker)。

BAD-FS是一个与MapReduce非常不同的编程模型。与MapReduce不同，其致力于跨广域网的执行job。
然而，这里有两个很相似的基本点。
两个系统都使用冗余的执行来恢复由故障导致的数据丢失。
两者都使用距离敏感的调度策略，用以减少在拥挤的网络链路上所发送数据的数量。

TACC是一个旨在简化高性能网络服务构造的框架。
和MapReduce一样，其也依赖重复执行机制来实现故障容错。

8 Conclusions(总结)

MapReduce编程模型已经成功的在谷歌中被广泛应用。
我们认为这一成功出于几个原因。
首先，这一模型很容易使用，因为其隐藏了并行化、故障容错、局部性优化以及负载均衡的细节，即使是没有并行计算和分布式系统经验的程序员也能轻松地使用。
其次，各种各样的问题都能用MapReduce计算轻松地表达。
例如，MapReduce被用于为谷歌的网络搜索产品生成数据、也被用于排序、用于数据挖掘、用于机器学习以及其它的很多系统。
再次，我们已开发的MapReduce实现可以被扩展到包含数千台机器的大型集群中。
这一实现使得众多机器资源能被有效的利用，因此其很适合谷歌所遇到的许多大型计算问题。

我们从这项工作中学到了一些事情。
首先，受限制的计算模型能够简化并行化和分布式计算，并且能够令这些计算具有容错性。
其次，网络带宽是一种稀缺资源。
因此我们的系统中有许多致力于减少在网络中传输数据数量的优化：局部性优化允许我们从本地磁盘中读取数据，以及将中间态数据的单个备份写入本地磁盘以节约网络带宽。
再次，冗余的重复执行可以用于减少慢机器的影响，以及处理机器故障和数据丢失。

Acknowledgements(致谢)

Josh Levenberg基于他使用MapReduce的经验以及其它人提出的优化建议，在修改MapReduce的用户级API和为其拓展很多新特性的过程中发挥了重要作用。
MapReduce是基于谷歌文件系统GFS读取输入数据和写出输出数据的。
我们要感谢Mohit Aron、Howard Gobioff、Markus Gutschke、David Kramer、Shun Tak Leung和Josh Redstone为开发GFS所做的工作。
我们也要感谢Percy Liang和Olcan Sercinoglu为开发MapReduce集群管理系统所做的工作。
Mike Burrows, Wilson Hsieh, Josh Levenberg, Sharon Perl, Rob Pike和Debby Wallach为这篇论文的前期草稿提供了很有帮助的建议。
匿名的OSDI评论员和我们的审核者Eric Brewer就论文可以改进的方面提供了许多有用的建议。
最后，我们感谢谷歌工程部的所有MapReduce用户，感谢他们提供的有价值的反馈、建议和bug报告。

A Word Frequency

一个单词频率统计程序

这一章节包含了一个程序，用于计算由命令行指定的一组输入文件集合中每个唯一单词的出现次数。

#include "mapreduce/mapreduce.h"
// User’s map function 
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; ) { 
            // Skip past leading whitespace 
            while ((i < n) && isspace(text[i])) 
               i++;
            // Find word end 
            int start = i; 
            while ((i < n) && !isspace(text[i]))
               i++;
            if (start < i) 
               Emit(text.substr(start,i-start),"1");
        }
    }
};   


REGISTER_MAPPER(WordCounter);
// User’s reduce function 
class Adder : public Reducer { 
    virtual void Reduce(ReduceInput* input) { 
        // Iterate over all entries with the 
        // same key and add the values 
        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;
    // Store list of input files into "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");
    }
    // Specify the output files: 
    // /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");
    // Optional: do partial sums within map 
    // tasks to save network bandwidth 
    out->set_combiner_class("Adder");


    // Tuning parameters: use at most 2000 
    // machines and 100 MB of memory per task 

    spec.set_machines(2000); 
    spec.set_map_megabytes(100); 
    spec.set_reduce_megabytes(100);

    // Now run it 
    MapReduceResult result; 
    if (!MapReduce(spec, &result)) abort();

    // Done: ’result’ structure contains info
    // about counters, time taken, number of 
    // machines used, etc.
    return 0;
}

针对MapReduce强语义、弱语义概念译者自己的理解

(译者小熊餐馆注：
上面这段内容比较晦涩，这里根据我举个简单的例子来帮助大家理解。
假设有一段话：“Your name is Tom? My name is Tom, too.”，原始需求是想利用MapReduce计算统计分词后每个单词出现的次数(例子里句子很短是为了描述，实际上可以是海量的文档)。
我们自定义的Map函数是确定性的函数算子，输入这个字符串进行Map操作后总是会返回以下9个kv对(key是单词，value是出现的次数): <Your,1>, <name,1>, <is,1>, <Tom,1>, <My,1>, <name,1>, <is,1>, <Tom,1>, <too,1>。
无论Map函数是单机单线程顺序执行，还是在集群中并行的执行，结果都是明确不变的，也就是上述的强语义的概念。
MapReduce库会把Key相同的kv对进行分组，并将其传递给我们自定义的reduce函数，下面是分组后会传给reduce函数算子的参数：
<Tom,list(1,1)>, <name,list(1,1)>, <is,list(1,1)>, <Your,list(1)>, <My,list(1)>, <too,list(1)>。
在原始需求下，当map函数计算的结果不变时，无论reduce函数算子何时执行，也无论出现故障重复执行了几次，得到的结果一定和单机单线程顺序执行相同，这也是强语义。
结果：<Tom,2>, <name,2>, <is,2>, <Your,1>, <My,1>, <too,1>。 (key为单词，value为出现的次数)
而如果改变原始需求，除了累加单词总共出现的次数还要返回reduce计算时的当前机器id。
那么此时的reduce函数就属于不确定的函数算子了，因为即使输入相同，但每一次的执行获得的结果都不一定相等（调度到不同机器上执行，机器id不同，输出的结果也就不同）。
假设有两台reduce任务worker，id分别为aaa和bbb。
id为aaa的worker机器上reduce任务的执行结果就是<Tom,2-aaa>, <name,2-aaa>, <is,2-aaa>, <Your,1-aaa>, <My,1-aaa>, <too,1-aaa>,是为结果result_aaa。
id为bbb的worker机器上reduce任务的执行结果则是<Tom,2-bbb>, <name,2-bbb>, <is,2-bbb>, <Your,1-bbb>, <My,1-bbb>, <too,1-bbb>,是为结果result_bbb。
上述的弱语义表示，无论出现了什么机器故障，虽然无法准确的得知结果到底是哪一个，但最终结果不是result_aaa就是result_bbb，反正一定是某一个reduce任务生成的完整输出数据，而绝不可能出现跨任务的数据重复、冗余、缺失等问题。
)

标签：map,翻译,机器,worker,论文,reduce,MapReduce,任务
From： https://www.cnblogs.com/INnoVationv2/p/18379261