最近看 mit-6.824，每个 lecture 会做一次笔记，每个笔记基本都分为，

Readings：课前阅读的论文
Lecture：讲义
Lab：实验

Readings

MapReduce: Simplified Data Processing on Large Clusters论文

实现

map task运行时，

定期将生成的kv pair写入本地磁盘，并被parition函数分为R个分区。
这些在磁盘上的pair的位置会被返回给master，master进而把这些位置发给reduce task。

reduce task运行时，

使用RPC读取map task缓存到磁盘的数据。
当reduce worker读取到所有数据时，按临时key对数据排序。因为多个intermediate key可能会由同一个Reducer处理，因此需要排序使得相同的key在一起。
reduce worker遍历已排序的中间数据，将key和中间value传给用户定义的reduce函数。

master数据结构

对于每个map task和reduce task，master保存了任务状态（idle, in-progress, or completed），以及每个worker机器的身份。
对于每个已完成的map task，master保存了R个分区的中间数据文件的位置和大小，每当一个map task完成，中间数据文件的位置和大小就被更新。这些信息以增量的形式，发送给有正在运行reduce task任务的机器。

容错

由于长时间未响应master的ping，worker故障，

正在运行的map task或reduce task重置为idle->重跑。
已完成的map task重置为idle->重跑，因为map task生成的中间数据是本地存储的。
已完成的reduce task无需重跑，因为输出已经保存到GFS。

当一个map task先被worker A执行，接着又被B执行（worker A失败），这次重跑会被通知到所有执行reduce task的机器。任何还没有从A读取数据的reduce task，都转而向B读取。

master故障，

定期保存master数据结构的checkpoint。

semantics in the presence of failures，

当用户提供的 map 和 reduce 操作的执行结果是确定的，那么分布式的实现也会产生相同的执行结果，这个结果与整个程序顺序执行产生的结果一致。这依赖于，map task 和 reduce task 输出的原子提交。每个正在执行的任务都会把输出写到临时文件。当某个map task完成时，worker向master发送包含有R个临时文件列表的消息。如果master已经收到了这个任务完成的消息，那么本条就会被忽略。而对于 reduce task，输出到GFS的时候，GFS保证了原子性。
当 map 和 reduce 的执行结果是不确定的，那么最终结果也是不确定的，每个 reduce 任务的执行结果都不同的不确定的串行执行的结果。

局部性

GFS 将文件分为 64MB 的块，并在不同机器上冗余存储。Master 可以利用文件的位置信息，在包含这个副本的机器上调度 map 任务。如果无法进行上述调度，那么就近调度（例如：同一个交换机下的 worker）。

任务粒度

MapReduce 把 map 任务和 reduce 任务分别分为 M 份和 R 份，理想情况下，M 和 R 应该远大于机器数目，这样有利于提高动态负载均衡和加速（当某个 worker 挂了情况下的）错误恢复：这个挂了的 worker 所执行的任务可以被重新调度到其他 worker 上进行。

实现时，master 需要进行 O(M + R) 次调度，需要 O(M * R) 状态跟踪 map 产生的临时文件位置（每个 map 产生 R 个临时文件），每个 map/reduce pair 需要一个 byte。

通常 R 的数量是由用户指定的，实际应用中对 M 的划分是要保证一个分片的数据量大小大约是 16-64M，R 的期望是一个比较小的数。

任务备份

一个导致任务执行总时间大大延长的重要原因是存在“拖后腿的任务”，这些执行缓慢的任务可能由多种原因导致。一个通用的解决方案是，当一个 MapReduce 任务将要执行完成时，master 调度一个备份执行。当主任务（原始的）或备份执行的其中之一完成时，这个任务被标记完成。这个方法在消耗少许额外资源的情况下，大大减少了总执行时间。

改进

提升效率的方法。

Partitioning 函数

用户指定 reducer 的数目为 R，map 的中间结果按照 partitioning 函数对临时 key 的计算结果分成了 R 个部分。一般来说，默认的 partitioning 函数是足够的，但也可以自己提供 partitioning 函数来实现特定的 partition 目标。

保证顺序

每个 partition 的 kv 是按照临时 key 升序的顺序处理的。这保证了，

每个 reduce 产生的结果是有序的。
利于按 key 的随机访问。
用户觉得有序的结果方便。

Combiner 函数

当，

每个 map 任务会产生大量重复的临时 key，
用户提供的 reduce 函数是可交换和可结合的，

