Spark 中 RDD的运行机制

1. rdd 的设计与运行原理

spark 的核心是建立在统一的抽象 rdd 之上，基于 rdd 的转换和行动操作使得 spark 的各个组件可以无缝进行集成，从而在同一个应用程序中完成大数据计算任务。

在实际应用中，存在许多迭代式算法和交互式数据挖掘工具，这些应用场景的共同之处在于不同计算阶段之间会重用中间结果，即一个阶段的输出结果会作为下一个阶段的输入。而 hadoop 中的 mapreduce 框架都是把中间结果写入到 hdfs 中，带来了大量的数据复制、磁盘 io 和序列化开销，并且通常只支持一些特定的计算模式。而 rdd 提供了一个抽象的数据架构，从而让开发者不必担心底层数据的分布式特性，只需将具体的应用逻辑表达为一系列转换处理，不同 rdd 之间的转换操作形成依赖关系，可以实现管道化，从而避免了中间结果的存储，大大降低了数据复制、磁盘 io 和序列化开销。

1.1. rdd 概念

一个 rdd 就是一个分布式对象集合，提供了一种高度受限的共享内存模型，其本质上是一个只读的分区记录集合，不能直接修改。每个 rdd 可以分成多个分区，每个分区就是一个数据集片段，并且一个 rdd 的不同分区可以保存到集群中不同的节点上，从而可以在集群中的不同节点上进行并行计算。

rdd 提供了一组丰富的操作以支持常见的数据运算，分为“行动”（action）和“转换”（transformation）两种类型，前者用于执行计算并指定输出的形式，后者指定 rdd 之间的相互依赖关系。rdd 提供的转换接口都非常简单，都是类似 map 、filter 、groupby 、join 等粗粒度的数据转换操作，而不是针对某个数据项的细粒度修改。因此，rdd 比较适合对于数据集中元素执行相同操作的批处理式应用，而不适合用于需要异步、细粒度状态的应用，比如 web 应用系统、增量式的网页爬虫等。

rdd 的典型的执行过程如下：

1. 读入外部的数据源（或者内存中的集合）进行 rdd 创建；
2. rdd 经过一系列的 “转换” 操作，每一次都会产生不同的 rdd，供给下一个转换使用；
3. 最后一个 rdd 经过 “行动” 操作进行处理，并输出指定的数据类型和值。

rdd 采用了惰性调用，即在 rdd 的执行过程中，所有的转换操作都不会执行真正的操作，只会记录依赖关系，而只有遇到了行动操作，才会触发真正的计算，并根据之前的依赖关系得到最终的结果。

下面以一个实例来描述 rdd 的实际执行过程，如下图所示，开始从输入中创建了两个 rdd，分别是 a 和 c，然后经过一系列的转换操作，最终生成了一个 f，这也是一个 rdd。注意，这些转换操作的执行过程中并没有执行真正的计算，基于创建的过程也没有执行真正的计算，而只是记录的数据流向轨迹。当 f 执行了行为操作并生成输出数据时，spark 才会根据 rdd 的依赖关系生成有向无环图（dag），并从起点开始执行真正的计算。正是 rdd 的这种惰性调用机制，使得转换操作得到的中间结果不需要保存，而是直接管道式的流入到下一个操作进行处理。

1.2. rdd 特性

总体而言，spark 采用 rdd 以后能够实现高效计算的主要原因如下：

1. 高效的容错性。在 rdd 的设计中，只能通过从父 rdd 转换到子 rdd 的方式来修改数据，这也就是说我们可以直接利用 rdd 之间的依赖关系来重新计算得到丢失的分区，而不需要通过数据冗余的方式。而且也不需要记录具体的数据和各种细粒度操作的日志，这大大降低了数据密集型应用中的容错开销。

2. 中间结果持久化到内存。数据在内存中的多个 rdd 操作之间进行传递，不需要在磁盘上进行存储和读取，避免了不必要的读写磁盘开销；

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

1.3. rdd 之间的依赖关系

rdd 中的不同的操作会使得不同 rdd 中的分区会产生不同的依赖关系，主要分为窄依赖（narrow dependency）与宽依赖（wide dependency）。其中，窄依赖表示的是父 rdd 和子 rdd 之间的一对一关系或者多对一关系，主要包括的操作有 map、filter、union 等；而宽依赖则表示父 rdd 与子 rdd 之间的一对多关系，即一个父 rdd 转换成多个子 rdd，主要包括的操作有 groupbykey、sortbykey 等。

对于窄依赖的 rdd，可以以流水线的方式计算所有父分区，不会造成网络之间的数据混合。对于宽依赖的 rdd，则通常伴随着 shuffle 操作，即首先需要计算好所有父分区数据，然后在节点之间进行 shuffle。因此，在进行数据恢复时，窄依赖只需要根据父 rdd 分区重新计算丢失的分区即可，而且可以并行地在不同节点进行重新计算。而对于宽依赖而言，单个节点失效通常意味着重新计算过程会涉及多个父 rdd 分区，开销较大。此外，spark 还提供了数据检查点和记录日志，用于持久化中间 rdd，从而使得在进行失败恢复时不需要追溯到最开始的阶段。在进行故障恢复时，spark 会对数据检查点开销和重新计算 rdd 分区的开销进行比较，从而自动选择最优的恢复策略。

1.4. 阶段的划分

spark 通过分析各个 rdd 的依赖关系生成了 dag ，再通过分析各个 rdd 中的分区之间的依赖关系来决定如何划分阶段，具体划分方法是：在 dag 中进行反向解析，遇到宽依赖就断开，遇到窄依赖就把当前的 rdd 加入到当前的阶段中；将窄依赖尽量划分在同一个阶段中，可以实现流水线计算。例如在下图中，首先根据数据的读取、转化和行为等操作生成 dag。然后在执行行为操作时，反向解析 dag，由于从 a 到 b 的转换和从 b、f 到 g 的转换都属于宽依赖，则需要从在宽依赖处进行断开，从而划分为三个阶段。把一个 dag 图划分成多个 “阶段” 以后，每个阶段都代表了一组关联的、相互之间没有 shuffle 依赖关系的任务组成的任务集合。每个任务集合会被提交给任务调度器（taskscheduler）进行处理，由任务调度器将任务分发给 executor 运行。

1.5. rdd 运行过程

通过上述对 rdd 概念、依赖关系和阶段划分的介绍，结合之前介绍的 spark 运行基本流程，这里再总结一下 rdd 在 spark 架构中的运行过程（如下图所示）：

1. 创建 rdd 对象；
2. sparkcontext 负责计算 rdd 之间的依赖关系，构建 dag；
3. dagschedule 负责把 dag 图反向解析成多个阶段，每个阶段中包含多个任务，每个任务会被任务调度器分发给工作节点上的 executor 上执行。