MapReduce阅读笔记

来自论文 MapReduce: Simplified Data Processing on Large Clusters

介绍

MapReduce使用了map函数来将key/value pair生成中间的key/value pair，然后用reduce函数将所有中间结果的value通过相同的key聚合起来

Programming Model

模型主要有两个函数，map和reduce

Map

Map函数是由用户来写的，将输入的数据处理为一系列中间结果的key/value pairs。MapReduce将所有相同key的value聚合在一起，传递给Reduce函数

Reduce

Reduce函数也是由用户来编写的，接受key $I$和对应的values，它将这些values合并成尽可能小的value的集合。

例子

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函数将每个单词，和1发送出去
reduce函数将map发送的数据中相同单词的1全部加起来，就是最后该单词在文档中出现的数量

Types

用户提供的map和reduce函数有相关联的类型

map(k1, v1)             ->list(k2, v2)
reduce(k2, list(v2))    ->list(v2)

输入的keys，values和输出的keys和values来自不同的域。但是中间结果的keys, values和输出的keys，values来自同一个域(这一句话不是太能理解…)

实现

• 首先MapReduce将用户输入的文件分割$M$个份(通常为16MB到64MB)，然后在集群中启动多个程序
• 其中一个程序为Master，剩下的为wokers，听从master的调度来辅助工作。其中有$M$个map任务和$R$个reduce任务。master会选取空闲的wokers分配给它map/reduce任务
• 一个被分配到map的worker会读取被分割的输入，解析出key/value pairs，然后将这些pairs输入到Map function(用户定义)，Map function会输出中间结果的key/value pairs并存放到内存中
• 每个放到内存中的pairs会被周期性的写入到磁盘中，并通过partitioning function分区成$R$个区域。这些pairs在硬盘存放的位置会告知master，然后master会将位置再传递给reduce函数
• 当一个reduece worker被master通知中间pairs存放在磁盘的位置时，会读取一个区域中的数据，当所有的中间结果的数据读取到后，它会将这些数据以key来排成有序，这时候，有相同key的value就会合在一起
• reduce woker会遍历所有已排好序的中间结果，然后对于每一个key，以及对应的values的set，都会送到用户定义的Reduce函数中，Reduce函数将最后的结果放到这个reduce区域的最后输出的文件中。
• 当所有的map任务和reduce任务都结束的时候，master被唤醒，将最后结果返回给用户
• MapReduce全部完成的时候，会产生$R$个输出文件，每个对应一个reduce产生一个，通常用户不用去将它们合并，因为它们大部分时间会被传送到下一个MapReduce中，或者下一个分布式的处理函数中。

Master Data Structures

master需要保存多个数据结构，对于每个map task和reduce task，master需要保存它们的状态，以及标识信息
master还需要保存R个中间key/value pairs存储的位置，以及大小。并且在每个map task完成后更新它。

容错

Woker Failure

master会周期性的ping每个woker，如果有woker在指定的时间内未回复，那么master就会标记这个woker失败了。当任意一个map task完成后，其对应的woker就会设置为idle状态，因此它可以被安排到其他任务上。同样当有一个map或者reduce失败的时候，其woker也会设置为idle等待重新调度。
当map task失败的时候，需要重新执行该task，因为map task将临时的数据存储在了本地磁盘中，无法在全局中获取。
reduce task失败的时候不需要重新执行，因为它们的数据存放在了全局的文件系统。
当一个map task先由woker A执行，当A失败后，再由B执行，此时会通知所有的reduce task，告知它们以后读数据从B中读取。

Masker Failuer

可以给master设立检查点，当master是失败的时候，可以让新的master从上一个检查点开始。

Semantics in the Presence of Failures

当用户提供的map和reduce操作输入值确定时，输出值也确定，那么分布式实现与不出错的顺序整体实现输出应当是一致的。
MapReduce通过map和reduce task使用atomic commites来实现上述特性。每个运行中的task都会将它们的输出写到私有的临时文件中。一个reduce task会产生一个这样的文件，一个map会产生R个这样的文件（每个文件对应一个reduce）。当一个map task完成后，对应的woker会想master发送完成信号以及R个临时文件的信息。当master重复接收到一个已经完成的map task的信息后，就会忽略这个重复的信息。如果不重复，那么master就会记录这R个文件的信息。
当一个reduce task完成后，reduce woker就会原子化的重命名它临时输出文件从而作为最后的输出文件。
//TODO: 较弱的失效处理没看懂…

Locality

论文中的输入文件是由GFS管理的，GFS将大文件分割成64MB的块，然后复制几份（一般是三份）放到不同的机器中，MapReduce master就会调度这些有复制文件的机器执行map task。如果调度失败，那么master就会调度里数据比较近的机器执行map

Task Granularity

// TODO