一、设计背景

1.某些应用场景中，不同计算阶段之间会重用中间结果，即一个阶段的输出结果会作为下一个阶段的输入。如：迭代式算法（比如机器学习、图算法等）和交互式数据挖掘工具。

2.目前的MapReduce框架都是把中间结果写入到HDFS中，带来了大量的数据复制、磁盘IO和序列化开销。
虽然，类似Pregel等图计算框架也是将结果保存在内存当中，但是，这些框架只能支持一些特定的计算模式，并没有提供一种通用的数据抽象。

3.针对以上问题，Spark提出了一种新的数据抽象模式称为RDD（弹性分布式数据集），RDD是容错的并行的数据结构，并且可以让用户显式的将数据保存在内存中，并且可以控制他们的分区来优化数据替代以及提供了一系列高级的操作接口。

二、RDD概念

1.一个RDD就是一个分布式对象集合，本质上是一个只读的分区记录集合，每个RDD可以分成多个分区，每个分区就是一个数据集片段，并且一个RDD的不同分区可以被保存到集群中不同的节点上，从而可以在集群中的不同节点上进行并行计算。

2.RDD是只读的记录分区的集合，不能直接修改，只能基于稳定的物理存储中的数据集来创建RDD，或者通过在其他RDD上执行确定的转换操作（如map、join和groupBy）而创建得到新的RDD。

3.RDD提供了一组丰富的操作以支持常见的数据运算，分为**“行动”（Action）和“转换”（Transformation）两种类型，前者用于执行计算并指定输出的形式**，后者指定RDD之间的相互依赖关系。

4.两类操作的主要区别是，转换操作（比如map、filter、groupBy、join等）接受RDD并返回RDD，而行动操作（比如count、collect等）接受RDD但是返回非RDD（即输出一个值或结果）。

5.RDD提供的转换接口都非常简单，都是类似map、filter、groupBy、join等粗粒度的数据转换操作，而不是针对某个数据项的细粒度修改。因此，RDD比较适合对于数据集中元素执行相同操作的批处理式应用，而不适合用于需要异步、细粒度状态的应用，比如Web应用系统、增量式的网页爬虫等。

6.RDD采用了惰性调用，即在RDD的执行过程中（如下图所示），真正的计算发生在RDD的**“行动”操作**，对于“行动”之前的所有“转换”操作，Spark只是记录下“转换”操作应用的一些基础数据集以及RDD生成的轨迹，即相互之间的依赖关系，而不会触发真正的计算。

7.例子：
从输入中逻辑上生成A和C两个RDD，经过一系列“转换”操作，逻辑上生成了F（也是一个RDD），之所以说是逻辑上，是因为这时候计算并没有发生，Spark只是记录了RDD之间的生成和依赖关系。
当F要进行输出时，也就是当F进行“行动”操作的时候，Spark才会根据RDD的依赖关系生成DAG，并从起点开始真正的计算。

8.上述这一系列处理称为一个“血缘关系（Lineage），即DAG拓扑排序的结果。
采用惰性调用，通过血缘关系连接起来的一系列RDD操作就可以实现管道化（pipeline），因为这些具有血缘关系的操作都管道化了，一个操作得到的结果不需要保存为中间数据，而是直接管道式地流入到下一个操作进行处理。

9.一个Spark的“Hello World”程序

