本部分描述RDD和编程模型。首先讨论设计目标（2.1），然后定义RDD（2.2），接着讨论Spark的编程模型（2.3），并给出一个示例（2.4），最后将RDD与分布式共享内存进行比较（2.5）。

我们的目标是为基于工作集（working set）的应用（即多个并行操作重用中间结果的这类应用）提供抽象，同时保持MapReduce及其相关模型的优势特性：即自动容错、位置感知性调度和可伸缩性。RDD比数据流模型更易于编程，同时基于工作集的计算也具有良好的描述能力。

在这些特性中，最难实现的是容错性。一般来说，分布式数据集的容错性有两种方式：即数据检查点和记录数据的更新。我们面向的是大规模数据分析，数据检查点操作成本很高：需要通过数据中心的网络连接在机器之间复制庞大的数据集，而网络带宽往往比内存带宽低得多，同时还需要消耗更多的存储资源（在RAM中复制数据可以减少需要缓存的数据量，而存储到磁盘则会拖慢应用程序）。所以，我们选择记录更新的方式。但是，如果更新太多，那么记录更新成本也不低。因此，RDD只支持粗粒度转换（coarse-grained transformation），即在大量记录上执行的单个操作。将创建RDD的一系列转换记录下来（即lineage），以便恢复丢失的分区。

虽然只支持粗粒度转换限制了编程模型，但我们发现RDD仍然可以很好地适用于很多应用，特别是支持数据并行的批量分析应用，包括数据挖掘，机器学习，图算法等，因为这些程序通常都会在很多记录上执行相同的操作。RDD不太适合那些异步更新共享状态的应用，例如并行web爬行器。因此，我们的目标是为大多数分析型应用提供有效的编程模型，而其他类型的应用交给专门的系统。

RDD是只读的记录分区的集合。RDD只能通过在——（1）稳定物理存储中的数据集；（2）其他已有的RDD——上执行确定性（deterministic）操作来创建。这些操作称之为转换（transformation），如map, filter, groupBy, join。（转换不是程序员在RDD上执行的操作。）

RDD不需要物化。RDD含有如何从其他RDD衍生（即计算）出本RDD的相关信息（即lineage），据此可以从物理存储的数据计算出相应的RDD分区（partition）。

在Spark中，RDD被表示为对象，通过这些对象上的方法（或函数）调用转换（transformation）。

定义RDD之后，程序员就可以在行为（action）中使用RDD了。行为（action）是向应用程序返回值，或向存储系统导出数据的那些操作，例如，count（返回RDD中的元素个数），collect（返回元素本身），save（将RDD输出到存储系统）。在Spark中，只有在action第一次使用RDD时，才会计算RDD（即懒计算，lazily evaluated）。这样在构建RDD的时候，运行时以流水线的方式执行（pipeline）多个转换。

程序员还可以从两个方面控制RDD，即缓存（caching）和分区（partitioning）。用户可以请求将RDD缓存，这样运行时将已经计算好的RDD分区存储起来，以加速后期的重用。缓存的RDD一般存储在内存中，但如果内存不够，可以溢出（spill）到磁盘。

另一方面，RDD还允许用户根据关键字（key）指定分区顺序，这是一个可选的功能。目前支持哈希分区（hash partition）和范围分区（range partition）。例如，应用程序请求将两个RDD按照同样的哈希分区方式进行分区（将同一机器上具有相同关键字的记录放在一个分区），以加速它们之间的join操作。在Pregel和HaLoop中，多次迭代之间采用一致性的分区放置策略（Consistent partition placement）进行优化，我们同样也允许用户指定这种优化。

本部分我们通过一个具体示例来阐述RDD。假定有一个大型网站出错，操作员想要检查Hadoop文件系统（HDFS）中的日志文件（TB级大小）来找出原因。通过使用Spark，操作员只需将日志中的错误信息装载到一组节点的RAM中，然后执行交互式查询。首先她需要在Spark解释器中敲入以下Scala命令：

lines = spark.textFile("hdfs://...")

errors = lines.filter(_.startsWith("ERROR"))

errors.cache()

第1行从HDFS文件定义了一个RDD（即一个文本行集合），第2行获得一个过滤后的RDD，第3行请求将errors缓存起来。注意在Scala语法中filter的参数是一个闭集（closure）。

这时集群还没有开始执行任何任务。但是，用户已经可以在这个RDD上执行action操作了，例如统计错误消息的数目：

errors.count()

用户还可以在RDD上执行更多的转换（transformation）操作，并使用转换结果，如：

// Count errors mentioning MySQL:

errors.filter(_.contains("MySQL")).count()

// Return the time fields of errors mentioning

// HDFS as an array (assuming time is field

// number 3 in a tab-separated format):

errors.filter(_.contains("HDFS"))

.map(_.split('t')(3))

.collect()

使用errors的第一个action运行以后，Spark会把errors的分区缓存在内存中，极大地加速了后续计算。注意，最初的RDD lines不会被缓存。因为错误信息可能只占原数据集的很小一部分（小到足以放入内存）。

最后，为了说明模型的容错性，图1给出了第3个查询的血统（lineage）关系图。在lines RDD上执行filter操作，得到errors，然后再filter、map后得到新的RDD，在这个RDD上执行collect行为。Spark调度器以流水线的方式执行后两个转换，向拥有errors分区缓存的节点发送一组任务。此外，如果某个errors分区丢失，Spark只在相应的lines分区上执行filter操作来重建该errors分区。

 （由于粘贴不便，图参见论文原文）

图1 示例中第三个查询的血统关系图。（方框表示RDD，箭头表示转换）

为了进一步理解RDD是一种分布式的内存抽象，表1列出了RDD与分布式共享内存（DSM，distributed shared memory）的对比。在DSM系统中，应用可以向全局地址空间的任意位置进行读写操作。（注意这里的DSM，不仅指传统的共享内存系统，还包括那些通过分布式哈希表或分布式文件系统进行数据共享的系统，比如Piccolo。）DSM是一种通用的抽象，但这种通用性同时也使得在商用集群上实现有效的容错性更加困难。

RDD与DSM主要区别在于，不仅可以通过批量转换创建（即“写”）RDD，还可以对任意内存位置读写。也就是说，RDD限制应用执行批量写操作，这样有利于实现有效的容错。特别地，RDD没有检查点开销，因为可以使用lineage来恢复RDD。而且，失效时只需要重新计算丢失的那些RDD分区，可以在不同节点上并行执行，而不需要回滚（roll back）整个程序。

表1 RDD与分布式共享内存的对比

注意，通过备份任务的拷贝，RDD还可以处理落后任务（即运行很慢的节点），这点与MapReduce类似。而DSM则难以实现备份任务，因为任务及其副本都需要读写同一个内存位置。

与DSM相比，RDD模型有两个好处。第一，对于RDD中的批量操作，运行时将根据数据存放的位置来调度任务，从而提高性能。第二，对于基于扫描的操作，如果内存不足以缓存整个RDD，就进行部分缓存。把内存放不下的分区存储到磁盘上，此时性能与现有的数据流系统差不多。

最后看一下读操作的粒度。RDD上的很多行为（action，如count和collect）都是批量读操作，即扫描整个数据集，可以将任务分配到距离数据最近的节点上。同时，RDD也支持细粒度操作，即在哈希或范围分区的RDD上执行关键字查找。

（未完待续）

本系列文章《弹性分布式数据集：基于内存的集群计算的容错性抽象》系《Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing》译文。原文参见这里。