深入理解Spark RDD——RDD实现的初次分析

RDD（Resilient Distributed Datasets，弹性分布式数据集）代表可并行操作元素的不可变分区集合。对于Spark的初学者来说，这个概念会十分陌生。即便是对于一些有Spark使用经验的人，要想说清楚什么是RDD，以及为什么需要RDD还是一件比较困难的事情。在《深入理解Spark RDD——为什么需要RDD?》一文解释了第二个问题，本文将开启对第一个问题的解答。

有些读者可能对本文的标题感到困惑，这是因为RDD的API非常多，所以本文首先对RDD中与调度系统息息相关的API方法进行分析，转换API、动作API及检查点API将在后续文章中进行介绍。

抽象类RDD定义了所有RDD的规范，我们从RDD的属性开始，逐步了解RDD的实现。

_sc：即SparkContext。_sc由@transient修饰，所以此属性不会被序列化。
deps：构造器参数之一，是Dependency的序列，用于存储当前RDD的依赖。RDD的子类在实现时不一定会传递此参数。由于deps由@transient修饰，所以此属性不会被序列化。
partitioner：当前RDD的分区计算器。partitioner由@transient修饰，所以此属性不会被序列化。
id：当前RDD的唯一身份标识。此属性通过调用SparkContext的nextRddId属性生成。
name：RDD的名称。name由@transient修饰，所以此属性不会被序列化。
dependencies_：与deps相同，但是可以被序列化。
partitions_：存储当前RDD的所有分区的数组。partitions_由@transient修饰，所以此属性不会被序列化。
storageLevel：当前RDD的存储级别。
creationSite：创建当前RDD的用户代码。creationSite由@transient修饰，所以此属性不会被序列化。
scope：当前RDD的操作作用域。scope由@transient修饰，所以此属性不会被序列化。
checkpointData：当前RDD的检查点数据。
checkpointAllMarkedAncestors：是否对所有标记了需要保存检查点的祖先保存检查点。
doCheckpointCalled：是否已经调用了doCheckpoint方法设置检查点。此属性可以阻止对RDD多次设置检查点。

RDD采用了模板方法模式的设计，抽象类RDD中定义了模板方法以及一些未实现的接口，这些接口将需要RDD的各个子类分别实现。下面先来介绍RDD中定义的接口。

compute：对RDD的分区进行计算。此方法的定义如下：

  @DeveloperApidef compute(split: Partition, context: TaskContext): Iterator[T]

getPartitions：获取当前RDD的所有分区。此方法的定义如下：

  protected def getPartitions: Array[Partition]

getDependencies：获取当前RDD的所有依赖。此方法的定义如下：

  protected def getDependencies: Seq[Dependency[_]] = deps

getPreferredLocations：获取某一分区的偏好位置。此方法的定义如下：

  protected def getPreferredLocations(split: Partition): Seq[String] = Nil

RDD中除定义了以上接口外，还实现了一些模板方法。

partitions

partitions方法（见代码清单1）用于获取RDD的分区数组。

代码清单1 partitions的实现

  final def partitions: Array[Partition] = {checkpointRDD.map(_.partitions).getOrElse {if (partitions_ == null) {partitions_ = getPartitions// 省略次要代码}partitions_}}

根据代码清单1，partitions方法查找分区数组的优先级为：从CheckPoint查找 > 读取partitions_属性 > 调用getPartitions方法获取。检查点的内容将在10.3节详细介绍。

preferredLocations

preferredLocations方法（见代码清单2）优先调用CheckPoint中保存的RDD的getPreferredLocations方法获取指定分区的偏好位置，当没有保存CheckPoint时调用自身的getPreferredLocations方法获取指定分区的偏好位置。

代码清单2 preferredLocations的实现

  final def preferredLocations(split: Partition): Seq[String] = {checkpointRDD.map(_.getPreferredLocations(split)).getOrElse {getPreferredLocations(split)}}

dependencies

dependencies方法（见代码清单3）用于获取当前RDD的所有依赖的序列。

代码清单3 dependencies的实现

  final def dependencies: Seq[Dependency[_]] = {checkpointRDD.map(r => List(new OneToOneDependency(r))).getOrElse {if (dependencies_ == null) {dependencies_ = getDependencies}dependencies_}}

根据代码清单3，dependencies方法的执行步骤如下。

从CheckPoint中获取RDD并将这些RDD封装为OneToOneDependency列表。如果从CheckPoint中获取到RDD的依赖，则返回RDD的依赖，否则进入下一步。
如果dependencies_等于null，那么调用子类实现的getDependencies方法获取当前RDD的依赖后赋予dependencies，最后返回dependencies_。

其他方法

除了以上的模板方法，RDD中还实现的方法如下：

（1）context

context方法（见代码清单4）实际返回了_sc（即SparkContext）。

代码清单4 context的实现

  def context: SparkContext = sc

（2）getStorageLevel

getStorageLevel方法（见代码清单5）实际返回了当前RDD的StorageLevel。

代码清单5 获取当前RDD的存储级别

  def getStorageLevel: StorageLevel = storageLevel

（3）getNarrowAncestors

getNarrowAncestors方法（见代码清单6）用于获取当前RDD的祖先依赖中属于窄依赖的RDD序列。

代码清单6 getNarrowAncestors的实现

  private[spark] def getNarrowAncestors: Seq[RDD[_]] = {val ancestors = new mutable.HashSet[RDD[_]]def visit(rdd: RDD[_]) {val narrowDependencies = rdd.dependencies.filter(_.isInstanceOf[NarrowDependency[_]])val narrowParents = narrowDependencies.map(_.rdd)val narrowParentsNotVisited = narrowParents.filterNot(ancestors.contains)narrowParentsNotVisited.foreach { parent =>ancestors.add(parent)visit(parent)}}visit(this)ancestors.filterNot(_ == this).toSeq}

如要继续了解RDD，请继续阅读《深入理解Spark RDD——RDD依赖（构建DAG的关键）》。

深入理解Spark RDD系列文章：

《深入理解Spark RDD——为什么需要RDD?》

《深入理解Spark RDD——RDD实现的初次分析》

《深入理解Spark RDD——RDD依赖（构建DAG的关键）》

《深入理解Spark RDD——RDD分区计算器Partitioner》

《深入理解Spark RDD——RDD信息对象》