MIT6.824|Introduction

引言

课程关注的内容

课程主要关注两个部分的内容：

• 性能
• 容错 Fault Tolerance

分布式系统的抽象

这门课程讲述的都是基础架构的东西，我们针对基础架构的研究一般是包含：

• 存储（分布式存储服务）
• 通信（一般都是用现成的通信模型）
• 计算（MapReduce 分布式计算模型）

其中分布式系统的一个核心抽象目标就是：对外提供服务接口并且极大程度隐藏分布式的特性，简单来说就是：

从应用程序的角度来看，整个系统是一个非分布式的系统，就像一个文件系统或者一个大家知道如何编程的普通系统，并且有一个非常简单的模型语句。但是实际上又是一个有极高的性能和容错性的分布式系统

在学习这门课程的过程中我们会逐步体会到研究者在这一个抽象目标上所做出的努力

构建分布式应用程序的常用工具

• RPC：掩盖了我们是基于不可靠的网络进行通信的这一个事实
• 线程：实际上是提供给开发者操作多核心CPU的一个手段
• Concurrency Control：分布式领域中常常需要解决的问题

上述这些实现思想会在课程中出现，我们也会在许多论文中看到，同时也是我们在课程的实验中需要不断去关注和处理的

Scalability

基本概念

这里说的可扩展性和我们日常讨论的软件可扩展性还是有一些细微区别的

并不是我们说的代码可扩展易于维护或者说是SPI这种组件化的开发方式便于扩展功能

而是说我们从一台计算机系统扩展到十台分布式计算机系统后，可以获得原先十倍的工作量/十分之一的处理相同任务的时间，也就是十倍的系统吞吐量，这样的设计角度如果可以实现的话将会是十分强大的，因为我们只需要花钱来购买一定数量的计算机，相比较于花钱来聘请程序员进行软件的重构以优化性能而言，这种方式最为简单粗暴以及强大。

但是实际上在现实应用中这种情况也需要一定程度上的系统设计来确保性能的扩展提升。

只要单台web服务器没有给数据库带来太多的压力，你可以在出现问题前添加很多web服务器，但是这种可扩展性并不是无限的。很可能在某个时间点你有了10台，20台，甚至100台web服务器，它们都在和同一个数据库通信。现在，数据库突然成为了瓶颈，并且增加更多的web服务器都无济于事了。所以很少有可以通过无限增加计算机来获取完整的可扩展性的场景。因为在某个临界点，你在系统中添加计算机的位置将不再是瓶颈了。在我们的例子中，如果你有了很多的web服务器，那么瓶颈就会转移到了别的地方，这里是从web服务器移到了数据库。

总结：在扩展性方面，我们希望能够实现随随意增加分布式系统中计算机的数目来实现等倍数的系统吞吐量，但是在实际应用中往往有多种因素的干扰（如上面的图例）因此需要进行合理的架构设计来确保扩容对于系统性能的提升

Fault Tolerance

容错性主要的目标主要有以下，实现的难度分别递减：

• 可用性：系统某节点宕机但是还有副本继续提供服务
• 可恢复性：系统节点都宕机了，在恢复的期间内不能接受请求，但是在恢复之后仍能继续工作，并且在系统恢复前后数据仍然保持正确性

因此可以说 可用性的容错性>可恢复性

并且，一个好的具备可用性的系统，也需要具备可恢复性，因为可用的系统仅仅是在一定的故障范围内才可用，如果故障太多，可用系统也会停止工作，停止一切响应。但是当足够的故障被修复之后，系统还是需要能继续工作

实现容错性的工具

• 采用非易失性存储：但是要避免大量读写场景，因为硬盘的写入还需要考虑转动机械臂等因素，频繁写入对性能有影响
• 主从复制：引入一个副本实现容错，但是副本之间的数据同步（副本管理），drift out of sync 也是一个很棘手的问题（Lab2就需要实现一个主从复制的分布式系统模型来实现容错性）

Consistency

以一个最简单的 KV 存储为例

假设服务器有两个副本，那么他们都有一个key-value表单，两个表单中key 1对应的值都是20

现在某个客户端发送了一个put请求，并希望将key 1改成值21。这里或许是KV服务里面的一个计数器。这个put请求发送给了第一台服务器

之后客户端的同一个请求会发送给第二台服务器，因为相同的put请求需要发送给两个副本，这样这两个副本才能保持同步。但是就在客户端准备给第二台服务器发送相同请求时，这个客户端故障了，可能是电源故障或者操作系统的bug之类的

这种情况不是我们想看到的，我们发送了一个put请求，更新了一个副本的值是21，但是另一个副本的值仍然是20

