spark RDD 结构最详解

我们都知道RDD是弹性分布数据集，但是弹性的分布数据集是什么呢？

一种简单的解释RDD是横向多分区的（这个数据集包括许多接口），纵向当计算过程中内存不足可刷写到磁盘等外存上，可与外存进行灵活的数据交换。

而另一种解释是RDD是由虚拟数据结构组成，并不包含真实数据本体，RDD使用了一种“血统”的容错机制，当数据发生丢失时，可以同时父节点计算复原。

在结构更新和丢失后可随时根据血统进行数据模型的重建。所谓“分布式”，就是可以分布在多台机器上进行并行计算。

RDD结构

RDD.png

RDD是一个只读的有属性的数据集。属性用来描述当前数据集的状态，数据集是由数据的分区（partition）组成，并（由block）映射成真实数据。RDD属性包括名称、分区类型、父RDD指针、数据本地化、数据依赖关系等，主要属性可以分为3类：

• Parents 表示的是与其他RDD 的关系
• partitioner,checkpoint,storagelevel,iterator数据相关
• rddname,sparkcontext,sparkconfRDD自身属性之后RDD源码解析里详细介绍每个属性。

下面来详细介绍下各个模块

1.rddname

即rdd的名称

2.sparkcontext

SparkContext为Spark job的入口，由Spark driver创建在client端，包括集群连接，RddID，创建抽样，累加器，广播变量等信息。

3.sparkconf配置信息，即sc.conf

Spark参数配置信息 提供三个位置用来配置系统： Spark api：控制大部分的应用程序参数，可以用SparkConf对象或者Java系统属性设置 环境变量：可以通过每个节点的conf/spark-env.sh脚本设置。例如IP地址、端口等信息 日志配置：可以通过log4j.properties配置

4.parent

指向依赖父RDD的partition id，利用dependencies方法可以查找该RDD所依赖的partiton id的List集合，即上图中的parents。

5.iterator

迭代器，用来查找当前RDD Partition与父RDD中Partition的血缘关系。并通过StorageLevel确定迭代位置，直到确定真实数据的位置。迭代方式分为checkpoint迭代和RDD迭代， 如果StorageLevel为NONE则执行computeOrReadCheckpoint计算并获取数据，此方法也是一个迭代器，迭代checkpoint数据存放位置，迭代出口为找到真实数据或内存。如果Storagelevel不为空，根据存储级别进入RDD迭代器，继续迭代父RDD的结构，迭代出口为真实数据或内存。迭代器内部有数据本地化判断，先从本地获取数据，如果没有则远程查找。

6.prisist

rdd存储的level，即通过storagelevel和是否可覆盖判断， storagelevel分为 5中状态 ，useDisk, useMemory, useOffHeap, deserialized, replication 可组合使用。

7.partitioner 分区方式

RDD的分区方式。RDD的分区方式主要包含两种（Hash和Range），这两种分区类型都是针对K-V类型的数据。如是非K-V类型，则分区为None。 Hash是以key作为分区条件的散列分布，分区数据不连续，极端情况也可能散列到少数几个分区上，导致数据不均等；Range按Key的排序平衡分布，分区内数据连续，大小也相对均等。

8.checkpoint

Spark提供的一种缓存机制，当需要计算的RDD过多时，为了避免重新计算之前的RDD，可以对RDD做checkpoint处理，检查RDD是否被物化或计算，并将结果持久化到磁盘或HDFS。与spark提供的另一种缓存机制cache相比， cache缓存数据由executor管理，当executor消失了，被cache的数据将被清除，RDD重新计算，而checkpoint将数据保存到磁盘或HDFS，job可以从checkpoint点继续计算。

9.storageLevel

一个枚举类型，用来记录RDD的存储级别。存储介质主要包括内存、磁盘和堆外内存，另外还包含是否序列化操作以及副本数量。如：MEMORY_AND_DISK_SER代表数据可以存储在内存和磁盘，并且以序列化的方式存储。是判断数据是否保存磁盘或者内存的条件。

窄依赖与宽依赖

窄依赖：父RDD中，每个分区内的数据，都只会被子RDD中特定的分区所消费，为窄依赖：例如map、filter、union等操作会产生窄依赖

