《MapReduce: Simplified Data Processing on Large Clusters》阅读

## 摘要

MapReduce是一种用来处理和生成大数据集的编程模型以及相关实现。用户选定一个map函数，该map函数的输入是一个键值对，输出是一系列中间键值对。然后用户选定一个reduce函数，该reduce函数用于处理具有相同键的所有中间键值对，并得到结果。实际生产中，有大量应用可以用此模型描述。

这种编程模型可以在集群上并行工作。运行时系统会自动分割输入数据、分发任务到各个机器、处理任务失败或者机器故障、管理必要的机器间通信。因此，即使没有分布式系统经验的程序员，也可以方便地利用分布式系统的资源。

现在每天有大量MapReduce模型的工程运行于谷歌集群，其具有高度的扩展性，一个典型的MapReduce计算可以利用集群处理TB级别的数据集。

## 1.介绍

过去数年，大量特殊目的的计算程序被设计出来，用于处理大量的原始数据集，比如爬虫结果、网络请求日志，来得到需要的衍生数据，比如倒排索引、web文档的图结构、某日查询最多的关键词。虽然这类计算在概念上非常直观，但是由于输入数据集非常庞大，因此需要将这些输入数据集切割、分发到多个机器上处理。但是除了原来本有的计算模型外，还需要编写大量复杂的代码去处理并行化计算、数据分发、故障处理等问题。

因此我们干脆设计一种新的抽象，将并行计算、容错、数据分发、负载均衡这些复杂技术细节隐藏起来，交给库去完成，我们只需要定义计算模型即可。这种抽象的灵感来自于Lisp等函数式编程语言中的Map和Reduce原语。我们意识到许多计算都需要对输入的每个记录应用一个"map"操作来得到中间一系列键值对，之后再对所有拥有相同键的值应用"reduce"操作，得到衍生数据集。因此如果一个编程模型，允许我们通过定义map操作和reduce操作，那我们就可以方便设计大量并行计算，并且我们可以使用重新执行作为主要的容错机制。

这次工作的主要贡献是提供一个简单而强大的接口，可以使用这个接口完成自动并行化和大规模分布式计算，并且该接口的实现在大型商用集群上性能卓然。

本文第二章给出基本编程模型以及一些栗子。第三章介绍为集群计算环境量身定做的MapReduce接口。第四章介绍一些对于编程模型的改进。第五章测试了我们的实现在不同任务下的性能。第六章介绍了谷歌内部关于MapReduce的使用情况，包括用其重写我们的产品索引系统。第七章相关讨论和未来展望。

## 2.编程模型

这些计算以输入作为一系列键值对，并输出一系列键值对。MapReduce库的用户调用Map和Reduce两个函数。

Map函数的输入是一个键值对，并输出一系列中间键值对。然后MapReduce库将相同键的中间键值对一起送入Reduce函数。

Reduce函数，接收一个中间键值对的键I和该键对应的所有值。Reduce函数将这些值进一步融合处理成更小的集合。通常每次Reduce函数的调用只有0个或者1个输出值。Reduce函数通过迭代器来中间键值对，因此我们也可以应对数据量超过内存大小的情况。

### 2.1 例子

如何统计大量文档中每个单词的频次？伪代码如下：

```
map(String key, String value): // key: 文档名, value: 文档内容
    for each word w in value:
        EmitIntermediate(w, "1");
        
reduce(String key, Iterator values): // key: 单词， values:一系列计数结果
    int result = 0;
    for each v in values:
        result += ParseInt(v);
    Emit(AsString(result));
```

map函数给每个单词出现时的计数加1.reduce函数将某个单词的所有计数相加。另外，用户需要填写输入输出文件名、可选的调度参数等。然后用户将自定义的map和reduce函数作为参数传入MapReduce库指定的结构中（本文库用C++实现）。附录A包含该例子的完整程序。

### 2.2 类型

尽管之前的伪代码的变量类型为字符串类型，但实际上map和reduce函数的参数可以泛化描述如下：