可以在 map 任务结束以后，使用 combiner 对 map 任务的结果进行部分合并，减少网络传输的开销。

输入输出类型

MapReduce 提供了读写多种文件格式的支持，用户也可以通过实现相应接口来自定义读写其他格式文件。

副作用

MapReduce 允许 map 或 reduce 生成辅助的输出文件，其原子性依赖于应用的实现。

跳过 Bad Records

某些特定的数据会让含有 bug 的 map 或 reduce 函数崩溃，有时无法修复这些 bug，且忽略这些数据是可接受的，那么可以使用 MapReduce 提供的机制来忽略。

每个 worker 都会监控 segmentation violations 和 bus errors，如果崩溃，发送记录 -> master，如果 master 发现某条记录失败次数大于 1，那么下次执行时就会跳过这条记录。

本地执行

分布式环境下，debug 一个 map 或 reduce 函数是困难的。MapReduce 库允许在本地执行任务，便于调试。

状态信息

master 会运行一个内部的 HTTP 服务器，展示状态信息。

计数器

MapReduce 库提供了一个计数器来统计事件发生的次数。worker 响应 master 的 ping 时，计数器的值就被放在 pong 中，发送到 master。master 聚合计数器的值，并在 map 和 reduce 任务结束后返回给用户。当聚合时，重复执行任务的计数器只会被统计一次。有的计数器值是由 MapReduce 库直接维护的。

Lectures

Distributed System

分布式系统提供了app使用的基础设施，通过抽象隐藏了实现的细节，这些抽象包括：

存储（storage）
通信（communication）
计算（computation）

在讨论分布式系统时，下面几个话题会时常出现：

实现 RPC，线程，并发控制。
性能
- 理想情况：可扩展的吞吐量。N倍的服务器数量->（通过并行的CPU，磁盘，网络实现）N倍的吞吐量。因此为了处理更多的负载，只需要添加更多的机器。
- 可扩展性随着N增加而变得困难：
  - 负载不均衡（Load im-blance），集群里有慢的机器（stragglers）。
  - 不可并行的代码：初始化，交互
  - 共享资源的瓶颈：网络
容错
- 大集群，复杂的网络->总会有出问题的地方
- 希望能够隐藏这些错误，使得错误对app不可见
  - 可用性（Availability）：在出错的情况下，app能继续使用数据。
  - 持久性（Durability）：修复错误后，app能够继续正常工作。
- big idea：多个服务器。如果一个server crash，client能继续使用其他的。
一致性（consistency）
- 通用目的的架构需要有良好定义的行为。
  - 例如：get(k)应该返回最近的put(k, v)。
- 实现良好定义的行为是困难的。
  - 难以保证多个服务器一致。
  - 在含有多个操作的更新中，客户端执行到一半可能crash了。
  - 服务器在尴尬的时刻崩溃，例如：执行了操作但没有返回结果。
  - 由于网络问题，服务器看起来好像挂了。
- 一致性和性能是冲突的。
  - 实现一致性需要通信。
  - “强一致性”常常导致系统性能低下。
  - 高性能通常会对app造成“弱一致性”。
- 开发者在这个范围内追求了很多设计点（People have pursued many design points in this spectrum）。

MapReduce

MapReduce概述

上下文：对海量数据进行多个小时的运算。
总目标：普通程序员能够在保证合理的效率的情况下，轻松的将数据处理切分到多个服务器。
需要定义Map和Reduce函数，非并发的代码，且常常是很简单的。
MR运行在集群上，处理海量数据，隐藏分布式的细节。

1
2
3
4
5
6
7
8
input is divided into M files
Input1 -> Map -> a,1 b,1 c,1
Input2 -> Map ->     b,1
Input3 -> Map -> a,1     c,1
                  |   |   |
                  |   |   -> Reduce -> c,2
                  |   -----> Reduce -> b,2
                  ---------> Reduce -> a,2

MR对每个输入文件调用Map()，生成<k2, v2>（中间数据）的集合
MR收集每个k2对应的所有v2，并输入给Reduce调用
最后从Reduce（）输出<k2, v3>的集合

MapReduce隐藏了很多痛苦的细节

在服务器上启动s/w
跟踪已结束的任务
数据移动
错误恢复

MapReduce有很好的可扩展性

N台机器->N倍吞吐量。假设M和R都>=R（大量的输入文件和map输出的key）。由于Map()之间无交互，可以并行执行，Reduce()也是，都能够并行执行，因此可以通过添加机器来增加吞吐量。

