Google的MapReduce介绍(转载)

转载自 ---- http://hi.baidu.com/leexeo/blog/item/cdd173f0979192b7a50f5209.html

 

前言

MapReduce 在当下绝对是IT技术界的一个热词，在网上，随便搜索一下就能够找到大量关于介绍MapReduce这个programming model的文章。所以，在本文中，对于MapReduce的原理和模型并不多做介绍，而更侧重于研究一个实现了MapReduce模型的系统的架构。主要的参考文献来自于Google的文献[1]，可以在网上找到这篇文章的中文版，不过还是更推荐读原版了。
注：本文中所说的MapReduce模型是指MapReduce的编程模式，而MapReduce系统是指实现了一个MapReduce模型的分布式计算系统。这这里讨论的是Google的MapReduce系统，基于Google Cluster 这样一个分布式环境。

接口

首先来看一看一个MapReduce系统对外的接口。
    Map函数：   map(String  key,  String  value) .....
    Reduce函数：    reduce(String  key,  Iterator  values) .....

了解MapReduce模型的人应该知道，Map函数的输出是一系列的Key/Value对（pair），这些Key/Value对是给 Reduce函数使用的。但是在这里可以发现，Map函数的输入参数也是Key和Value。但其实，输入的Key/Value和输出的Key /Value不是同一组东西。对于一个Map函数调用，输入的是一个Key/Value对，而输出呢，是一组Key/Value对。举例来说，如果要统计一篇文章中所有单词的数目，那么输入的Key可能就是行数，输入的Value就是该行的内容。对于一次Map调用，就是要统计出一行中各个单词出现的次数，所以输出的Key是单词，Value是这个单词的出现次数。
Map函数的输入之所以也用Key/Value的形式，可能是因为一般来说一个任务并不是用一次MapReduce就能够完成，而是需要用到多次MapReduce调用，下一个Map调用的输入很有可能就是上一个Reduce的输出结果（Key/Value对）。

系统流程

结合上面的图，这里描述了当一个用户程序执行MapReduce调用后，系统的流程：
1.首先，用户程序中的MapReduce Library会将输入的文件（就是要处理的文件）切分成大小为16M——64M之间的M份（就是图中的split 0，split 1，...），一般来说，这M份数据是放在了cluster的多台机器上。然后在整个cluster的机器上启动程序的多份拷贝，如图所示，（1）。

2. 其中的一份程序拷贝是master，其余的拷贝是workers。master会将任务分配给workers。任务包括了M个map的任务和R个 reduce的任务，master会找到空闲的worker然后将map或者reduce的任务分配给它。（M的值取决于input files被切分的块数；R值则是根据map中所能hash出的key的数量。）如图所示，（2）。

3. 当一个worker被分配到map任务，那么它就会处理M份中的一份split。InputData中的数据也是Key/Value形式的很多 pairs（这个在前面中提到了），Map任务的worker会将这些pairs一个一个传递给Map函数，Map函数产生的很多新的Key/Value 对会被缓存在内存中。

4. 周期性的，buffer中缓存的那些pairs会写入到本地disk中。如图所示，（3）（4）。由于有R个reduce任务，所以disk中的pair 会按照一定规则分为R个区，每个区的数据都特定的给某一个reduceWorker。在图中的每个Map Worker所对应的Intermediate files其实被分成了R个区（很有可能就是R个文件）。这些区的位置会被传递到master上，它会负责将这些位置交给reduce workers。可以看到，master在这里起到了一个“承上启下”的作用，Map函数和Reduce函数之间的连接是由master完成的。

5. Reduce Worker从master获取了这些区的位置，然后将这些Key/Value对从mapWorker处通过RPC读取。注意，Reduce Worker只读那些对应于自己的区的数据。如图所示，（5）。读取完数据后需要对于这些Key/Value对按照Key进行排序，如果需要，可能还采取外排的方式（external sort）。之所以需要排序是因为对于一个Key，会有很多组不同的pair，通过排序可以将它们聚合在一起。