因此针对 KV 的 put 和 get 操作，很显然我们需要定义一些特殊的规则来实现分布式系统中的一致性，这种被定义规则的一致性，我们分为强一致性和弱一致性

强一致性和弱一致性

强一致性

需要牺牲性能来实现，因为强一致性的实现需要系统中节点进行频繁的通信

以上述的 KV 存储为例，强一致性的实现规则可以是：每次的 GET 读取操作读取最近一次的 PUT 操作的数据

具体的实现可以是：客户端轮询系统中所有的副本，比对数据的 PUT 修改时间，选择最近的一个数据作为最终的结果，这样的场景需要大量的通信，对性能造成巨大的开销

因此就引出了弱一致性

弱一致性

弱一致性不保证不会读到旧数据，在上述的 KV 存储例子中，就是有可能会读取到旧的那个值为 20 的数据

在构建距离较远，系统一致性需求不强的场景下可以考虑采用弱一致性的实现

尽可能避免使用强一致性的设计 & 弱一致性也不差

在上面我们也提过了，强一致性的设计可能会影响系统的性能

因为分布式系统中的副本往往都不可能部署在一个机架甚至是一个机房内，因为如果实在同一个数据中心中，往往可能一挂就是都挂了，因此我们的数据副本往往会部署在多个距离很远的数据中心，随便一个通信也是超远距离的数据包传输，最少也需要几十毫秒的响应时间，现在的处理器每秒可以执行数十亿条指令，等待几十毫秒会大大影响系统的处理速度

此外，在学术界以及实际应用中，有大量关于构建弱一致性保证的研究，弱一致性并不代表不一致，我们可以增加许多的约束和规则来实现弱一致性的数据保证

MapReduce

分布式计算任务框架需求

最早的时候其实是 Google 提出的一个用于解决自己建立全网网页索引需求的一个实现方案

由于全网的网页索引太多了是TB级别的海量数据，如果单机处理则需要很长的运行时间

谷歌希望能够将这些构建索引的计算任务分配到海量的计算机上进行操作，虽然当时他们完全可以编写专用的分布式系统软件来实现这一点，但是谷歌更希望能够构建出一个更为通用的分布式应用框架，让使用者无需关心具体的分布式底层通信等逻辑，只需要提供 map 和 reduce 函数，就能够将自己的分布式计算需求分配到多个计算机系统上，能够进行任意的数据分析，例如排序，网络索引器，链接分析器以及任何的运算

使用 MapReduce 框架的工程师 只需要实现应用程序的核心，就能将应用程序运行在数千台计算机上，而不用考虑如何将运算工作分发到数千台计算机，如何组织这些计算机，如何移动数据，如何处理故障等等这些细节

MapReduce实现思路

MapReduce 分布式任务处理框架的核心实现思路就是，给用户提供编写 Map 和 Reduce 函数的接口

Google MapReduce 所执行的分布式计算会以一组键值对作为输入，输出另一组键值对，用户则通过编写 Map 函数和 Reduce 函数来指定所要进行的计算

这也是函数式编程的一种应用实践

MapReduce工作流程

MapReduce 的核心包含几个概念：

• Map Function
• Intermidiate
• Reduce Function

我们以最简单的 Word Cound 为例来解释这些概念

假设我们海量的文档被抽象分为三个文档，他们分别包含内容：

• ab
• b
• ac

Map 函数会以文件作为输入，类似的三个文档经过 Map 函数后的中间值也就是论文中说到的 Intermidiate Output 如下图

在之后，Map输出的所有中间变量的 key 会被作为唯一标识进行 收集，进而作为一个 list 传入 Reduce 函数中作为 Reduce 函数的输入

也就是说 Map 产生的中间变量中有几个不同的 key ，就会进行几次的 Reduce 操作

（需要注意的是这里的收集应当是由框架完成的，在分布式的网络环境中进行收集）

为了便于理解，下面我们讨论 Map 和 Reduce 函数的伪代码实现

Map函数实现

Map 函数的定义在上面的例子中第一个文档，Map的入参是：Map(file1name,”ab”)，而Map的输出则是 (a,1) (b,1)

key 为文件名，当然这个我们并不关心

value 为文件内容值

我们关心的实现可以用如下代码概括

Map(k,v):
     # split into words
     for each word w:
          emit(w,"1")

Reduce函数实现

Reduce 函数的入参是中间量中某一个 key（也就是字母）对应的所有实例（就算只出现了一次也要被统计）

在上面的例子中，Reduce 的入参为： Reduce("a",[1,1]) Reduce("b",[1,1]) Reduce("c",[1])

在这个简单的 WordCount 例子中，我们假定所有输入的内容字幕出现都不重复，因此每一个 Map key 对应的 value 都只会是1

