分布式系统——MapReduce：Simplified Data Processing on Large Clusters论文

1.简要介绍

MapReduce：Simplified Data Processing on Large Clusters最初发表在2004年，本次分享的是2008年的版本，内容较2004版本进行了精简和补充。
在建立MapReduce之前，Google工程师会实现数百种特定的、大规模数据的计算，如：网上爬取文档，计算派生的数据（如数据图结构计算）等等。大多数此类计算在概念上是直截了当的，但是实现起来要分门别类的实现。输入数据通常很大，并且计算必须分布在数百或数千台机器上，以便在合理的时间内完成。为了应对这种复杂性，计算机工程中经常采用在复杂性之上抽象出一层，这样就可以很好的面对复杂的场景。MapReduce就是提供了一种抽象，允许使用的人将计算表达出来，同时隐藏了复杂的实现。
文中提到，这种抽象受到Lisp语言中map和reduce，以及其他函数式编程语言的启发。文章作者认为，大多数相关计算都包括一个map操作，应用于输入中的每个record，以便计算一组intermediate key/value对，然后对所有共享相同key的value应用reduce操作。使用中，用户明确map和reduce函数的计算，运行系统自动在大规模集群机器上并行计算。文章认为程序员会发现MapReduce系统易于使用，从2004年到2008年，Google内部实现了超过一万个不同的MapReduce程序，平均每天在Google集群上执行十万个MapReduce作业，每天处理的数据超过二十PB，应用非常广泛。作者认为这项工作的最大贡献是一个简洁和高效的接口，可以在大规模商用PC集群（如GFS）上来进行并行化计算。

2.编程模型

计算任务中，输入为key/value对，输出也是key/value对。后面会介绍这些key/value对的异同。用户使用MapReduce可以通过两个函数：map和reduce。
由用户编写map，采用输入对，并产生一组intermediate key/value。MapReduce库将同一中间key，比如key L关联的所有中间value组合在一起，并将它们传递给reduce函数。
reduce函数也是由用户编写的，它接受中间key L，以及key L对应的value。它将这些value合并在一起以形成可能更小的value集合。通常，每次reduce调用只产生零个或一个输出值。中间值value通过迭代器提供给用户的reduce函数。
对于MapReduce接口的不同实现是可能的，正确的选择是基于需要的具体环境来考虑。论文发表之后，有几个不同的关于Map Reduce的开源实现：Hadoop，The Phoenix system等。
文章关于MapReduce的实现，是针对Google大规模采用的计算环境：通过交换式千兆以太网连接在一起的大型商用PC集群。在Google中，机器通常是运行在x86处理器上的Linux系统，每台机器有4-8GB内存。单个机器通常具有1 gigabit/second的网络带宽。计算集群包含数千台计算机，因此机器故障很常见。存储由直接连接到各个机器的廉价IDE磁盘提供。 GFS是一种内部开发的分布式文件系统，用于管理存储在这些磁盘上的数据。文件系统使用复制（通常为三份），在不可靠的硬件之上提供可用性和可靠性。用户将Job提交到调度系统。每个Job由一组task组成，并由调度系统映射到集群中的一组可用计算机。

3.MapReduce使用例子

考虑计算大量文档中每个单词出现次数的问题。即WordCount问题。用户将编写类似于以下伪代码的代码。

 Map（String key，String value）：//key：文件名字//value：文件内容For each word w in value:EmitIntermediate(w,”1”);     //引号内为数字一Reduce(String key,Iterator values)://key:一个词//value:count列表Int result = 0;For each v in values:Result += ParseInt(v);Emit(AsString(result));

map函数发出(emit)每个单词（word）加上相关的出现计数（在这个简单的例子中为1）。reduce函数将特定单词发出的所有计数汇总在一起。此外，用户可以编写代码，使用输入文件和输出文件的名称以及可选的调整参数，来填充mapreduce的规范对象（mapreduce specification object）。然后，用户调用MapReduce函数，将其传递给规范对象。用户的代码与MapReduce库（用C ++实现）链接在一起。
2004的MapReduce论文包含此示例的完整程序文本。
使用Google的MapReduce实现了超过一万个不同的程序，包括用于大规模图形处理，文本处理，数据挖掘，机器学习，统计机器翻译以及许多其他领域的算法。
Key/value对之间的关系：

 Map          （k1,v1）          ->list(k2,v2)Reduce        (k2,list(v2))     ->list(v2)

即map输入的键值对和reduce输出的值属性不必相同；中间值的key/value对属性相同。