```
map(k1, v1) -> list(k2, v2)
reduce(k2, list(v2)) -> list(v2)
```
输入键值的类型和输出键值的类型不同，但是中间键值对的键值类型和输出键值类型相同。我们的C++实现的输入参数为字符串，用户需要自己将字符串类型转化为所需要的类型。

## 2.3 更多示例

以下介绍一些可以用MapReduce计算表达的简单例子。

分布式Grep: map函数获取匹配行，并传入reduce函数，reduce函数只将中间结果添加到输出

URL访问频次统计： map函数处理web网页请求日志，并输出`<URL, 1>`。reduce函数将相同url键的值加在一起，获取`<URL, total count>`键值对

反向web链接图：source网页中有指向target网页的链接，map函数输入为`<source, target>`，中间输出`<target, source>`键值对。reduce函数将某个给定target对应的所有source网页输出，得到`<target, list(source)>`

Term-Vector per Host: term vector表示文档中频率较高的一些单词，表示为`<word, frequency>`。map函数输出`<hostname, term vector>`（hostname从文档的url中解析）。Reduce函数获取某个host的所有`<hostname, term vector>`，融合并丢弃频率不高的单词，得到最终的`<hostname, term vector>`

倒排索引：map函数解析每个文档，输出一系列`<word, document ID>`。reduce函数将某个单词对应的所有文档排序输出得到`<word, list(document 。ID)>`。所有输出集合组合成一个简单的倒排索引。通过MapReduce模型可以很容易的增大算力来追踪单词位置

分布式排序：map函数从每个输入记录中提出要排序的指标key，输出`<key, record>`。reduce函数只将中间结果输出即可。

## 3. 实现

MapReduce适用于许多场景，但是正确的实现是随着场景改变而改变的。例如，一种实现也许适用于小型共享内存机器，另一种实现适用于大型NUMA多核机器，也有适用于更大机器集群的实现。

本章节描述Google中最常见的计算环境：大型商用集群。环境描述如下：

（1）机器通常都是x86的双核处理器，其上运行Linux，每台机器拥有2-4G的内存

（2）商用网络硬件，通常是100 M/s或者1 G/s，但是综合起来要小于平均带宽

（3）一个集群由成千上万台机器组成，因此机器故障是常有的事

（4）存储由便宜的IDE磁盘提供，它们都与独立的机器直接相连。一个内部研发的文件系统用于管理所有存储于这些硬盘上的文件。该文件系统通过Replication在不可靠的硬件上提供了可用性和可靠性

（5）用户提交jobs给调度系统。每个job由一系列的task组成，并且由调度器分配到集群中一系列可用的机器上

### 3.1 执行总过程