6. reduceWorker迭代整个被排序后的Key/Value数据，将每个独一无二的key和它所对应的value序列（之所以是序列，是因为对于一个 key可能有很多个Key/Value，每个项中的value都不同，它们已经通过排序聚合在了一起）送入Reduce函数中。Reduce函数对于这样的输入产生的结果往往是0或者1个值（比如将所有Value相加，和就是最后的结果）。Reduce函数对于Reduce函数的结果会被添加到最终的 output file里（每次reduce Worker都会产生一个输出文件，最终一共有R个文件，因为一共有R个reduce的任务）。如图所示，（6）。

7. 当所有的map和reduce任务结束后，master会唤醒用户的程序。这时候，用户的MapReduce调用返回。

还需要提及的是，当整个流程顺利完成以后，最终得到的其实是R个文件。如果想要最终的结果，应该将这R个文件合并。但一般来说，都是将这R个文件作为另一个MapReduce任务的输入。

在这里，master起到了三个作用：
1. 它将任务分配给worker；
2. 它维护了所有worker的信息和所处的状态；
3. 它将Map任务结果的位置告诉给Reduce任务，起到了一个桥梁的作用。

容错
Google的这套系统的容错功能在我看来，非常的简单但却实用。举例来说，MapReduce系统通过master来监控其它的 worker，测试它们是否正常。测试的方法非常的简单，就是用ping。master会每隔一段时间就去ping一次worker，如果一定时间内没有返回，那么 master就认为这个机器挂了。
在这里可以和Google Chubby （关于更多Chubby的分析见我上一篇文章 ）做一个有意思对比：Chubby中，server与client之间也有类似的监控需求，但是它们是通过一个自定义的握手协议KeepAlive实现的。之所以没有采用ping的方式，是因为这个握手协议不仅仅提供了ping的功能，还传输了其它的一些交互信息。从这里看出，做分布式系统，并没有一定的规则，所有的设计都是根据实际的需求来决定的。

对于一个挂了的机器，根据它上面的任务和这个任务处的状态，分为4类：
1. Map任务，正在执行。很显然，该Map任务需要被分配到其它worker上重新执行。
2. Map任务，已经完成。对于这种情况，该Map任务也需要被分配到其它worker上重新执行。原因在于Map任务的结果是存放在本地disk上的，如果机器挂了，那么这些结果自然就访问不了了。
3. Reduce任务，正在执行。很显然，该任务需要重新执行。
4. Reduce任务，已经完成。这个情况下，该任务不需要重新执行了。因为Reduce任务的结果是放在了一个公共的分布式文件系统中。机器挂了对于该结果的读取不会有任何的影响。

容错的另一部分，是针对于master。google在这里采取的策略很有意思。如果master挂了，整个MapReduce任务就会被异常中止。用户程序可以捕获到这种异常并决定是否重新执行。也就是说，Google的MapReduce系统对于master的错误并没有“容错”，而是任其发生，将错误处理丢给了用户。

总结

除了上面两节以外，文献[1]中还包括了很多其它的细节和设计，在此就不啰嗦了。
在网上，对于MapReduce这个概念基本上是叫好声一片，毕竟对于很多从来没有做过分布式或者并行计算的人（比如我）来说，MapReduce模型给我们提供了一种崭新的解决问题的思路。但是在这里，我并不想替它唱赞歌（已经有太多人唱了，不差我一个），相反，我觉着另外一些负面的声音可能会更有意思一些。在文章[2]《MapReduce: A major step backwards》中，几位数据库方面的大牛对于MapReduce提出了非常严厉的批评，可以说是骂了个狗血淋头。有兴趣的可以看看，很有意思。文章 [3]是它的中文版。
虽然这篇文章的思想我不太苟同，但我也认为不应该将MapReduce过于神化。就连MapReduce的作者也没有回避，MapReduce这个思想其实并不新鲜，很早就在函数式编程中出现了。而且，MapReduce也不是万能灵药，实际上，它只适用于某一类任务，那种可以“分而治之”的任务，MapReduce只是分治思想的一种实现方式。
当然，总的来说MapReduce模型还是成功的。一方面是借了些 google的光，但更重要的，我认为还是在于它的简单。虽然MapReduce可能不是学术界最先进的技术，但是确实迄今为止最实用的一种分布式计算模型，原因就在于它的simple but effective。我想这多少能对其它的技术有些启示。