MIT-6.824 Distributed Systems-LEC 1 Introduction

MIT-6.824（Spring 2022）LEC 1 Introduction

MapReduce论文阅读

参考翻译

方法提出

• 大量的数据分布在不同的机器上，为了能让某些算法在可以接受的时间内完成，需要将算法分配到不同的机器上一起并行运行
• 受Lisp的启发，我们发现大多数的操作都可以分为两个部分，map和reduce
  • 首先将输入中的逻辑记录应用map操作转化为过渡的键值对
  • 然后将相同的键对应的值应用reduce操作，从而合并上一步产生的过渡数据

编程模型

（WordCount）

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函数输出每个单词和计数的数量，reduce汇总其中某个特定单词的数量并输出。

• 分布式查找：map函数匹配到了就直接输出，reduce函数不发挥作用
• 计数URL访问频率：map函数对网页的日志进行处理并输出中间键值对，reduce函数再进行汇总处理
• 反转网页-链接图：map函数在source网页中寻找target URL，输出 <target, source>键值对，reduce函数对目标URL汇总source并输出
• 节点的主干词向量
• 倒排索引
• 分布式排序：map从每一条记录中提取键，reduce输出所有的键值对（后面详细说明）

Google对MapReduce的一种实现

如上图所示，map的过程是在多机器上调用的，其中分配的过程是自动化的，共分配了个节点进行。reduce过程是通过用户指定的节点数量，通过某种方法（如计算哈希值等）分配台机器进行。

其中有一个master节点，这个节点负责将任务进行分配，有些机器进行map操作，有些机器进行reduce操作等。

被分配到map任务的节点读取输入，将处理好的内容写入缓存，周期性的存入硬盘。存入时直接分为部分，并将数据存放的位置告知master

当一个节点被master通知要进行reduce时，通过RPC的方式从硬盘中读取数据到缓存中，进行处理并排序，保证相同的key出现在相同的位置

最终输出的时的文件，但是并不需要用户进行手动合并，因为这些文件通常是作为下一阶段的输入。

Master数据结构

对于每一个map任务或者reduce任务，都要保存任务的状态（已经完成或者未完成）以及工作节点的信息

对于每一个完成后的map任务，还要保存完成后的中间数据的位置和大小等信息

容错机制

机器太多了肯定有的机器会失效

Worker失效：Master会定期ping每一个Worker，如果没有得到响应，将这个节点标记为失效

• 如果节点的任务正在进行，将分配给它的任务还原到初始状态，给没有失效的节点去完成
• 如果节点的任务已经完成，对于map任务要重做，因为无法访问这个节点的存储。对于reduce来说不需要，因为已经输出到文件了
• map任务重做时会通知所有的reduce任务的节点

Master失效：Master的数据要经常备份，且由于只有一个Master，不太可能失效（因为被保护好了？），因此如果Master失效了会终止整个任务

故障时处理的机制：用户提供的Map和Reduce操作是输入确定性函数时，分布式的计算要保证任何情况下的输出都要一致没有错误.

使用map和reduce的原子提交特点来实现。map和reduce操作都写入临时文件中，完成操作后通知Master节点。如果Master节点被通知了另外一次，则直接忽略掉。reduce操作结束后将临时文件重命名为最终输出的文件，重命名操作也是原子性，最终只会有一个符合条件的文件名。

存储位置

尽量存储在本地的硬盘中，通过GFS把每个文件按64MB一个块，并在不同的机器上存储三份冗余的数据。

任务粒度

理想情况下和都应该比物理节点数量大得多，在每台机器都执行大量的不同任务能够提高集群的动态的负载均衡能力，并且能够加快故障恢复的速度。

在我们的具体实现中对和的取值有一定的限制，因为master必须执行)次调度，并且在内存中保存个状态（一个字节一个状态）

值通常由用户指定，实际使用中选择合适的值，以使得每一个独立任务都是处理大约到的输入数据

MapReduce的合适执行比例：，，使用台机器节点

备份任务

在运算过程中，如果有一台机器花了很长的时间才完成最后几个Map或Reduce任务，会导致MapReduce操作总的执行时间超过预期。

当一个MapReduce操作接近完成的时候，master会调度备用任务进程来一起执行最后的任务，谁完成了整个任务都算完成。

任务细节

在具体的实现上，对上面描述的简单mapreduce过程可以进行优化

1. reduce前需要先分配map的结果，使用哈希函数的方式分配的比较均衡，但是可能有一些场景下需要将特定的键值对分配到一起，因此用户可以传入自定义的类似于哈希的函数进行分配
2. 确保在给定的分区中，键值对数据的处理顺序是按照键进行排序后的。排序后对后面的任务都有利
3. Map函数产生的中间key值的重复数据会占很大的比重（成千上万个<the,1>），因此允许用户指定一个可选的combiner函数，combiner函数首先在本地将这些记录进行一次合并，然后将合并的结果再通过网络发送出去。一般情况下，Combiner和Reduce函数相同。区别在于输出到最终文件还是中间文件。
4. MapReduce支持不同的格式的输入数据，如文本或者键值对等，同时提供Reader接口使用户可以自定义输出，只要保证输入是可以分割的就可以
5. 某些情况下，在Map或Reduce操作过程中增加辅助的输出文件会比较省事。（但是这里不支持？）
6. 用户程序中的bug导致Map或者Reduce函数在处理某些记录的时候会崩溃掉。这个bug可能很难找。因此提供了一种执行模式，在这种模式下，为了保证保证整个处理能继续进行，MapReduce会检测哪些记录导致确定性的crash，并且跳过这些记录不处理。
7. 在远程分布式节点上调试程序非常困难，因此开发了一套MapReduce库的本地实现版本，可以调试使用
8. master使用嵌入式的HTTP服务器（如Jetty）显示一组状态信息页面，用户可以监控各种执行状态
9. MapReduce库使用计数器统计不同事件发生次数。比如，用户可能想统计已经处理了多少个单词、已经索引的多少篇German文档等等。可以用于MapReduce操作的完整性检查。