![title](https://leanote.com/api/file/getImage?fileId=5e5cd9dbab64411fcb006749)

将输入数据切割为M份，分发多台机器上并行调用map函数处理，得到中间键值对。中间键值对通过hash函数划分为R份，R的值和划分函数由用户定义。上图展示了MapReduce的总体流程。

1. MapReduce库首先将输入文件切割成M个切片，每片大小为16M~64M（亦可通过可选参数指定），之后将程序拷贝到集群中的其他机器上。
2. 程序的拷贝过程中，有一个程序的拷贝比较特殊，即master。其余均是workers，受master调度。有M个map任务和R个reduce任务要分发。master选举空闲workers机器，每台机器分发一个map任务或者一个reduce任务
3. 分发到map任务的worker读取对应序号的输入切片。它会从输入数据中解析键值对，并由map函数迭代，map函数输出的中间键值对缓存在内存中
4. 被缓存的中间键值对会被阶段性的写入本地磁盘，并通过划分函数分给为R份。中间键值对在本地磁盘中的保存位置会被回传给master，master再负责将中间键值对的位置信息转发至reduce任务的机器。
5. 当reduce worker拿到中间键值对的位置信息，通过rpc读取map workers本地磁盘上的缓存数据。当reduce worker读取了它对应序号的所有中间键值对，其会按照键排序，让相同键的值连续摆放。之所以要排序，是因为不同的key会分到同一个reduce任务中。如果中间键值对的数目过多、内存不够，那么需要借助额外的排序步骤。
6. reduce work迭代排序过的中间键值对，对于每一个键和其对应的所有值，调用reduce函数处理，reduce函数迭代输出结果被添加到这个reduce任务对应的输出文件中
7. 当所有map任务和reduce任务完成时，master唤醒用户程序，至此，MapReduce程序调用结束

当MapReduce完成之后，其结果存放于R个输出文件中（每个reduce任务对应一个输出文件），通常用户不需要将R个文件合并成一个，而是将其作为另外一个MapReduce程序的输入

### 3.2 Master数据结构

master保持多个数据结构，对于每个map task和reduce任务而言，master存储每个任务的状态（idle, in-process, completed），任务所在机器id等等。

master相当于一个管道，map任务通过这个管道将中间结果的位置信息传递给reduce任务。因此，对于每个完成的map任务，master会存储R个中间文件的位置信息及其大小。当map任务完成时，master就会接收到这些信息，然后转发给具有in-process状态reduce任务的worker。

### 3.3 容错

因为MapReduce库的设计初衷是，用成千上万的机器处理大数据集，所以它必须在故障处理方面足够优雅。

#### Worker故障

master周期性的ping每个worker，如果某个worker在给定时间内每一响应，则标记为failed。该worker完成的所有map task都会被回退为idle状态，然后被调度到其他worker上重新执行。同理，所有failed worker上正在执行的map和reduce任务均回退为idle状态，并重新调度到其他worker上执行。

由于failed worker上的所有map务的结果都保存在本地，机器故障导致中间结果无法获取，因此只能重新调度运行。但是已经完成的reduce任务不需要重新运行，因为reduce的输出是保存在全局的文件系统中。

当一个map任务首先由worker A执行，稍后又由worker B执行（因为A失败了），所有正在执行reduce task的worker都会收到该map任务重新执行的通知，所有尚未从worker A读取数据的reduce task将会从B中读取数据。

MapReduce对于大规模的worker故障是有弹性的。比如，在一次MapReduce操作中，某集群上的网络维护导致80台机器一段时间内不可达。MapReduce只需要将不可达机器上的任务重新调度执行即可。

#### Master故障

可以通过周期性的快照来备份master的数据结构，当master worker故障，只需要根据最新的快照重新启动一个新的worker即可。并且，当有且仅有一个master worker时，master几乎不可能发生故障。因此在我们的实现中，如果master故障，则直接舍弃本次MapReduce的结果，用户可以根据情况选择重新执行MapReduce。

#### 故障发生时的处理

如果用户定义的map和reduce函数是关于输入值的确定性函数，那么在我们的分布式实现，在经过无故障的顺序执行之后，会有同样的输出。

我们可以通过map和reduce任务原子性地提交来实现该特性。每一个in-process任务将输出写到私有临时文件中，一个reduce task创建一个私有临时文件，一个map task会创建R个私有文件。当map task完成时，worker将R个文件的名称汇报给master。当master接收到一个已经完成的map task发送来的信息时，则忽略；否则，master记录R个文件的名称。

当reduce task完成时，reduce worker自动重命名临时输出文件为最终输出文件。如果多台机器上运行了相同的reduce task，那么多个重命名调用之间会发生数据竞险。我们依靠底层文件系统的原子性重命名操作来保证，最终的文件只来自一个reduce task。

绝大多数map和reduce操作是确定性的，语义上，可以将程序理解为顺序执行，因此用户可以方便的理解和解释程序。即使map操作或者reduce操作是不确定性的，我们仍然提供了较弱、但仍然合理的语义。当map操作或reduce不是关于输入的确定性函数时，每次执行的结果也可能不同，但是仍然可以解释。

考虑map task M和reduce task R1、R2。令e(R_i)为已经提交的R_i操作。那么可能会发生更弱语义的情况，比如e(R_1)可能会从M的某次执行中读取结果，而e(R_2)可能会读取M的另外一次执行的结果。

### 局部性

网络带宽在我们的计算环境中是较为稀缺的资源。因此我们将输入数据保存在集群机器的本地磁盘。GFS文件系统将文件切割成64M的文件块，将这些文件块在不同的机器上存储备份（通常是3个备份）。MapReduce的master会感知所有文件的位置信息，将map task分配给包含对应输入文件的机器上。如果map worker发生故障，则master尝试将map task调度到邻近备份机器上（比如同一个局域网下的、包含备份数据的机器）。当集群中运行MapReduce操作时，由于输入数据从本地磁盘读入，因此不会在MapReduce操作时浪费带宽资源。

### 任务粒度

我们将map操作划分为M个，reduce操作划分为R个。通常，M和R的数量远大于机器数量，所以一个机器上可以运行多个task，来提高负载均衡，同样当worker失败时task可以迅速恢复。

在实现中，M和R的大小应该有一个界限。master需要做O(M+R)个调度决定，在内存中保持O(M*R)个状态（但是内存使用的常数还是比较小的，O（M*R）个Map task\/Reduce task 状态对，每个的大小大概在一个字节）。R的大小由用户定义，因为每个reduce task的输出文件都分散在不同的文件中，M的大小建议在16M~64M之间（使得上述的局部性优化效果最好），我们一般会将R设置为机器数目的较小倍数。比如，我们通常执行MapReduce计算的配置为，M = 200000，R = 5000，机器数目2000。

### 3.6 备份任务

导致MapReduce操作耗时较长的一个常见原因是Straggler——“落伍的士兵”：一台机器花了过长时间计算最后几个map或者reduce任务。Straggler出现的原因很多。例如，磁盘损坏导致读取性能从30MB/s降到1MB/s。或者，集群系统调度了其他任务给这台机器，使得这台机器上的CPU、内存、磁盘、网络带宽等资源竞争加剧。最近我们遇到的一个问题是，机器的初始化代码有一个bug导致处理器缓存被禁用，以至于机器的计算速度下降到百分之一。

我们设计了一个通用机制来缓解Straggler问题。当MapReduce操作快完成时，master会备份剩余几个正在执行的tasks。不管是备份的task执行完还是原来的执行完，该task都会被标记完成。我们通过调整将该操作导致的计算资源消耗仅仅提高了几个百分点。这种机制在大型MapReduce操作时，以提高一点点资源的代价，减少了大量空耗的时间。例如，5.3中描述的排序程序，备份机制关闭，会多消耗44%的时间。

## 4. 改进

尽管通过简单的定义Map和Reduce函数可以满足绝大部分需求，但是我们还是提出了一些其他有用的扩展。

### 4.1 划分函数

用户定义reduce task的数目R，通过划分函数将中间键值对分类。默认的划分函数为`hash(key) mod R`，会产生比较均匀的划分。但是在一些其他例子中，用其他划分函数会更好点。比如，当输出键为URLS，一个host对应所有urls放在一个文件中，为了达到这个目的，用户可以将划分函数定义为`hash(Hostname(urlkey)) mod R`。

### 4.2 顺序保证

在一个划分区域内，中间键值对按照键的升序排列，这样可以方便地得到一个排序好的输出文件，当在输出文件需要根据key来查找的时候会比较方便有效，或者当用户需要排序好的数据时更加便利。

### 4.3 合并函数（Combiner function）

在一些例子中，一个map task会产生大量重复键的中间键值对。比如，在2.1中提及的单词统计中，单词的频率分布满足Zipf分布，每个map task可能会产生成百上千个`<the, 1>`，这些计数结果通过网络转发给reduce函数由reduce函数合并。我们允许用户自定义可选的Combiner函数，在网络发送结果前，预先进行一些合并操作。

Combiner函数在每台map task的机器上执行，通常cominer函数和reduce函数相同，唯一不同的地方在于，reduce函数将结果写入到最终输出文件，而combiner函数将结果写到一个中间文件，而这个中间文件再会被发送给reduce task。

部分合并操作可以加速某类特定的MapReduce操作，附录A包含一个使用了combiner函数的例子。

### 4.4 输入和输出类型

MapReduce库支持读取不同格式的输入数据。例如，txt格式文件的每一行输入会被当做一个键值对，key为文件内偏移，值为该行的内容。另外一种比较常见的格式就是，存储着一系列排好序的键值对。每一种格式的数据都会被合理的切割。用户也可以通过reader接口添加对新的输入数据类型的支持，但通常用户只会用到预定义类型的一部分。

reader接口未必是从文件中读取数据，也定义成从数据库中读取数据，或者从内存中读取数据。

类似地，我们支持了不同格式的输出数据类型，用户也可以通过接口添加对新的输出数据类型的支持。

### 4.5 副作用

在一些例子中，map或者reduce操作中产生额外的辅助文件，这需要用户自己保证这些辅助文件的原子性和幂等性。比如说，应用将结果写到一个临时文件中去，一旦所有结果被写入完成之后，可以原子性的将其重命名为最终文件。

在单独的任务中，我们不保证其多个输出文件的原子性两相提交支持。因此，会产生交叉输出文件的task需要保证足够的一致性。

### 4.6 跳过坏记录

有时候用户代码中存在bug，导致Map或者Reduce函数在某些记录上执行时程序崩溃，以致于MapReduce操作无法完成。遇到这种情况，一般我们会去修复Bug，但是有时候这个bug是闭源第三方库导致的，我们无法拿到第三方库的源代码。同时，有时候在大数据集上作分析时，允许我们忽略掉一些记录。我们提供了一种可选的执行模式，即，当MapReduce哭检测到某些记录致使造成崩溃时，直接跳过，继续执行之后的记录。

每一个worker进程配对了捕捉段错误和bug的信号句柄。在调用用户的Map或者Reduce操作前，MapReduce将错误码存储在全局变量中。如果用户代码触发一个信号，信号句柄会发送一个"last gasp"的UDP包给master，UDP包中包含了记录的序列号。当master发现某个记录上触发了不止一次的错误后，该记录会在之后的map或者reduce操作中直接被跳过。

### 4.7 本地执行

Map或者Reduce函数的调试比较tricky，因为其通常运行于分布式系统中。为了帮助调试、分析以及小规模测试，我们开发了一种另外一种实现，其可以在本地机器上顺序执行MapReduce的所有操作。用户可以指定到限定的map task中执行程序。用户使用指定的flag调用程序，并使用debug或者测试工具（比如，gdb）。

### 4.8 状态信息

master运行了一个内置的HTTP服务，并配备了一些人类可读的状态页面。状态页展示了计算 进度，比如多少task已经完成、多少正在进行中、输入数据大小、中间数据的大小、输出数据大小、处理速度等。该网页还包含指向每个task生成的标准错误和标准输出文件的链接。用户可以利用这些数据预判计算持续时间、计算是否还需要添加更多资源等。这些页面还可以用来检查什么时候处理速度下降。

此外，顶层的状态页展示哪些worker失败，当worker失败时哪些map和reduce task正在处理中。这些信息在诊断bug时十分有用。

### 4.9 计数器（Counters）设施

MapReduce库提供了一个counter设施来统计各种事件发生的频次。例如，用户也许想统计已经处理过的单词总数或者德国文件索引数量。

用户代码可以通过创建counter对象，然后在Map或Reduce函数中对counter增加，比如：
```
Counter *uppercase;
uppercase = GetCounter("uppercase");

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

每个机器上counter值会被定期回传给master(附带在ping的响应中)。master将成功执行的map和reduce任务回传的counter值聚集，然后在MapReduce操作完成时返回给用户代码。master的状态页上也可以看到当前的counter值，所以用户可以实时观看计算的进行。在聚集counter值时，master会消除因Map或者Reduce task的重复执行造成的重复计算。（重复执行可能由backup tasks或者因为错误重新执行的task引起）。

MapReduce库会自动维护一些counter值，比如，已处理的输入键值对数目以及已输出键值对数目。

counter设施适用于检查MapReduce的执行情况。比如，在一些MapReduce操作中，用户代码需要确保输出键值对数目和已经处理的输入键值对数目相同，又或者，处理的德国文档占总处理文档的比例在一个适当的范围内。

## 5. 性能

本节我们测试MapReduce性能，手段是在一个集群上运行两个MapReduce计算。第一个计算是搜索一个T的数据，从查找特定模式的数据。第二个计算是对1个T的数据进行排序。

这两个程序分别代表了MapReduce程序的两个大类。一类程序则从大数据集中抽取出一小部分需要的数据，另一类程序用于将数据从一种表示方法切换到另一种表示方法。

### 5.1 集群配置

集群包含1800台机器。每台机器都有两个2GHz的Intel Xeon处理器，且Hyper-Threading打开， 4GB内存，两个160GB的IDE磁盘，以及一个有线网卡。机器放置在一个两层树状的交换机网络中，根节点的带宽大概在100-200Gbps。所有机器在同一个机房中，因此任意两台机器间的通信时间少于1ms。

其中4GB中的1~1.5G是为集群中运行的其他任务预留的。程序在一个周末的下午运行，此时CPU、磁盘、网络基本都处于空闲状态。

### 5.2 Grep

![title](https://leanote.com/api/file/getImage?fileId=5e5e14eeab6441629800252d)

grep程序需要扫描10^10个100字节的记录，搜索一个相对罕见的三字符模式（出现了92337次）。输入数据被分成大小为64MB的块（M = 15000），所有输出存放在一个文件中（R = 1）。

Figure 2显示了计算随着时间的变化过程。Y轴代表了输入数据的扫描速度。随着更多机器被分配到MapReduce的计算当中，速度越来越快。当有1764个worker加入时，达到峰值30GB/s。随着Map task的结束，速度开始下降并且在80s的时候到达0,。整个Computation从开始到结束总共花费了大概150s。这其中还包括了1分钟的启动开销。开销主要来源于将程序分发到worker machine中、和GFS交互并打开1000个输入文件、以及获取局部性优化所需信息的延时。

### 5.3 Sort

![title](https://leanote.com/api/file/getImage?fileId=5e5e1a04ab6441609b002607)

sort程序需要将10^10个100字节的记录排序（大概1TB数据）。程序用TeraSort benchmark建模。该sort程序的用户代码不超过50行。map函数有三行，其从文本行中提取10字节的key，并将key和原文本航作为中间键值对。我们使用内置的Identity函数作为Reduce操作，该函数将中间键值对原封不动作为输出。最终排序的输出被写入2通道的GFS文件集合中（即该程序的输出是2TB）。

在程序开始之前，输入数据被切割成大小64MB的文件块（M = 15000）。我们将排序输出划分为4000个文件（R = 4000）。划分函数使用key的前几个字节，划分到R片中的一个。

本次测试使用的划分函数基于key的分布。通常，我们会添加一个预处理的MapReduce程序收集keys样本，然后使用keys的分布来计算最终排序程序的数据分割点。

图3（a）显示了排序程序的正常执行进度。左上角的图表显示读取输入的速率。该速率峰值约为13GB/s，并很快消失，因为所有映射任务都在200秒之前完成。注意，输入速率小于Grep程序。这是因为排序映射任务将大约一半的时间和I/O带宽在写入到它们的本地磁盘上。而Grep程序相应的中间结果大小可以忽略不计。

左中图显示通过网络将数据从map task发送到reduce task的速率。第一个map task完成后，shuffling将立即开始。图中出现第一个峰值时，是第一批大约1700个reduce任务（整个MapReduce被分配了大约1700台机器，每台机器一次最多执行一个reduce任务）正在执行。计算大约300秒后，第一批reduce任务中的一些任务完成，我们开始为剩余的reduce任务传送数据。所有的shuffling都是在大约600秒的时间内完成的。

左下角的图表显示了reduce任务将已排序数据写入最终输出文件的速率。第一次shuffling结束与写入周期开始之间存在延迟，因为机器正忙于对中间数据进行排序。写操作会以大约2-4 GB/s的速度继续一段时间。所有的写操作在计算过程中大约完成850秒。包括启动开销，整个计算需要891秒。这与TeraSort基准测试当前1057秒的最佳报告结果类似。

需要注意的是：由于我们的局部性优化，输入速率高于shuffling速率和输出速率–大多数数据是从本地磁盘读取的，并绕过了我们相对带宽受限的网络。由于输出阶段写入两个已排序数据的副本（出于可靠性和可用性的原因，我们制作两个输出副本），因此shuffle速率高于输出速率。我们编写两个副本，因为这是底层文件系统提供的可靠性和可用性机制。如果底层文件系统使用擦除编码而不是复制，则写入数据的网络带宽要求将降低。

## 5.4 Backup tasks的效果

在图3（b）中，我们显示了在禁用备份任务的情况下执行sort 程序。执行过程与图3（a）类似，只是有一个很长的尾巴，几乎没有任何写活动发生。960秒后，除5个reduce任务外的所有任务都完成。然而，这最后几名落伍者直到300秒后才落网。整个计算需要1283秒，用时增加44%。

## 5.5 机器故障

在图3（c）中，我们展示了排序程序的执行过程，在计算的几分钟内，我们故意杀死了1746个工作进程中的200个。底层的集群调度程序立即重新启动这些计算机上的新worker进程（因为只有进程被终止，所以这些计算机仍然正常工作）。

worker被杀死，图标中输入速率出现负数，这是因为以前完成的一些map worker消失（因为相应的map worker被杀死），需要重新进行。map worker的重新执行相对较快。包括启动开销，整个计算在933秒内完成，（仅比正常执行时间增加5%）。

## 6.谷歌内部使用的经验之谈

我们在2003年2月编写了MapReduce库的第一个版本，并在2003年8月对其进行了重大改进，包括本地优化、跨工作机执行任务的动态负载平衡等，我们对MapReduce库在我们所处理的各种问题中的广泛应用感到惊喜，包括：

* 大规模机器学习问题
* Google新闻和Froogle产品
* 提取用于生成热门查询报告的数据（如谷歌时代精神）
* 为新的实验和产品提取网页属性（例如，从大量网页中提取地理位置进行本地化搜索）
* 大规模图计算

![title](https://leanote.com/api/file/getImage?fileId=5e5e2646ab6441609b002853)

Figure 4显示了随着时间的推移，检入我们的源码管理系统的独立MapReduce程序的数量显著增加，从2003年初的0个增加到2004年9月底的近900个独立实例。MapReduce非常成功，因为它可以在半小时内编写一个简单的程序并在1000台机器上高效运行，大大加快了开发和原型制作周期。此外，它允许没有分布式和/或并行系统经验的程序员轻松地利用大量资源。

![title](https://leanote.com/api/file/getImage?fileId=5e5e26d4ab6441609b00286b)
在每个工作结束时，MapReduce库统计并记录该项工作使用的计算资源信息。在table 1中，我们显示了2004年8月在Google运行的一部分MapReduce项目的一些统计数据

### 6.1 大规模索引

目前，我们对MapReduce最重要的使用之一是完全重写了产品索引系统，该系统生成用于Google web搜索服务的数据结构。索引系统将我们的爬虫系统检索到的大量文档作为输入，存储为一组GFS文件。这些文档的原始内容超过20TB的数据。索引过程经历5~10个MapReduce过程。使用MapReduce（而不是索引系统老版本中的ad-hoc分布式传递）提供了几个好处：

* 代码更简单、更少、更易读，因为容错、分布式和并行化的代码隐藏在MapReduce库中。例如，某个计算阶段的代码行数从近似3800行的C++代码下降到大约700行。
* MapReduce库的性能足够好，我们可以将概念上不相关的计算分开，而不是将它们混合在一起来避免额外的数据传递。这使得更改索引过程变得很容易。例如，在我们的旧索引系统中进行的一项更改花费了几个月的时间，而在新系统中实现这项更改只花了几天的时间。
* 索引过程变得更容易操作，因为大多数由机器故障、机器速度慢和网络故障引起的问题都由MapReduce库自动处理，而无需程序员干预。此外，通过在索引集群中添加新机器，可以方便地提高索引过程的性能

## 附录A

```cpp

#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;
}
```

逆风起笔