Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing

本文提出了一种称之为RDDs的分布式内存抽象，以此解决在大规模集群中以容错的方式提供内存计算的方式。当前的计算框架对于迭代算法和交互式数据挖掘的效率都很低，RDDs通过将数据留在内存来提高性能。本文通过Spark系统来实现RDDs。

Introduction

诸如MapReduce 和Dryad之类的集群计算框架已被广泛用于大规模数据分析。这些系统使用户可以使用一组高级API来编写并行计算，而不必担心工作分配和容错能力。

尽管当前的框架为集群的计算资源提供了许多抽象，但它们还是缺乏对利用分布式内存的抽象。这就导致了在多个计算之间复用中间结果时，显得非常低效。数据重用在许多迭代机器学习和图计算中很常见。另外在交互式数据挖掘中，用户会需要对数据的同一子集运行多个临时查询。然而在大多数框架中，在计算之间（例如在两个MapReducce作业之间）重用数据的唯一方法是将其写入外部稳定的存储系统，例如分布式文件系统。由于数据复制，磁盘IO和序列化，这会导致相当大的开销，这可能会影响应用程序的执行时间。

在这篇论文中提出了一个全新的抽象，叫做RDDs(Resilient Distributed Datasets)，它可以在广泛的应用程序中实现有效的数据重用。RDDs 是一个可以容错且并行的数据结构，它可以让用户显式的将中间结果数据集保存在内中。

现存的分布式内存抽象系统，都是基于对可变状态的细粒度更新。这种接口保证容错的方式无非是将数据进行多副本备份，需要在机器节点间复制大量的数据，宽带传输数据的速度远远比RAM 内存慢。

与这些系统相比，RDD提供了基于粗粒度转换的接口（map，reduce，filter）。这些接口可以对多条数据条目应用相同的操作，这样就可以通过记录来生成某个数据集的一系列转换，而不是记录真实的数据。如果RDD丢失，则RDD具有足够的有关如何从其他RDD派生的信息，可以仅重新计算该分区。因此，丢失的数据通常可以很快恢复。

Resilient Distributed Datasets

本章主要介绍RDD和Spark编程接口，并与细粒度共享内存做对比。

RDD Abstraction

RDD是一个只读的、可分区的数据集，可以通过对稳定的存储系统或者其他的RDD进行操作来创建一个新的RDD，这些操作称之为transformations，比如map，filter 以及join。另外用户可以控制RDD的存储和分区，指定存储策略，也可以根据key做hash来做数据分区。

Spark Programming Interface

Spark通过集成编程语言API来表示RDD，每一个数据集就是一个对象，通过对象的方法来操作对象。RDD有两种操作，一种是上面说的transformations，另一种则是action，action操作可以得到应用结果值，比如count可以返回数据集的元素个数、collect返回数据集的所有元素以及save则是将输出结果写入到存储系统中。

Spark定义RDDs是并不会计算，只是采取lazy特性，可以将transformations组成pipeline，触发了actions操作才会真正计算。用户可以通过RDDs的preset方法来缓存数据，也可以调整缓存策略。

Advantages of the RDD Model

论文将RDD和分布式共享内存系统DSM做了比较，RDD只能粗粒度的操作转换，而DSM可以在任意内存位置进行写入。这样RDD的容错机制更加高效，不需要发生非常耗时的checkpoint，只需重新计算丢数据的分区。另外一个好处就是任务备份比较简单，因为RDD是不变的。还有就是，RDD可以进行进行任务调度来提高大批量的写入效率，在scan-base的操作中也能根据需要将内存数据写到磁盘中。

Applications Not Suitable for RDDs

RDD更适合批量的数据处理场景，并不适合于需要异步且细粒度的更新共享状态的应用。

Spark Programming Interface

Spark提供了一个用Scala编写的语言集成API。为了使用Spark，开发者编写了一个driver，该driver会连接workers集群，并定义若干个RDDs，在RDDs上执行action，在driver上的Spark代码会追踪RDDs的lineage。workers是一直运行的进程，能在内存中存储RDD分区。

