概要

为了能解决在大规模的集群中以一种容错的方式进行内存计算这个问题, 我们提出了 RDDs 的概念. 当前的很多框架对迭代式算法场景与交互性数据挖掘场景的处理性能非常差, 这个是 RDDs 的提出的动机. 如果能将数据保存在内存中, 将会使的上面两种场景的性能提高一个数量级. 为了能达到高效的容错, RDDs 提供了一种受限制的共享内存的方式, 这种方式是基于粗粒度的转换共享状态而非细粒度的更新共享状态.

介绍

RDDs可以高效的处理广泛的应用中涉及到的数据复用的场景. RDDs 是一个可以容错且并行的数据结构, 它可以让用户显式的将中间结果数据集保存在内存中、控制数据集的分区来达到数据存放处理最优以及可以使用丰富的操作 api 来操作数据集.
RDDs 提供了基于粗粒度转换（比如 map, filter 以及 join）的接口, 这些接口可以对大量的数据条目应用相同的操作.这样就可以通过记录来生成某个数据集的一系列转换 （就是这个数据集 lineage）而不是记录真实的数据来达到提供高效的容错机制. 这个 RDD 就有足够的信息知道它是从哪 RDDs 转换计算来的, 如果一个 RDD 的分区数据丢失掉了, 那么重新计算这个 RDD 所依赖的那个 RDD 对应的区就行了. 因此可以很快且不用通过复制备份方式来恢复丢失的数据.

RDD

RDD抽象

一个 RDD 是一个只读, 被分区的数据集.可以通过两种方式对稳定的存储系统和其他的 RDDs 进行操作而创建一个新的 RDDs.为了区别开 RDDs 的其他操作, 我们称这些操作为 transformations, 比如 map, filter 以及 join 等都是 transformations 操作.最后, 用户可以控制 RDDs 的两个方面：数据存储和分区.对于需要复用的 RDD, 用户可以明确的选择一个数据存储策略（比如内存缓存）. 他们也可以基于一个元素的 key 来为 RDD 所有的元素在机器节点间进行数据分区, 这样非常利于数据分布优化, 比如给两个数据集进行相同的 hash 分区, 然后进行 join, 可以提高 join 的性能.

Spark编程接口

Spark通过集成编程语言 api 来暴露 RDDs, 这样的话, 每一个数据集就代表一个对象, 我们可以调用这个对象中的方法来操作这个对象.
编程者可以通过对数据进行转换操作（即 transformations, 比如 map 和 filter 等）来得到一个或者多个 RDDs. 然后可以对这些 RDDs 进行 actions 操作, 这些操作可以是得到应用的结果值, 也可以是将结果数据写入到存储系统中, actions 包括: count（表示返回这个数据集的元素的个数）、collect（表示返回数据集的所有元素）以及 save（表示将输出结果写入到存储系统中）. spark 在定义 RDDs 的时候并不会真正的计算, 而是要等到对这个 RDDs 触发了 actions 操作才会真正的触发计算, 这个称之为 RDDs 的 lazy 特性, 所以我们可以先对 transformations 进行组装一系列的 pipelines, 然后再计算.
另外, 编程者可以通过调用 RDDs 的 persist 方法或者cache方法来缓存后续需要复用的 RDDs. Spark 默认是将缓存数据放在内存中, 但是如果内存不足的话则会写入到磁盘中. 用户可以通过 persist 的参数来调整缓存策略, 比如只将数据存储在磁盘中或者复制备份数据到多台机器. 最后, 用户可以为每一个 RDDs 的缓存设置优先级, 以达到哪个在内存中的 RDDs 应该首先写道磁盘中。
Spark 使用 scala 语言实现了抽象的 RDD, scala 是建立在 java JVM 上的静态类型函数式编程语言. 开发员需要写连接集群中的 workers 的 driver 程序来使用 spark, Driver 端程序定义了一系列的 RDDs 并且调用了 RDD 的 action 操作. Driver 的程序同时也会跟踪 RDDs 之间的的血缘关系. workers 是长期存活在可以将 RDD 分区数据存储在内存中的的进程(那些节点上有输入数据，这些节点上就会起worker进程).用户写的 driver 端程序启动多个 workers, 这些 workers 可以从分布式的存储系统中读取数据块并且可以将计算出来的 RDD 分区数据存放在内存中.

RDD之间的依赖关系

我们发现将依赖定义成两种类型就足够了: 窄依赖, 表示父亲 RDDs 的一个分区最多被子 RDDs 一个分区所依赖. 宽依赖, 表示父亲 RDDs 的一个分区可以被子 RDDs 的多个子分区所依赖. 比如, map 操作是一个窄依赖, join 操作是一个宽依赖操作

调度

当一个用户对某个 RDD 调用了 action 操作（比如 count 或者 save ）的时候调度器会检查这个 RDD 的血缘关系图, 然后根据这个血缘关系图构建一个含有 stages 的有向无环图（ DAG ）, 最后按照步骤执行这个 DAG 中的 stages , 每一个 stage 包含了尽可能多的带有窄依赖的 transformations 操作. 这个 stage 的划分是根据需要 shuffle 操作的宽依赖来进行的。 然后调度器可以调度启动 tasks 来执行没有父亲 stage 的 stage ,一直到计算完我们的最后的目标 RDD .
我们调度器在分配 tasks 的时候是采用延迟调度来达到数据本地性的目的（说白了, 就是数据在哪里, 计算就在哪里）.

内存管理

Spark 在持久化 RDDs 的时候提供了 3 种存储选：存在内存中的非序列化的 java 对象、存在内存中的序列化的数据以及存储在磁盘中. 第一种选择的性能是最好的, 因为 java JVM 可以很快的访问 RDD 的每一个元素. 第二种选择是在内存有限的情况下, 使的用户可以以很低的性能代价而选择的比 java 对象图更加高效的内存存储的方式. 如果内存完全不够存储的下很大的 RDDs , 而且计算这个 RDD 又很费时的, 那么选择第三种方式.
为了管理有限的内存资源, 我们在 RDDs 的层面上采用 LRU （最近最少使用）回收策略. 当一个新的 RDD 分区被计算但是没有足够的内存空间来存储这个分区的数据的时候, 我们回收掉最近很少使用的 RDD 的分区数据的占用内存。

对Checkpointing的支持

虽然我们总是可以使用 RDDs 的血缘关系来恢复失败的 RDDs 的计算, 但是如果这个血缘关系链很长的话, 则恢复是需要耗费不少时间的.因此, 将一些 RDDs 的数据持久化到稳定存储系统中是有必要的，一般来说, checkpointing 对具有很长的血缘关系链且包含了宽依赖的 RDDs 是非常有用的。