宽依赖：父RDD中，分区内的数据，会被子RDD内多个分区消费，则为宽依赖：例如 groupByKey、reduceByKey、sortByKey等操作会产生宽依赖，会产生shuffle

join操作有两种情况：如果两个RDD在进行join操作时，一个RDD的partition仅仅和另一个RDD中已知个数的Partition进行join，那么这种类型的join操作就是窄依赖，例如图1中左半部分的join操作(join with inputsco-partitioned)；其它情况的join操作就是宽依赖,例如图1中右半部分的join操作(join with inputsnot co-partitioned)，由于是需要父RDD的所有partition进行join的转换，这就涉及到了shuffle，因此这种类型的join操作也是宽依赖。

那么为什么Spark要将依赖分成这两种呢？

首先，从计算过程来看，窄依赖是数据以管道方式经一系列计算操作可以运行在了一个集群节点上，如（map、filter等），宽依赖则可能需要将数据通过跨节点传递后运行（如groupByKey），有点类似于MR的shuffle过程。 其次，从失败恢复来看，窄依赖的失败恢复起来更高效，因为它只需找到父RDD的一个对应分区即可，而且可以在不同节点上并行计算做恢复；宽依赖则牵涉到父RDD的多个分区，恢复起来相对复杂些。 综上， 这里引入了一个新的概念Stage。Stage可以简单理解为是由一组RDD组成的可进行优化的执行计划。如果RDD的衍生关系都是窄依赖，则可放在同一个Stage中运行，若RDD的依赖关系为宽依赖，则要划分到不同的Stage。这样Spark在执行作业时，会按照Stage的划分, 生成一个完整的最优的执行计划。下面引用一张比较流行的图片辅助大家理解Stage，如图RDD¬-A到RDD-B和RDD-F到RDD-G均属于宽依赖，所以与前面的父RDD划分到了不同的Stage中。

RDD算子

用来生成或处理RDD的方法叫做RDD算子。RDD算子就是一些方法，在Spark框架中起到运算符的作用。

在spark计算框架有自己的运算单位（RDD）和自己的运算符（RDD算子）。

Spark算子非常丰富，有几十个，开发者把算子组合使用，从一个基础的RDD计算出想要的结果。

RDDsuan.png

• 图中的RDD dependency正是RDD结构中的private var deps: Seq[Dependency[_]]，dependency类被两个类继承，NarrowDependency（窄依赖）和ShuffleDependency（宽依赖）。窄依赖又分onetoonedependency和rangedependency，这是窄依赖提供的2种抽样方式1对1数据抽样和平衡数据抽样，返回值都是一个partitonid的list集合。
• 第二层，是提供RDD底层计算的基本算法，继承了RDD，并实现了dependency的一种或多种依赖关系的计算逻辑，并互相调用实现更复杂的功能。
• 最下层是Spark API，利用RDD基本的计算实现RDD所有的算子，并调用多个底层RDD算子实现复杂的功能。 右边的泛型，是scala的一种类型，可以理解为类的泛型，泛指编译时被抽象的类型。Spark利用scala的这一特性把依赖关系抽象成一种泛型结构，并不需要真实的数据类型参与编译过程。编译的结构类由序列化和反序列化到集群的计算节点取数并计算。
• Transformation：转换算子，这类转换并不触发提交作业，完成作业中间过程处理。Transformation按照数据类型又分为两种，value数据类型算子和key-value数据类型算子。
• 1） Value数据类型的Transformation算子 Map，flatMap，mapPartitions，glom，union，cartesian，groupBy，filter，distinct，subtract，sample，takeSample
• 2）Key-Value数据类型的Transfromation算子 mapValues，combineByKey，reduceByKey，partitionBy，cogroup，join，leftOuterJoin和rightOuterJoin Action： 行动算子，这类算子会触发SparkContext提交Job作业。
• Action算子是用来整合和输出数据的，主要包括以下几种： Foreach，HDFS，saveAsTextFile，saveAsObjectFile， collect，collectAsMap，reduceByKeyLocally，lookup，count，top，reduce，fold，aggregate