因此在 Reduce 函数中只需要计算入参的 list 的长度即可快速得出答案了，因为不重复字母的情况下默认每一个 list 元素对应的内容都是1

因此 Reduce 的实现思路大致如下：

Reduce(k,v):
     emit(len(v))

Master & Worker 流程细节

Map批处理

在论文中提到了 MR 框架中的 Matser 和 Worker 节点，其中 Master 节点知道客户端请求中的总文件数量

在整个 Map Reduce 的过程中，Master 节点根据文件的输入总数，分配任务到 worker 节点

如果有1000台服务器，就会有1000个 worker 节点，之后 Master 节点将 Map 函数分发到不同的 worker。所以，它会向 worker 服务器发送一条消息说，请对这个输入文件执行 Map 函数吧，因此就会进行 1000 次的 Map 函数调用。

worker进程还需要实现emit，每次Map函数调用emit，worker进程就会将数据写入到本地磁盘的文件中。所以，Map函数中调用emit的效果是在worker的本地磁盘上创建文件，这些文件包含了当前worker的Map函数生成的所有的key和value，也就是中间数据

中间量分发

在进行Reduce过程之前，每一个worker节点会接收到 Master 节点的一个开始Reduce数据收集的指令

每一个 workder 节点都会向MapReduce集群中剩下的所有节点发消息，请求获取所有 key = xxx 的中间数据，并发送给这个worker节点

数据存储&网络吞吐

在前面的 Map 过程中我们提到了 worker 节点会将数据存储在本地磁盘中，但是对于海量的数据，我们更希望借助一个网络文件系统来便于数据的收发。

论文中提到的实现方式是 GFS(Google File System) 网络共享存储服务

GFS会运行在每一个 worker 节点的物理服务器上，并且针对需要进行 MapReduce 处理的大文件，已经在输入阶段被 GFS 平均分配到不同的 worker 节点上了，以一块一块的方式存储

因此对于 1000 台服务器，就会有 1000 个 worker 节点，总输入的海量数据就会被分为 1/1000 在每一个 worker 节点上进行处理

但是这样也就存在一定的问题，那就是海量的网络开销

假设所有的输入都被分配到了 GFS 节点上，那么一开始 worker 节点为了获取输入，需要通过集群间的网络（一般都不会在内部网络）来向 GFS 请求分块数据，读取到本地进行 Map 处理。最坏的情况，假设 Map 任务和数据块都对不上，是完全 “混乱的” 。如果总输入是 10TB 的数据，那么这一次的 MapReduce 操作就会需要 10TB 的网络开销，这是不能被接受的

Google 为了解决这一个问题，设置集群中所有节点都运行 GFS 和 MapReduce任务。 Matser 节点在分配 Map 任务的时候，会去寻找集群中所有节点里，对应这个 Map 任务的数据块所在的 worker 节点，将这个任务分配给这个 worker 节点，这样在读取输入的时候，每一个 worker 节点获取到的 Map 任务所需的数据块都在本地，从而避免了网络开销

因此，获取 Map 输入的数据不需要网络通信，本地获取即可，Master 节点保证。产生中间数据也不需要网络通信，保存在本地 GFS 上。但是中间数据分发到不同的 Reduce Worker 节点的时候还是需要网络通信开销。

数据Shuffle洗牌

正如上文所说，Reduce Worker 节点在开始之前会先收集中间数据

这些中间数据最初是 Map Worker 产生的一行一行的数据

但是最终在网络通信后成为了在 Reduce Worker 节点的一列一列的具有相同 key 的数据进行输入

论文中将这一个过程称之为 Shuffle （数据清洗/洗牌）

在之前描述的 MapReduce 的过程中，Reduce Worker 需要一直要等到所有的数据都获取到了才会进行 Reduce 处理，这是一种批量处理。我们也可以让 Reduce Worker 尝试通过 Stream 流水线的方式获取输入数据

我们分析这一个阶段，就会发现有大量的网络通信

在一些场景中，Reduce 的输出结果可能会非常巨大，比如排序，比如网页索引器

10TB的输入对应的是10TB的输出

同时为了提高性能并保留容错性，数据会有2-3份副本

这意味着，不论你写什么，你总是需要通过网络将一份 Reduce 的结果数据拷贝写到2-3台服务器上

所以，这里会有大量的网络通信，这里的网络通信，是2004年限制 MapReduce 的瓶颈

而现在，在现代数据中心中，root交换机比过去快了很多

一个典型的现代数据中心网络，会有很多的root交换机而不是一个交换机（spine-leaf架构）

每个机架交换机都与每个root交换机相连，网络流量在多个root交换机之间做负载分担，所以，现代数据中心网络吞吐比之当年 Google 提出 MapReduce 的时候可以说是大多了