本文的Google MapReduce中文翻译源自 http://blademaster.ixiezi.com/2010/03/27/google-mapreduce中文版/

摘要

1、介绍

2、编程模型

2.1、例子

2.2、类型

（alex注：原文中这个domain的含义不是很清楚，我参考Hadoop、KFS等实现，map和reduce都使用了泛型，因此，我把domain翻译成类型推导的域）。
我们的C++中使用字符串类型作为用户自定义函数的输入输出，用户在自己的代码中对字符串进行适当的类型转换。

2.3、更多的例子

3、实现

3.1、执行概括

在成功完成任务之后，MapReduce的输出存放在R个输出文件中（对应每个Reduce任务产生一个输出文件，文件名由用户指定）。一般情况 下，用户不需要将这R个输出文件合并成一个文件–他们经常把这些文件作为另外一个MapReduce的输入，或者在另外一个可以处理多个分割文件的分布式 应用中使用。

3.2、Master数据结构

3.3、容错

当worker故障时，由于已经完成的Map任务的输出存储在这台机器上，Map任务的输出已不可访问了，因此必须重新执行。而已经完成的Reduce任务的输出存储在全局文件系统上，因此不需要再次执行。

使用MapReduce模型的程序员可以很容易的理解他们程序的行为，因为我们绝大多数的Map和Reduce操作是确定性的，而且存在这样的一个 事实：我们的失效处理机制等价于一个顺序的执行的操作。当Map或/和Reduce操作是不确定性的时候，我们提供虽然较弱但是依然合理的处理机制。当使 用非确定操作的时候，一个Reduce任务R1的输出等价于一个非确定性程序顺序执行产生时的输出。但是，另一个Reduce任务R2的输出也许符合一个 不同的非确定顺序程序执行产生的R2的输出。

3.4、存储位置

3.5、任务粒度

但是实际上，在我们的具体实现中对M和R的取值都有一定的客观限制，因为master必须执行O(M+R)次调度，并且在内存中保存O(M*R)个状态（对影响内存使用的因素还是比较小的：O(M*R)块状态，大概每对Map任务/Reduce任务1个字节就可以了）。

3.6、备用任务

4、技巧

虽然简单的Map和Reduce函数提供的基本功能已经能够满足大部分的计算需要，我们还是发掘出了一些有价值的扩展功能。本节将描述这些扩展功能。

4.1、分区函数

4.2、顺序保证

4.3、Combiner函数

部分的合并中间结果可以显著的提高一些MapReduce操作的速度。附录A包含一个使用combiner函数的例子。

4.4、输入和输出的类型

Reader并非一定要从文件中读取数据，比如，我们可以很容易的实现一个从数据库里读记录的Reader，或者从内存中的数据结构读取数据的Reader。

类似的，我们提供了一些预定义的输出数据的类型，通过这些预定义类型能够产生不同格式的数据。用户采用类似添加新的输入数据类型的方式增加新的输出类型。

4.5、副作用

如果一个任务产生了多个输出文件，我们没有提供类似两阶段提交的原子操作支持这种情况。因此，对于会产生多个输出文件、并且对于跨文件有一致性要求的任务，都必须是确定性的任务。但是在实际应用过程中，这个限制还没有给我们带来过麻烦。

4.6、跳过损坏的记录

每个worker进程都设置了信号处理函数捕获内存段异常（segmentation violation）和总线错误（bus error）。在执行Map或者Reduce操作之前，MapReduce库通过全局变量保存记录序号。如果用户程序触发了一个系统信号，消息处理函数将 用“最后一口气”通过UDP包向master发送处理的最后一条记录的序号。当master看到在处理某条特定记录不止失败一次时，master就标志着 条记录需要被跳过，并且在下次重新执行相关的Map或者Reduce任务的时候跳过这条记录。

4.7、本地执行

4.8、状态信息

另外，处于最顶层的状态页面显示了哪些worker失效了，以及他们失效的时候正在运行的Map和Reduce任务。这些信息对于调试用户代码中的bug很有帮助。

4.9、计数器

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

计数器机制对于MapReduce操作的完整性检查非常有用。比如，在某些MapReduce操作中，用户需要确保输出的key value pair精确的等于输入的key value pair，或者处理的German文档数量在处理的整个文档数量中属于合理范围。

5、性能

5.1、集群配置

在4GB内存里，大概有1-1.5G用于运行在集群上的其他任务。测试程序在周末下午开始执行，这时主机的CPU、磁盘和网络基本上处于空闲状态。

5.2、GREP

这个分布式的grep程序需要扫描大概10的10次方个由100个字节组成的记录，查找出现概率较小的3个字符的模式（这个模式在92337个记录中出现）。输入数据被拆分成大约64M的Block（M=15000），整个输出数据存放在一个文件中（R=1）。

图2显示了这个运算随时间的处理过程。其中Y轴表示输入数据的处理速度。处理速度随着参与MapReduce计算的机器数量的增加而增加，当 1764台worker参与计算的时，处理速度达到了30GB/s。当Map任务结束的时候，即在计算开始后80秒，输入的处理速度降到0。整个计算过程 从开始到结束一共花了大概150秒。这包括了大约一分钟的初始启动阶段。初始启动阶段消耗的时间包括了是把这个程序传送到各个worker机器上的时间、 等待GFS文件系统打开1000个输入文件集合的时间、获取相关的文件本地位置优化信息的时间。

5.3、排序

在这个benchmark测试中，我们使用的分区函数知道key的分区情况。通常对于排序程序来说，我们会增加一个预处理的MapReduce操作用于采样key值的分布情况，通过采样的数据来计算对最终排序处理的分区点。

还有一些值得注意的现象：输入数据的读取速度比排序速度和输出数据写入磁盘速度要高不少，这是因为我们的输入数据本地化优化策略起了作用 — 绝大部分数据都是从本地硬盘读取的，从而节省了网络带宽。排序速度比输出数据写入到磁盘的速度快，这是因为输出数据写了两份（我们使用了2路的GFS文件 系统，写入复制节点的原因是为了保证数据可靠性和可用性）。我们把输出数据写入到两个复制节点的原因是因为这是底层文件系统的保证数据可靠性和可用性的实 现机制。如果底层文件系统使用类似容错编码[14](erasure coding)的方式而不是复制的方式保证数据的可靠性和可用性，那么在输出数据写入磁盘的时候，就可以降低网络带宽的使用。

5.4、高效的backup任务

5.5、失效的机器

6、经验

在每个任务结束的时候，MapReduce库统计计算资源的使用状况。在表1，我们列出了2004年8月份MapReduce运行的任务所占用的相关资源。

6.1、大规模索引

7、相关工作

TACC[7]是一个用于简化构造高可用性网络服务的系统。和MapReduce一样，它也依靠重新执行机制来实现的容错处理。

8、结束语

我们也从MapReduce开发过程中学到了不少东西。首先，约束编程模式使得并行和分布式计算非常容易，也易于构造容错的计算环境；其次，网络带 宽是稀有资源。大量的系统优化是针对减少网络传输量为目的的：本地优化策略使大量的数据从本地磁盘读取，中间文件写入本地磁盘、并且只写一份中间文件也节 约了网络带宽；第三，多次执行相同的任务可以减少性能缓慢的机器带来的负面影响（alex注：即硬件配置的不平衡），同时解决了由于机器失效导致的数据丢失问题。

9、感谢

10、参考资料

附录A、单词频率统计

// 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;
}