大数据技术，Spark之RDD，RDD超详细讲解（二）

一、RDD的特性

Spark之所以成为目前比较主流的大数据处理技术，其中RDD的特性和机制占到很大比重，没有RDD的这些机制，Spark性能会大打折扣。总体而言，Spark采用RDD后能够实现高效计算的主要原因有以下几点：

1、高效的容错机制。

现有的分布式共享内存、键值存储、内存数据库等，为了实现容错，必须在集群节点之间进行数据复制（主从复制）或者记录日志。

这就造成各个节点之间大量数据传输，这对于数据密集型应用而言会带来很大的开销。而在设计RDD的过程中，赋予RDD数据只读的能力，不可被修改。如果需要修改数据，必须从父RDD转换到子RDD从新进行计算，由此在不同RDD之间建立了血缘关系。

RDD是一种天生具有容错机制的特殊集合，不需要通过数据冗余的方式实现容错，而只需通过RDD父子依赖（血缘）关系重新计算得到丢失的分区来实现容错。而不需要回滚整个系统，这样就避免了数据复制的高开销。而且重算过程可以在不同节点之间并行进行，实现了高效的容错。

RDD之间的转换

最主要的，RDD提供的转换操作都是一些粗粒度的操作。比如map、filter和join这些转换算子，并不触发真正的计算，只记录他们之间的逻辑关系，只有当遇到count、foreach这类输出算子才触发计算机制。

RDD依赖关系只需要记录这种粗粒度的转换操作，而不需要记录具体的数据和各种细粒度操作的日志，比如对哪个数据项进行了修改，这就大大降低了数据密集型应用中的容错开销。

2、中间结果持久化落地到内存。

数据在内存中的多个RDD操作之间进行传递，不需要“落地”到磁盘上，避免了不必要的读写磁盘的网络和IO开销，有效地提升了数据之间的处理或转换效率。

3、存放的数据可以是Java对象。

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

二、RDD的窄依赖和宽依赖

不同的操作会使得不同RDD中的分区会产生不同的依赖。RDD中的依赖关系分为窄依赖与宽依赖。

1、什么是窄依赖和宽依赖?

窄依赖：一个父RDD的分区对应于一个子RDD的分区，或多个父RDD的分区对应于一个子RDD的分区。最终结果是一个或多个父RDD分区对于一个子RDD分区，是 一或多 对一 的关系。窄依赖典型的操作包括map、filter、union等算子。

窄依赖

宽依赖：一个父RDD的一个分区对应一个子RDD的多个分区。是一个分区对应多个分区的关系，一对多。宽依赖典型的操作包括groupByKey、sortByKey等算子。

宽依赖

总的来说，如果父RDD的一个分区只被一个子RDD的一个分区所使用就是窄依赖，否则就是宽依赖。

特别注意：对于连接Join操作的窄宽依赖的划分可以分为两种情况。

（1）、对输入进行协同划分，属于窄依赖。所谓协同划分是指多个父RDD的某一分区的所有“键”，落在子RDD的同一个分区内，不会产生同一个父RDD的某一分区，落在子RDD的两个分区的情况。

（2）、对输入做非协同划分，属于宽依赖。对于窄依赖的RDD，可以以流水线的方式计算所有父分区，不会造成网络之间的数据混合。对于宽依赖的RDD，则通常伴随着Shuffle操作，即首先需要计算好所有父分区数据，然后在节点之间进行Shuffle。

2、窄依赖与宽依赖的比较

RDD的这种窄依赖和宽依赖的关系设计，使其具有了天生的容错性，大大加快了Spark的执行速度。

RDD数据集通过“血缘关系”记录了它是如何从其它RDD中转换过来的，血缘关系记录的是粗数据粒度的转换操作行为。当这个RDD的部分分区数据丢失时，可以通过血缘关系获取足够的信息来重新运算和恢复丢失的数据分区，由此带来了性能的提升。

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

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

三、RDD的阶段划分

通过分析各个RDD的依赖关系生成了DAG，再通过分析各个RDD中的分区之间的依赖关系来决定如何划分阶段。

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

RDD阶段划分

如上图，从HDFS中读入数据生成3个不同的RDD，即A、C和E。通过一系列转换操作后再将计算结果保存回HDFS。

对DAG进行解析时，在依赖图中进行反向解析。从RDD A到RDD B的转换，以及从RDD B和RDD F到RDD G的转换，都属于宽依赖。

因此，在宽依赖处断开后可以得到三个阶段，即阶段1、阶段2和阶段3。在阶段2中，从map到union都是窄依赖，这两步操作可以形成一个流水线操作。如，分区7通过map操作生成的分区9，可以不用等待分区8到分区10这个转换操作的计算结束，而是继续进行union操作，转换得到分区13，这样流水线执行大大提高了计算的效率。

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

四、Spark的RDD运行过程

通过对RDD概念、依赖关系和阶段划分的介绍，结合之前介绍的Spark运行基本流程，这里再总结一下RDD在Spark架构中的运行过程：

1、创建RDD对象

2、SparkContext负责计算RDD之间的依赖关系，构建DAG；

3、DAGScheduler负责把DAG图分解成多个阶段，每个阶段中包含了多个任务，每个任务会被任务调度器分发给各个工作节点（Worker Node）上的Executor去执行。

一个完整的应用从提交到执行的完整流程：

1，通过spark-submit命令，提交spark应用程序。

2，在driver端会执行代码，首先通过构造的sparkConf，把配置传给sparkContext对象，用于创建sparkContext上下文环境。

3，在遇到action算子时，会生成一个Job。DAGScheduler会将Job划分为多个Stage，每个Stage会创建一个任务集taskSet。

4，把任务集传递给TaskScheduler，TaskScheduler会执行以下过程：

4.1，TaskScheduler会去连接Master向其申请注册Application。Master接收到App的注册请求时，会为其在Worker节点上启动多个Executor。

4.2，Executor启动以后会反向注册到TaskScheduler。Executor会启动线程池，TaskRunner就是把函数反序列化之后通过线程去运行的。Task有两个ShuffleMapTask和ResultTask，只有最后一个Stage是ResultTask，之前所有的都是ShuffleMapTask。

4.3，TaskScheduler接到Executor注册请求后，会把任务集里的每一个任务都提交到Executor上执行。在这里有任务分配的算法，移动计算不移动数据。