val sc= new SparkContext(“spark://localhost:7077”,”Hello World”, “YOUR_SPARK_HOME”,”YOUR_APP_JAR”)
val fileRDD = sc.textFile(“hdfs://192.168.0.103:9000/examplefile”)
val filterRDD = fileRDD.filter(_.contains(“Hello World”))
filterRDD.cache()//表示对filterRDD进行持久化，把它保存在内存或磁盘中（这里采用cache接口把数据集保存在内存中），方便后续重复使用
filterRDD.count()

一个Spark应用程序，基本是基于RDD的一系列计算操作。
这个程序的执行过程如下：

• 创建这个Spark程序的执行上下文，即创建SparkContext对象；
• 从外部数据源（即HDFS文件）中读取数据创建fileRDD对象；
• 构建起fileRDD和filterRDD之间的依赖关系，形成DAG图，这时候并没有发生真正的计算，只是记录转换的轨迹；
• 执行到第5行代码时，count()是一个行动类型的操作，触发真正的计算，开始实际执行从fileRDD到filterRDD的转换操作，并把结果持久化到内存中，最后计算出filterRDD中包含的元素个数。

三、RDD特性

1.总体而言，Spark采用RDD以后能够实现高效计算的主要原因如下：
（1）高效的容错性。
现有的分布式共享内存、键值存储、内存数据库等，为了实现容错，必须在集群节点之间进行数据复制或者记录日志，也就是在节点之间会发生大量的数据传输，这对于数据密集型应用而言会带来很大的开销。
在RDD的设计中，数据只读，不可修改，如果需要修改数据，必须从父RDD转换到子RDD，由此在不同RDD之间建立了血缘关系。所以，RDD是一种天生具有容错机制的特殊集合，不需要通过数据冗余的方式（比如检查点）实现容错，而只需通过RDD父子依赖（血缘）关系重新计算得到丢失的分区来实现容错，无需回滚整个系统。
此外，RDD提供的转换操作都是一些粗粒度的操作（比如map、filter和join），RDD依赖关系只需要记录这种粗粒度的转换操作，而不需要记录具体的数据和各种细粒度操作的日志（比如对哪个数据项进行了修改），这就大大降低了数据密集型应用中的容错开销。

（2）中间结果持久化到内存。
数据在内存中的多个RDD操作之间进行传递，不需要“落地”到磁盘上，避免了不必要的读写磁盘开销；

（3）存放的数据可以是Java对象，避免了不必要的对象序列化和反序列化开销。

2.Memory Management
Spark为持久化存储RDD提供三种选择：
（1）内存存储反序列化的java对象；
（2）内存存储序列化的数据。
（3）磁盘存储方式。
第一种选择能够提供最快的性能，由于java虚拟机能够本地化获取RDD。另外一种选择能够让用户选择一个相对于空间受限的java对象图（UML图）而言更加具有内存效率的方法（无需组织java对象，仅仅是使用纯粹的数据对象），可是要以减少性能为代价。第三种方法对于过大而不能保存在RAM中的RDD是实用的。可是重计算会有消耗。
为了管理有限的内存。我们在RDDs的级别使用LRU回收策略。
（参考）

四、RDD之间的依赖关系

1.RDD中的依赖关系分为窄依赖（Narrow Dependency）与宽依赖（Wide Dependency），下图展示了两种依赖之间的区别。

2.窄依赖表现为一个父RDD的分区对应于一个子RDD的分区，或多个父RDD的分区对应于一个子RDD的分区；（一父一子、多父一子，即每个父分区有一个出度）
宽依赖则表现为存在一个父RDD的一个分区对应一个子RDD的多个分区。（一父多子，每个父分区有多个出度）

3.相对而言，在两种依赖关系中，窄依赖的失败恢复更为高效，它只需要根据父RDD分区重新计算丢失的分区即可（不需要重新计算所有分区），而且可以并行地在不同节点进行重新计算。
而对于宽依赖而言，单个节点失效通常意味着重新计算过程会涉及多个父RDD分区，开销较大。

4.此外，Spark还提供了数据检查点和记录日志，用于持久化中间RDD，从而使得在进行失败恢复时不需要追溯到最开始的阶段。在进行故障恢复时，Spark会对数据检查点开销和重新计算RDD分区的开销进行比较，从而自动选择最优的恢复策略。

五、阶段的划分

1.具体划分方法是：在DAG中进行反向解析，遇到宽依赖就断开，遇到窄依赖就把当前的RDD加入到当前的阶段中；将窄依赖尽量划分在同一个阶段中，可以实现流水线计算。

2.把一个DAG图划分成多个“阶段”以后，每个阶段都代表了一组关联的、相互之间没有Shuffle依赖关系的任务组成的任务集合（可以并行执行的任务集）。每个任务集合会被提交给任务调度器（TaskScheduler）进行处理，由任务调度器将任务分发给Executor运行。

六、RDD运行过程

总结一下RDD在Spark架构中的运行过程（如下图所示）：
（1）创建RDD对象；
（2）SparkContext负责计算RDD之间的依赖关系，构建DAG；
（3）DAGScheduler负责把DAG图分解成多个阶段，每个阶段中包含了多个任务，每个任务会被任务调度器分发给各个工作节点（Worker Node）上的Executor去执行。

七、RDD的实现

1.每个RDD都包含：（1）一组RDD分区（partition，即数据集的原子组成部分）；（2）对父RDD的一组依赖，这些依赖描述了RDD的Lineage；（3）一个函数，即在父RDD上执行何种计算；（4）元数据，描述分区模式和数据存放的位置。

2.为什么RDD上需要携带METADATA？
因为依赖多个RDD的变换需要知道是否需要重新整理数据,宽依赖需要，窄依赖不需要。 同时也可以让用户控制更好的局部性和减少整理的工作量。

3.为什么要区分窄依赖和宽依赖？
为了防止失败，窄依赖可以很快的被重计算，掉1-2个PARITION就可以重计算。宽依赖需要一个完整的RDD才可以被重计算。