实验表现

• 在大约1TB的数据中进行特定的模式匹配（从海量数据中抽取感兴趣的数据）
• 对大约1TB的数据进行排序（对数据的形式进行转换）

应用

1. 大规模机器学习问题
2. Google News和Froogle产品的集群问题
3. 从公众查询产品（比如Google的Zeitgeist）的报告中抽取数据。
4. 从大量的新应用和新产品的网页中提取有用信息（比如，从大量的位置搜索网页中抽取地理位置信息）。
5. 大规模的图形计算。

MapReduce的成功取决于采用MapReduce库能够在不到半个小时时间内写出一个简单的程序，这个简单的程序能够在上千台机器的组成的集群上做大规模并发处理，极大的加快了开发和原形设计的周期。另外，采用MapReduce库，可以让完全没有分布式和/或并行系统开发经验的程序员很容易的利用大量的资源，开发出分布式和/或并行处理的应用。

结论

MapReduce的成功有几个方面：

1. MapReduce封装了并行处理、容错处理、数据本地化优化、负载均衡等等技术难点的细节，使得MapReduce库易于使用。
2. 大量不同类型的问题都可以通过MapReduce简单解决。
3. 实现了在数千台计算机组成的大型集群上灵活部署运行的MapReduce，使得有效利用这些计算资源变得非常简单，适合用来解决其他需要大量计算的问题。

从MapReduce开发过程中也学到了不少东西。

1. 使用固定的编程模式使得并行和分布式计算非常容易，也易于构造容错的计算环境；
2. 网络带宽是稀有资源。大量的系统优化是针对减少网络传输量为目的的：本地优化策略使大量的数据从本地磁盘读取，中间文件写入本地磁盘、并且只写一份中间文件也节约了网络带宽
3. 备份服务器执行相同的任务可以减少性能缓慢的机器带来的负面影响（硬件配置的不平衡），同时解决了由于机器失效导致的数据丢失问题。

LEC 1

什么是分布式系统

• 多个计算机通过网络连接，因此只能通过发送和接收数据包的形式进行交互，不能共享内存等等。
• 支持应用程序的基础设施主干架构

分布式系统的作用

• 连接物理上分离的机器-允许用户之间的数据共享
• 通过并行提升性能
• 容错机制-挂掉的机器不能影响服务
• 通过将程序分布在不同的机器上获得安全性（例如一台机器只用于登录服务的验证）

分布式系统的发展历程

• 起始于局域网出现（AFS）-DNS、Email
• 数据中心（大量数据）和大型网站（大量用户）
• 云计算
• 很难跟上时代发展节奏，一直在不断努力

分布式系统的挑战

• 很多并行的部分
• 容错机制
• 很难实现分布式的性能优势

判断系统是否正常工作非常困难，例如两台机器间的网络挂掉，两边都认为对方挂掉了，因此对外提供了两份服务。

课程关注的内容

课程不关注应用程序，只关注基础设施，也就是支撑这些应用程序正确工作的部分。

关注的三个方面：存储、计算和通信

抽象：分布式系统的抽象与单机系统的抽象基本相同

重点内容

容错机制

• 可用性：使系统高可用的技术，某个节点挂掉仍然可以正常工作
  • 关键：复制
• 可恢复性：挂掉的机器重启后还能回到分布式系统中继续工作
  • 关键：日志或事务

一致性：分布式系统与单机的行为相同

性能：不同类型的一致性和容错机制与性能相关

• 吞吐量
• 低延迟：某些很慢的机器会拖慢整个程序的运行过程

实现细节：如何实现并发、远程过程调用等等

MapReduce

背景

在Google早期的数据中心，有一个搜索引擎，需要构建万维网的倒排索引，允许用户上网查询。

在这个过程中处理TB级别的数据需要耗费几个小时。

为每一个应用都编写一个这种系统很困难，因此提出了MapReduce，使得构建不同应用的分布式程序比较轻松

不过这些应用必须要能分成map和reduce两个部分，然后放到MapReduce框架下运行，不需要再关注其他细节（如容错机制等等）

框架图

1. Map操作统计所有的输入文件，不同机器节点之间没有通信
2. Shuffle：从每个Map获取输出，按照键进行排序（最难的操作）
3. 在键相同的字段上运行Reduce

主要的网络通信在于传输map产生的中间文件给reduce使用

容错机制

如果一个机器在一定的时间内没有对Coordinator作出响应，就认为这个机器已经挂掉了，因此Coordinator会重新安排其他机器重启它的任务。

map和reduce任务可能会运行两次，例如Coordinator认为这个机器挂掉了，把它的任务分配给别人了，但是实际上这个机器并没有挂掉。最终使用重命名操作的原子性确保只存储一个结果。

Coordinator会挂掉吗？挂掉了整个任务就都要重新跑了，一般不会挂掉。

一些机器可能会运行很慢从而拖累整个任务的进程。当整个任务快要结束的时候，会复制任务到其他的空闲节点上一起做，谁先做完取谁的。