RDD Operations in Spark

下图列出了Spark中RDD的transformations和actions操作。transformations是定义新RDD的lazy操作，而actions才是真正计算结果或者写数据到外部存储；

Representing RDDs

抽象RDDs会带来一个问题：如何在广泛转换中表示追踪lineage。理想情况下，一个实现RDDs的系统应该能够提供丰富的·转换算子，用户可以以任意方式进行组合。在Spark中则是提出了一个简单的图表示来达到以上目的。

论文提出了一个通用接口去表示RDD，接口表达了五种信息：

一组分片（partitions），数据集的原子组成；
一组父RDDs上的依赖；
一个基于父数据集计算的函数；
分片策略元数据，一个分片函数partitioner；
数据位置策略，存储每个partition的优先位置；

论文将RDDs之间的依赖分为了两类：

窄依赖：父RDD的每个分片被子RDD至多一个分片使用；
宽依赖：多个子分片依赖一个父分片；

例如，代表HDFS文件的RDD对文件的每个块都有一个分片，并且通过数据位置策略知道每个块在哪台计算机上。

窄依赖能在一个节点上流水线执行，节点故障的时候也能高效地通过重新计算父分片来进行恢复；而宽依赖，单一节点故障可能会导致一个RDD的所有祖先分片丢失，需要完全重新执行。

Implementation

Spark可以从任何的Hadoop输入源中读取数据，比如HDFS和HBase。本章主要关注下面的几个部分：任务调度、Spark解释器的交互式使用、内存管理和checkpoint。

Job Scheduling

Spark的调度器与Dryad类似，另外还会考虑持久化了的RDD的哪些分片在内存中可用。任何时候用户在RDD上执行action，调度器就会检查RDD的lineage，建立由stages组成的DAG，然后执行这个图。调度器会使每个stage包含尽可能多的窄依赖，stages的边界是宽依赖shuffle操作，或者任何计算过的分片。

调度器会根据数据存放位置使用延迟调度给机器指派任务。

若一个任务失败了，只要stage的父分片还在，就可以在另一个节点重新运行。如果一些stages都不可用了，就需要重新提交任务去并行计算丢失分片。

Interpreter Integration

Scala包含一个类似于Ruby和Python的交互式shell，考虑到内存数据的低延迟，Spark可以让用户在解释器上运行。

Spark中的编译器相对Scala做了一些改变：

类传输：通过HTTP传输创建类的字节码；
代码生成：代码生成的单例对象是通过生成类的静态方法访问的，为了避免序列化一个访问不到前面定义变量的闭包，Spark将代码生成逻辑改成直接引用每行对象的实例；

Memory Management

Spark对RDD的持久化提供了三个选项：

序列成Java对象，存在内存中；性能最好
作为序列化数据存在内存中；内存空间有限时使用
存在硬盘中；RDDs过大无法存入内存

当计算新的RDD分片后，如果没有足够空间去存储，就会基于LRU的淘汰策略去淘汰一个分片。但如果新旧分片属于同一个RDD，则会将旧的分片写入内存，避免相同RDD的分片循环读写。

Support for Checkpointing

虽然lineage可以帮助恢复RDDs，但如果lineage很长的时候就会变得很耗时，因此RDD可以执行checkpoint存入稳定内存。

Spark为checkpoint提供了一个API，让用户决定checkpoint哪个数据。同样，Spark的调度器也制定每个数据集大小，了解第一次计算的耗时，因此也会基于一定的策略选择一个优化RDDs集合来执行checkpoint，缩短系统恢复时间。

总结

本文主要介绍了一个在集群中共享数据的高效的、具备容错能力的的抽象——RDD。RDD能表达通用的并行应用，提供了一个基于粗粒度转换的API，也能通过lineage来快速恢复数据。