4.执行概览

当用户调用MapReduce函数，如下过程将被执行（每一步对应于图中数字序号）：

用户程序调用MapReduce库，首先将输入文件拆分为M件，每件通常为16-64MB（可由用户通过可选参数控制）。然后，它会在一组计算机上启动该程序的许多副本。
该程序的副本之一-master-是特别的。其余的则称为worker。有M个map任务和R个Rreduce任务要分配。 master挑选闲置的worker并为每个worker分配一个map任务或reduce任务。
被分配了map任务的worker，读相应的输入split。worker从输入数据中解析key/value对，将其传入用户定义的map函数。intermediate key/value缓存在内存中。
周期性地，缓存对被写入本地磁盘，通过partitioning函数划分为R区域。位置信息将传递给master，进而在需要时候传递给reduce worker。
master将位置信息发送给reduce worker。当reduce worker读取其中间数据时，它会通过intermediate key对其进行排序，以便将所有出现的相同key组合在一起。如果中间数据量太大而无法容纳在内存中，则使用外部排序。
reduce worker遍历排好序的中间键值对，对每一个不同的key，将相应的key/value发送给reduce函数。reduce函数的输出附加到此reduce分区的最终输出文件。

5.本地性

在Google的计算环境中，网络带宽是一种相对稀缺的资源。利用输入数据存储在集群机器的本地磁盘上，从而可以节省网络带宽。GFS将每个文件划分为64MB块，并在不同的机器上存储每个块的几个副本（通常为3个副本）。 MapReduce中的master会考虑输入文件的位置信息，并尝试在包含相应输入数据副本的计算机上计划map任务。如果这个行不通，它会尝试在该任务的输入数据的副本附近安排map任务（例如，位于同一交换机的机器上）。运行大型MapReduce操作时，对于集群中的大部分worker，大多数输入数据在本地读取，不会消耗网络带宽。

6.任务粒度

选择M和R的大小。首先M和R的数量应大于worker数量很多，这样可以提高负载均衡，并且发生故障可以尽快恢复。其次，在实践中，倾向于选择M使得每一个任务为16MB或者64MB，R则为worker的数倍。文中提到经常采用的worker数量为2000，M为200000，R为5000。
延长MapReduce操作所花费的总时间的常见原因之一是落后的机器（straggler），即，在计算中完成最后几个map或reduce任务之一需要非常长时间的机器。Stragglers可能出于各种原因而出现。例如，具有坏磁盘的计算机可能会遇到频繁的可纠正错误，从而将其读取性能从30MB / s降低到1MB / s。当MapReduce操作接近完成时，master会调度剩余正在进行的任务的备份执行。无论主要执行还是备份执行完成，任务都会标记为已完成。它通常会将操作使用的计算资源增加不超过百分之几，但这种操作大大减少了完成大型MapReduce操作的时间。例如，当禁用备份任务机制时，排序程序需要多花44％的时间才能完成。

7.实际集群例子

在大型机器集群上运行的两个计算来测量MapReduce的性能。一个计算搜索大约1TB的数据以寻找特定模式。另一个计算对大约1TB的数据进行排序。
集群配置：程序在由大约1800台机器组成的集群上执行。每台机器都有两个2GHz Intel Xeon处理器，支持超线程，4GB内存，两个160GB 的IDE磁盘和一个千兆以太网链路。在4GB内存中，保留了大约1-1.5GB的内存为其他任务使用。

以计算一为例，计算一的程序称为grep。grep程序扫描1TB的记录，搜索相对罕见的三字符模式（模式出现在92,337条记录中）。输入分为大约64MB（M = 15000），整个输出放在一个文件中（R = 1）。图中显示了计算随时间的进度。Y轴显示扫描输入数据的速率。随着为此MapReduce计算分配更多机器，并且在分配了1764名worker时峰值超过30GB / s。当map任务完成时，速率开始下降并在计算中大约80秒时达到零。整个计算从开始到结束大约需要150秒。还包括大约一分钟的启动开销。

8.总结

1.MapReduce优点：
提供了抽象接口，分布式过程完全隐藏，使得程序员容易使用；
使得大规模处理数据变得可能；
自动的负载均衡；
自动的容错性；
2.MapReduce缺点：
极其严格的数据流；
很多常见的操作也必须手写代码；
程序内部实现隐藏，优化比较困难。

下一篇为灵活的数据流框架，PIG LATIN
下一篇：Pig Latin：A Not-So-Foreign Language For Data Processing