性能受限的潜在因素

网络。在Map->Reduce的shuffle时，所有数据都需要通过网络传输，因此减少通过网络来移动的数据是关键。

设计细节：减少低速网络的影响

Map从本地磁盘读入GFS上的数据副本（replica），而不是从网络。
中间数据仅仅在网络中传输一次。Map worker将数据写入本地磁盘，而不是GFS。
中间数据被partition到包含很多key的文件。大网络中的传输更加高效。

如何实现良好的load balance？

load balance对于可扩展性很重要，N-1个server都等待1个server结束是不好的。但是有的任务就是会比其他花费更长的时间。

解决方案：任务数比worker数更多。 master为已完成当前任务的worker分配新的任务。一般来说，就不会有大任务占据主要的计算时间。那么快的server会比慢的server完成更多的任务，最终同时完成所有任务。

如何实现容错？

如果某个server在执行MR任务期间crash怎么办？不是重启整个job，MR只重新运行失败的Map()和Reduce()。这两个操作必须是纯函数：

不保留调用之间的状态。
不读取和写入除了MR输入/输入的文件。
任务之间没有隐藏的通信。

因此重新执行会生成相同的结果。纯函数的要求是MR相比于其他并发编程模型的主要局限，当它也是MR简洁的关键。

crash recovery的细节

Map worker crashes：
- master观察到worker再也不响应ping。
- crash的worker的中间Map输出丢失，但这个数据有可能每个Reduce任务都需要。
- master根据GFS上输入数据的其他副本来分配任务，并重新执行。
- 某些Reduce worker可能已经读取了crash的worker生成的中间数据。此时就需要依赖于Map()的纯函数特性和确定性。
- 如果Reduce获取到了所有的中间数据，那么master就不需要重新运行Map。然后接下来的一个Reduce会crash，进而导致强制重跑失败的Map。
Reduce worker crashes：
- 已结束的任务不受影响，数据以冗余的形式存储在GFS。
- master重新执行其他worker上未完成的任务。
Reduce worker在写入输出数据期间crash：
- GFS的rename是atomic的，在写入完成前数据不可见。因此master可以安全的在其他地方重跑Reduce任务。

其他错误和问题

master给两个worker分配了相同的Map()任务怎么办？原因可能是master错误的认为worker挂了。master只会将其中一个告诉给Reduce worker。
master给两个worker分配了相同的Reduce()任务怎么办？两个都会尝试在GFS中写入相同的文件。GFS rename的atomic性质避免了结果是两者的混合，只有一个完整的文件可见。而Reduce()的纯函数特性使得两次输出的文件是一样的。
如果某个worker很慢怎么办？原因可能是硬件问题。master重新运行最后几个任务。
如果worker由于h/w或s/w问题计算出了错误的输出怎么办？ ╮(╯_╰)╭，MR假设CPU和software是“fail-stop”的。
如果master crash了怎么办？
- 从check-point恢复。
- 放弃执行任务。
- 使用多个master，一个可用，剩余standby。

什么样的app不适用于MapReduce？

不是所有app都适用于map/shuffle/reduce模式。
小数据量，因为开销很高。例如：网站的后端。
对海量数据的小更新。例如：为一个大索引添加小文件。
随机读写，因为Map和Reduce都不能选择输入数据。
多次shuffle，例如：page-rank。可以使用多个MR来实现，但是不高效。
更灵活的系统可以实现上述目标，但会导致更复杂的模型。

结论

MapReduce让集群计算受欢迎。

不是最高效和灵活的。
可扩展性良好。
易于编程，错误和数据移动被隐藏了。

这些在实践中是很好的权衡。

Lab 1: MapReduce

实验给出了框架代码，需要完成关键函数。
MapReduce的分布式实现除了需要关注task，还需要考虑存储，分布式存储不是这里的重点，因此实验在一台机器上运行worker thread，使用系统的文件系统模拟分布式存储。
实验要求实现两种模式的MR，顺序执行所有task，以及分别并行执行map和reduce task。实验给出了的框架代码使用了go的channel来实现并行执行task。而对于顺序执行task的实现，实际上是分别把所有map和reduce task做了一次封装，得到“一个map task”和“一个reduce task”，分别执行封装好的“map task”和“reduce task”，其中每个实际的map和reduce都是顺序执行的。
lab的代码在github.com/chaomai/mit-6.824

6.824 - Spring 2017 Lecture 1 Introduction