MapReduce编程模型

MapReduce是一个Google发明的编程模型，也是一个处理和生成超大规模数据集的算法模型的相关实现。用户首先创建一个Map函数处理一个基于<Key,Value>对的数据集合，输出的中间结果基于<k,v>对的数据集合，然后再创建一个Reduce函数用来合并所有的具有相同中间Key值的中间Value值。

MapReduce架构的程序可以实现在大量普通配置的设备上实现分布式计算。在Google的集群中，每天都有1000多个MapReduce程序在执行。

例如下面的例子：计算系一个大的文档集合中每个单词出现的次数，伪代码如下：（代码来自Google的论文）

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函数输出文档中的每个单词、以及这个词的出现次数。Reduce函数把Map函数产生的每一个特定的词的计数累加起来。

MapReduce的执行概括：

通过将Map调用的输入数据自动分割为M个数据片断的集合，Map调用被分布到多台机器上执行。输入的数据片断能够在不同的机器上并行处理。使用分区函数将Map调用产生的中间key值分成R个不同的分区（例如，hash(key)mod R），Reduce调用好也被分布到多台机器上执行。分区数量R和分区函数由用户来指定。

下图展示了MapReduce的执行流程：

1）用户程序首先调用MapReduce库将将输入文件分成M个数据片断，每个数据片断的大小一般从16MB到64MB。然后用户程序在集群中创建大量的程序副本。

2）程序副本包括一个Master程序，若干个Worker程序，master负责分配任务，M个map任务，和R个reduce任务。master将一个Map任务或Reduce任务分配给一个空闲的worker。

3）worker程序收到分配到的map任务后将读取相关的数据片断，解析出<k,v>对，然后传递给用户定义的map函数，由map函数生成中间的<k,v>对，并缓存在内存中。

4）缓存中的<k,v>对通过分区函数分成R个区域，然后周期性的写入到本地磁盘中。并且缓存中的<k,v>对在本地磁盘上的存储位置将被回传给master，由master再负责把这些存储位置传送给负责reduce任务的worker们。

5）但负责reduce的worker们收到master程序送来的数据存储位置信息后，使用RPC（远程程序调用）从负责map的woker们所在的主机上的磁盘里读取缓存数据。负责reduce的worker们得到中间数据后，对key进行排序后将具有相同key值的数据聚合在一起。

6）然后reduce worker将按key将中间value值传递给用户自定义的reduce函数。reduce函数的输出被追加到所属分区的输出文件。

7）当所有的map和reduce任务都完成后，master唤醒用户程序。这时，用户程序对MapReduce的调用将返回到用户代码中。

计数器

MapReduce库使用计数器统计不同事件的发生次数。例如，用户可能想统计已经处理了多少个单词，已经索引了多少篇German文章等。为了使用这个特性，用户在程序中创建一个命名的计数器对象，在map和reduce函数中相应的增加计数器的值。例如：

Counter* uppercase;
uppercase = getCounter("uppercase");

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

计数器的值周期性的从各个单独的worker机器上传递给master。master把执行成功的Map和Reduce任务的计数器值进行累加，但MapReduce操作完成后，返回给用户代码。计数器对于MapReduce操作的完整性检查非常有用。