Google
MapReduce阅读笔记
摘要：
MapReduce不是一个产品，它是一种基于分治思想，一种解决问题的思路。Google MapReduce是Google产出的一个编程模型，算法模型，当然Google也有其相关实现，提供给了用户相关函数接口：
（1）Map函数接口处理一个基于key/value（后简称kv）的成对（pair）数据集合，同时也输出基于kv的数据集合；
（2）Reduce函数接口用来合并Map输出的kv数据集合；
现实中有许多应用需求都能用这种模型处理，许多应用都能用这种方法解决。
MapReduce架构的程序能在大规模普通PC集群上实现并行处理，整个集群系统在程序运行时需要关心：
（1）如何分割数据；
（2）集群调度；
（3）集群错误处理；
（3）集群通信；
用户仅仅提供Map函数接口，Reduce函数接口即可，在MapReduce架构下，能让那些没有分布式计算处理程序开发经验的程序也有效的利用分布式集群的资源。
1.介绍
现实中有许多基于分治的应用需求：
（1）网页抓取；
（2）日志处理；
（3）索引倒排；
（4）查询请求汇总；
（5）…
单机的计算容易理解与完成，但在输入数据量巨大（TB级别）的情况下，如何能够在短时间内完成处理呢，只有将这些计算分布在成百上千的主机上，但此时，并行计算、数据分发、错误处理、集群通讯等等问题综合到一起，就成为了一个困难的问题。
为了解决这个问题，抽象出一个计算模型：该模型下，不必关心并行、容错、数据分布、负载均衡等细节，而只需提供Map和Reduce函数。
2.编程模型
MapReduce编程模型原理：
（1）Map函数接受一个kv输入数据集合，输出一个中间kv数据集合；
（2）Reduce函数接受Map输出的kv数据集合，再进行汇聚。
2.1例子
以统计大量文档中单词出现的个数为例，以下是一段伪代码，
Map(string $key_doc_name, string $value_doc_content) // 文件名=>文件内容
foreach(string $word in $value_doc_content) // 文件中的单词
array[$word] ++; // 单词计数增加
Reduce(string $word, iterator $value) // 单词=>计数链表
int $count = 0;
foreach(int $count in $value) // 计数合并
$count += $value;
emit($word, $count)
很多应用的例子都适合上述MapReduce模型，例如：
（1）分布式grep：Map输出匹配行，Reduce直接输出；
（2）URL访问频率计算：Map输出(URL, 1)，Reduce累加，产生(URL, count)；
2.2类型
用户定义的Map和Reduce函数相关类型：
map(k1, v1) => list (k2, v2)
reduce(k2, list(v2)) => list(v2)
3.实现
MapReduce模型有多种实现方式，一种是小型共享内存式，一种是基于NUMA架构的大型多处理器。
Google的实现基于：
（1）x86架构、Linux操作系统、双核处理器，4G内存；
（2）普通网络设备，100M或1000M带宽；
（3）成千上百台普通PC机器，故障是常态（GFS里也常提到这个，故障是常态）；
（4）存储为廉价的IDE硬盘；
（5）用户提交的作业（job）由系统统一调度：每个工作包含一系列任务（task），调度系统会完成任务的分发、分配；
3.1执行流程
通过将Map的输入数据集分割为M个Map片段集合，可将这些Map片段集合分配到多台机器上执行Map，以实现并行处理；
Map产生的结果数据集又被分配为R个数据分区，该项工作由 分区函数 完成，分区数量R和分区函数都由用户指定，如：
hash(key) mod R，下图展示了MapReduce实现中操作的全部流程：
图一：MapReduce执行流程
当用户调用MapReduce时，将发生以下一系列动作：
（1）用户调用MapReduce，将输入数据分成M个数据片段集合，然后在集群中创建大量程序副本（fork）；
（2）这些副本中有一个主程序（master），其他均为工作程序（worker），任务的分配由master完成，它将M个map任务和R个reduce任务分给不同的worker；
（3）被分配到map任务的worker，从数据片段集合中读取kv对，交给用户的Map函数处理，生成的结果kv对放在缓存中；
（4）缓存中的kv对通过分区函数，分成R个区域，周期性的写回本地磁盘，由master再把它们传给负责reduce的worker进行处理；
（5）负责reduce工作的worker从远程读取中间kv数据，对key进行排序，使得相同的key的数据聚合在一起（如果数据量太大，可外部排序）；
（6）相同key的中间数据，其value集合交给用户的reduce函数处理，处理结果写入对应分区的输出文件；
（7）所以map和reduce的worker都结束工作后，master唤醒用户程序，MapReduce调用返回，结果被输出到了R个文件中。
3.2master
master会存储一些元信息（GFS的master也用来存元信息），包括每个map和reduce的状态，是“空闲”、“工作”还是“完成”，它当然也保存各个worker机器的标识；
每个map完成后，master会知道R个中间kv数据集合的位置，并把这些数据推给reduce处理；
master就像一个管道，R个中间文件从这个管道从map传递给reduce。
3.3容错
（1）worker失效
master会周期性的ping每个worker，约定时间内仍未收到worker返回的信息，master将标记这个worker失效，这个任务将会分配给其他worker；
失效的worker的状态会重置为空闲，等待其他任务调度；
一个map任务如果先被worker A执行，失效后调度给worker B执行，master会将“重新执行”的命令通知给所有reduce的worker，数据将从worker B中读取；
（2）master失效
一个简单的方法是，将元数据写入磁盘，即加如检查点（checkpoint）。master任务失败了，由另一个master读取检查点，继续执行。
现实的实现是，只设置一个master进程，master失效就终止MapReduce计算，并通知用户，可以让其选择重新执行。
（3）原子性提交
在用户提供的map、reduce函数一定，且无出错的情况下，MapR的产出一定是一样的，使用“原子性提交”来保证这个特性。
每个map完成时，会产生R个私有临时文件，该任务完成的通知，以及R个文件的元信息会传递给master（如果该map任务有多个worker执行，后续的完成通知将忽略）；
每个reduce完成时，会产生1个最终的输出文件，其文件名唯一（如果该reduce任务有多个worker执行，GFS提供的文件唯一性，或者叫文件重命名原子性将保证数据只有一份）；
3.4中间数据的存储
中间数据的存储都由GFS（Google File System）管理，GFS保证每个文件按照64M一个block分隔，每个block的副本保存在多台机器上。
为了减少数据传输成本，map的任务调度会尽量放到副本机器上执行（实在不行，则调度到较近的机器，实施拷贝）；
这样是为了保证大部分数据都是从本地读取，减少网络带宽，提高运行效率。
3.5任务粒度
map分成M个片段执行，reduce分成了R个片段执行，理论上M和R比worker机器数量多得多，任务分配需要具备负载均衡能力；
worker机器的故障需要有快速恢复能力，故障机器上的任务又能快速分配到其他worker机器上去；
实际上，具体实现中M和R都有一定限制，master必须执行O(M+R)次调度，并保存O(M*R)个状态（这个很好理解，每个M要产出R份数据）；
进一步，R值通常由用户指定，M根据经验，一个任务大约处理64M的输入数据（本地数据存储优化最有效），通常M=200000，R=5000，worker机器数量=2000。
3.6备用任务
影响一个MapReduce总执行时间，最主要的因素往往是“落伍者”，即短板，常常被称作“长尾效应”：一台机器花了很长的时间才处理完最后几个map或reducer（原因也是多种多样的，例如硬盘坏了，数据读取速度奇慢无比等），导致总执行时间超时。
有一个“备用任务”机制解决“长尾效应”，master启用备用任务完成短板任务。
4.扩展与改进
4.1分区函数
用户通常会指定reduce任务输出文件数量R，map产出的中间数据集使用分区函数对数据进行再划分，之后再输入到后续任务执行。
默认的分区函数是哈希方法，hash(key) mod R，这种方法能够平衡负载。有时候用户有特殊的需求，例如希望每个主机（hostname）的URLs保持在同一个文件中。MapReduce库支持此类扩展，使用hash(hostname(url)) mod R分区函数就能满足上述需求。
4.2顺序保证
实现保证在给定分区中，kv数据的梳理是按照key增序处理的。
4.3合并函数（Combiner Function）
有时，map产生的中间key的重复数据比重很大，例如词频统计的应用，map将产生大量($word, 1)这样的kv数据，然后reduce把这些记录累加起来。
可以提供给用户一个合并函数，在map完成后，本地就做一次合并，生成($word, n)，这样reduce处理的量就会减少很多。
合并函数在每个map任务上都会执行一次，一般来说，合并函数与reduce函数是一样的，它们的区别是，合并函数执行本地数据合并，结果生成在临时文件中；reduce函数执行最终的合并，结果生产在最终的文件里。
4.4输入输出类型
文本行是一种最常用的格式：key是偏移量，value是一行的内容；
当然也能够实现从数据库里读取记录。
4.5忽略损坏的记录
某些时候，部分记录损坏了，MapReduce可以跳过这些记录，部分数据的损坏可能不影响全局“统计”结果。
4.6计数器
MapReduce库使用计数器统计不同事件发生次数，例如，用户想统计已经处理了多少个单词。
为了支持这个特性，用户可在程序中创建命名的计数器对象，在Map和Reduce函数中增加相应的计数器的值。
worker会周期性的把这些计数器的值汇报给master，由master负责这些值的累积（提供查看进度的可能性）。
4.7其他
（1）map和reduce过程中，根据用户逻辑，可能会产出一些中间文件临时保存数据，没有提供类似“两阶段提交”的机制保障这种情况的原子性；
（2）本地执行：分布式bug难于调试，提供了MapReduce库的本地实现版本，可以方便的gdb，差错等；
（3）状态信息：master使用嵌入式http服务器可以显示状态信息：执行进度、已完成任务、成立百分比等；
5.性能测试
5.1集群配置
（1）1800台机器；
（2）双核2GCPU；
（3）4G内存；
（4）160G硬盘；
（5）千兆网卡；
（6）双层树形交换网络，100-200G传输带宽。
5.2grep
5.3排序
5.4备份
…
结论是：Google MapReduce很牛逼。
n.结束语
为什么Google MapReduce能成功：
（1）Google MapReduce库封装了并行处理、容错处理、本地数据优化、负载均衡等技术难点细节，使得库易于使用；
（2）大量需求能够通过这种模型解决：检索、排序、挖掘、机器学习等；
（3）实现并部署了千台机器组成的MapReduce，Google用它解决了很多问题。
经验：
（1）约束编程模式使得并行与分布式计算变得容易，也易于构造容错计算环境；
（2）网络带宽是稀有资源：大量优化用在减少网络传输上；
（3）多次执行相同的任务能减少“长尾”带了的负面影响，同时解决了机器失效导致的数据丢失问题。