MIT 6.824学习心得(1) 浅谈分布式系统概论与MapReduce
一个月前机缘巧合，有朋友向我推荐了麻省理工学院非常著名的分布式系统课程MIT 6.824，是由世界五大黑客之一，蠕虫病毒之父Robert Morris教授进行授课。由于我自己也在做基于分布式微服务架构的业务项目，所以对构建分布式系统这个课题非常感兴趣，想要探寻其中的一些底层原理。经过一段时间的学习确实感觉受益匪浅！目前还在学习课程和做lab的过程中，不得不说还是很有挑战性的，所以也想用文字的形式来记录下自己曾经的学习心得体会分享给大家，有不当之处还请多多批评指正！
一.分布式系统概述
在早期的实际应用中，大多数系统属于“集中式”架构。所有的存储，计算都集中在一台服务器上处理。确实，这种系统在架构层面很简单，作为开发人员只需要重点关心服务端的处理逻辑即可，但是毫无疑问，存在明显的瓶颈。首先系统性能会受限于单机资源，因为一台物理服务器它的内存，磁盘，CPU资源都是很有限的；其次单点故障会使得整个系统不可用。而且这种单机系统可扩展性很差，难以应对业务规模的爆发增长。
在上世纪80~90年代，人们开始意识到了集中式架构已经很难满足互联网发展的需要了，分布式系统架构的设计思想由此应运而生。分布式系统采用多台计算机进行互联，共同完成某个任务，而对外屏蔽了系统内部的细节，从用户或者应用开发者的角度来看仍然只是一个系统。分布式系统中的每台机器称为“节点”。由此也衍生出了一些比较轻量级的RPC框架，分布式数据库等等。
随着互联网行业的快速发展，系统架构也在逐渐演变。在业务层面上将单体架构拆分成多个微服务部署到不同的服务器上，利用更多的物理计算资源并行工作，分布式微服务架构也由此诞生。同时，数据存储的方式也发生了巨大的变化，由集中存储到分片存储，也由此引出了水平分片，主从复制等架构方案。特别是Google三大论文（GFS,MapReduce,BigTable）更是推进了分布式系统发展的速度。而现代的分布式系统逐渐趋向于云原生。充分利用云计算资源的弹性伸缩，自动化编排等能力。分布式系统的主要设计目标主要是应对以下几个方面的挑战：
（一）可扩展性
请看下图中的场景：
假如你在一台Web服务器上搭建了一个网站，而此时有多个客户端同时访问这台服务器的资源。由于服务器的CPU，内存等硬件资源有限，当客户端请求达到一定数量之后，服务器会难以承受压力而宕机。所以我们会很自然的想到增加Web服务器的数量，把这些请求分配到多台机器上进行处理，这就是水平扩展的策略。但是，无限增加Web服务器的数量，数据库又会成为系统的性能瓶颈，而我们又很难保证这些请求能够平均分配到每台机器上。所以我们使用一致性哈希等分片策略保证请求的均匀分配。可扩展性追求的是在增加硬件资源的同时，系统的性能可以获得接近线性的提升。
（二）容错性
假如你设计了一个分布式系统，如果你需要让这个系统足够健壮，在部分节点或者组件发生故障时仍然能正常对外提供服务，或者采用降级策略，就必须要在容错层面上有比较科学的设计。其共同思想是可用性和自身可恢复性。比如在一个电商系统当中，如果因为某一个订单服务所在的服务器节点宕机，而导致用户无法下单，那说明这个系统的容错性还有待提升。我们真正期望的设计是如果你部署了多个订单服务节点实例，当其中一个节点挂了，另外一个节点可以顶上，如果实在没有办法，仍然有例如返回默认数据等服务降级策略。而保证容错性的两个关键要点分别是非易失性存储和复制。由此我们需要在设计架构时根据实际业务需求考虑部署多个副本，主从也好，集群也罢。而且要有合理的限流，熔断，降级策略和健康检查机制。
（三）一致性
在分布式系统中，只要存在多个服务器节点，那必然会存在数据同步的相关问题。在分布式系统中，多个副本同时进行读写IO操作，很容易引发数据不一致。而实际上，一致性追求的是所有节点的数据状况是相同的。一致性分为两种情况，一种是强一致性。追求的是所有节点的数据必须始终保持同步，比如说在用户注册的时候，在写完注册数据之后，立刻可以读取返回新的信息，而且所有节点都必须能够读取到新数据，这个过程中不能有延迟。因为这种全量即时同步的机制，所以如果要追求强一致性，那一定意味着更复杂的系统设计和更昂贵的通信成本。另外一种是弱一致性，数据最终会同步，但是会存在延迟。比如说用户刚注册完就从服务器读取用户数据时可能会读取到旧数据，几秒之后新的数据才同步到服务端。我们在系统设计时具体需要采用哪种理念，需要严格结合业务场景进行分析。开发者对于一致性的接受范围往往取决于他们对异常行为的可接受度，比如会妥协追求最终一致性或版本一致性。
（四）CAP定理
首先我们要引出一个概念--分区容忍性。这个性质是针对节点间的通信来讲的。当分布式系统中发生网络分区（通信失败）时，系统仍然能够运行，而不是直接挂掉。假如你的公司在福州厦门两地都部署了机房，某一时刻如果网络出现问题，服务间的通信断了，我们依然能够访问某些服务，哪怕只是功能受限。
2000年，加州大学伯克利分校的Eric Brewer教授提出了CAP定理，被视为分布式系统设计的的“铁律”。即在分布式系统中，不可能同时满足以下三点：一致性（C），可用性（A），分区容忍性（P），因为你不能同时在网络断开的情况下，既响应用户请求，又确保所有副本的数据是一样的。而分布式系统中一定会存在网络分区，所以分区容忍性（P）是必须保证的。所以在实际架构设计中，我们只能在一致性和可用性二者之间做权衡。如果牺牲可用性，追求强一致性（如Zookeeper,Etcd），网络异常时会直接拒绝服务。如果牺牲一致性，追求高可用（如Cassandra），则必须允许短时间内的数据不一致。比如在一个实际的后端项目中，Redis缓存的设计会更偏向AP，缓存中的数据可能不是最新的，但是保证了高可用和容错性。而Etcd配置中心的设计偏向CP，配置必须保证强一致性不能有偏差。
二.分布式高性能计算框架-MapReduce
（一）引子-单词计数
MapReduce是一个用于大规模数据集（TB级）并行运算的分布式框架，于2003年由Google公司首先提出。不少公司在面试过程中会出现类似MapReduce框架设计的场景题，所以对于我们来说大致理解MapReduce的设计思想对于加深对分布式系统的理解是很有帮助的。这里我把MapReduce的论文链接直接贴在下方，如果想要真正深入理解MapReduce框架还是建议去读一读这篇论文，也许会有不一样的收获。
MapReduce论文原文
MapReduce设计的初衷在于对业务开发者屏蔽掉分布式计算的相关细节，由MapReduce框架自动完成，开发者只需关注Map和Reduce两个函数即可。在这里我们先不直接讲述Map函数和Reduce函数的含义，我们先通过一个简单的单词计数实例大概了解下MapReduce框架。
如上图所示，有三个客户端分别向服务器输入了不同的单词数据，我们服务端的设计目的，是为了计算出每个单词出现的次数并且返回。对于这个框架，只有Map函数和Reduce函数是提供给开发者的显式接口。Map函数负责读取输入，并将其转换为（key,value）键值对。在上面这个例子当中，Map函数读取输入的单词并以此为key，统计词频为value。当然，key和value的实际意义是根据开发者的实际需求来定制的。而由Map产生的这个(key,value)键值对便是这个计算过程的一个中间结果。之后在框架内部，有一个Shuffle函数（对开发者透明，由框架自动执行）将所有Map阶段输出相同的key的记录汇总到一起，交给对应的Reduce函数进行处理，由Reduce函数进行汇总过程的执行。而emit函数存在于Map/Reduce函数内部，负责输出(key,value)键值对。由上面的过程我们不难看出，运用MapReduce进行单词计数的过程可以用“分词-分组-计数”来概括。而MapReduce框架的核心处理流程便是“局部处理-分组-汇总处理”。简单了解了MapReduce处理问题的方法之后，我们来结合MapReduce的论文和架构图来更详细严谨的讨论MapReduce框架。
（二）逻辑模型
MapReduce论文中对于Map和Reduce函数的定义，翻译过来是这样的：“Map函数由用户编写，其主要功能是获取一个输入的(key,value)键值对并生成一个中间态的(key,value)键值对。
MapReduce框架会自动对所有的(key,value)键值对进行分组，使得所有有着相同中间态key值的(key,value)键值对的value组合在一起，然后将其传递给Reduce函数进行处理。Reduce函数也由用户编写，其主要功能是接收一个中间态的key值和与该键对应的一组value值的集合。它会将这些value值进行统一的合并以形成一个可能更小的value值集合。”如果能够理解上面的例子，那么相信对于这段定义的理解就很容易了。MapReduce不仅仅可以用于进行词频统计，还可以应用于日志统计，倒排索引，社交推荐，分布式排序等更复杂的场景。
接下来我们来讨论下MapReduce框架的逻辑模型，不过实际的系统设计远远不会这么简单，但是一定与这个逻辑模型有共通之处。上图是MapReduce论文中的架构图，我们结合着这张架构图来进行分析会比较直观。
首先用户程序的MapReduce库，也就是客户端会将输入的数据进行分片（假设分片数为M）。论文给出的每个分片的参考大小通常为16MB至64MB，可以由用户通过相关的参数自行设定。之后用户程序便在服务器集群的其中一组物理机器上通过fork()系统调用，启动多个该程序的副本，此时用户程序。这些副本构成了MapReduce框架的处理集群。其中，有一个副本是特殊的，也就是图中的Master。Master是这个计算集群的中心节点，主要用于分配任务。Master中维护了一些数据结构，存储了每个Map/Reduce的任务状态以及Worker机器的ID。而其他的副本都是Worker，这些才是真正的处理程序。Master选择空闲的Worker，并给其分配Map任务(assign map)或者Reduce任务(assign reduce)，被分配了相应任务的Worker节点也称Map Worker或Reduce Worker。分配任务之后Worker会进行下一步的处理。
MapWorker会通过read()系统调用读取分片后的数据内容。从输入的数据中通过某些逻辑解析出(key,value)键值对，然后将这个(key,value)键值对作为参数传递给用户定义的Map函数，由Map函数进行处理生成中间态(key,value)键值对（该过程也叫emit）。这个中间态键值对会被缓存在内存中。缓存在内存中的中间态(key,value)对会被定期写入本地磁盘(local disks)中进行持久化。而这些被缓存的中间态(key,value)对的位置会被回传给Master节点，而Master本质上是一个管道，负责将这些位置信息传递给对应的Reduce Worker。
当一个Reduce Worker被Master节点告知了某个中间态(key,value)键值对的位置信息，会使用RPC（远程过程调用）从对应Map Worker的本地磁盘上读取这些缓存数据（remote read）。当一个Reduce Worker已经读取了所有的中间态数据时，会根据其key值进行排序，拥有相同key值的(key,value)键值对会被分类到一起，这便是我们前面例子中的shuffle过程。需要进行排序的原因是通常会有不同key的(key,value)对会被映射到同一Reduce任务。如果需要排序的中间态数据量过大，内存无法一次装载，可能需要考虑使用一些外部排序的方法。之后Reduce Worker会遍历排好序的中间态(key,value)键值对，并将所遇到的每一个唯一的key值和对应的中间态value值集合作为参数传递给用户自定义的reduce函数。reduce函数产生的输出将会追加在一个该分区的输出文件内，每个Reduce任务都会对应一个输出文件。在很多情况下，用户无需对这些输出文件进行合并，而是传递这些文件，也许它们也可以作为下一个MapReduce任务的输入。在所有的Map和Reduce任务都结束之后，Master会唤醒用户程序。此时MapReduce框架的执行结果才返回给用户程序。
（三）问题讨论
以上便是MapReduce框架工作流程的概述。毕竟这是一个基于分布式架构设计的框架，所以接下来我们要讨论这个系统中可能发生的问题，以及如何保证这个系统应有的分布式特性。
首先我们来讨论MapReduce框架如何保证容错性呢？在实际场景中，很有可能会出现以下两种情况，即Worker节点挂掉或者Master节点挂掉。首先我们需要了解，Master节点和Worker节点之间是如何通信的。在MapReduce论文中提到的通信方式是Master会定期向Worker发送心跳包确认其活性，并且当有任务需要分配时会主动通知Worker。
假如是Worker节点出了故障，Master会周期性的ping每一个worker，当在一定时间内没有收到来自Worker的响应时，Master就会将这些Worker标记为有故障。所有由该Worker完成的Map任务将会失效，重新初始化，因为这些Map任务产生的中间态数据是存储在Map Worker所在服务器的本地磁盘的，宕机后会访问失败。由此这些Map任务可以被其他Worker调度执行。执行中的Map，Reduce任务也是同理，为什么执行完成的Reduce任务不是这样呢？因为Reduce任务产生的结果是储存在全局文件系统中的。由此，MapReduce能从大范围的Worker故障中迅速的修复。
假如是Master节点出现了故障，其实这是个非常棘手的问题，毕竟Master节点是MapReduce框架的核心。所以我们只能做一些预防措施在Master节点挂了的时候进行补救，而补救的最好方法就是通过持久化的数据。我们可以让master周期性的将master数据结构以检查点的形式进行持久化。如果Master节点挂了，新的master备份节点将会从最新的检查点状态处启动。
接下来，在一个庞大的分布式系统中，成千上万台物理机器要进行通信，此时网络IO无疑成为了这个系统最大的性能瓶颈。为了尽可能的节省网络带宽，开发者会将GFS服务器和MapReduce Worker运行在同一个计算机集群中（GFS的相关知识我们后续会单独出专题进行介绍）。GFS将每个文件分割为64MB的块，同时为每一个块存储几个备份在不同的机器上。MapReduce的master调度map任务时尽量在包含对应输入数据副本的机器上调度执行一个map任务。如果任务失败了，调度map任务时会让执行任务的机器尽量靠近任务所需输入数据所在的机器，例如被选中的worker机器与包含数据的机器位于同一网段中。当集群中的相当一部分worker都在执行大型MapReduce操作时，绝大多数的输入数据都在本地读取从而不会消耗网络带宽。
现在有一个疑问，在一个MapReduce框架中，Map任务数量（M）和Reduce任务数量（R）具体应该怎么确定呢，也就是说MapReduce框架的任务粒度是怎么确定的？论文中给出的阐述是这样的：M的数量取决于输入文件的分片个数，即输入数据的切分粒度。如果数量太少，则系统并行度太低；如果数量太多，则会给Master的管理造成很大的负担，IO压力也随之增加。在Map输出阶段之后，所有的中间态(key,value)键值对会根据key进行哈希。而R的数量应该是相对固定的，根据最终的输出量进行确定，因为所有 map 输出都要知道最终 key 属于哪个 reduce 文件，必须事先固定好 R的数量，整个系统才能协调中间结果。理想情况下，M和R的值都应该远大于worker机器的数量，Worker数量才是动态运行的实体数，这样可以让每一个worker执行很多不同的任务可以提高动态负载均衡的效率，同时也能加快worker故障时的恢复速度。
以上便是我对分布式系统和MapReduce框架的粗浅认知，目前还在做MapReduce框架的lab，所以这篇文章的观点会根据我做lab出现的问题实时更新，lab完成后我也会在本文加上lab的实现思路供大家参考，有不当之处还请批评指正，我们一起进步！