Spark 源码分析 -- RDD

trait Partition extends Serializable {
/**
* Get the split's index within its parent RDD
*/
def index: Int
// A better default implementation of HashCode
override def hashCode(): Int = index
}

abstract class RDD[T: ClassManifest](
@transient private var sc: SparkContext, //@transient, 不需要序列化
@transient private var deps: Seq[Dependency[_]]
) extends Serializable with Logging {

/**辅助构造函数, 专门用于初始化1对1依赖关系的RDD,这种还是很多的, filter, map...

Construct an RDD with just a one-to-one dependency on one parent */
def this(@transient oneParent: RDD[_]) =
this(oneParent.context , List(new OneToOneDependency(oneParent)))// 不同于一般的RDD, 这种情况因为只有一个parent, 所以直接传入parent RDD对象即可

// =======================================================================
// Methods that should be implemented by subclasses of RDD
// =======================================================================
/** Implemented by subclasses to compute a given partition. */
def compute(split: Partition, context: TaskContext): Iterator[T]

/**
* Implemented by subclasses to return the set of partitions in this RDD. This method will only
* be called once, so it is safe to implement a time-consuming computation in it.
*/
protected def getPartitions: Array[Partition]

/**
* Implemented by subclasses to return how this RDD depends on parent RDDs. This method will only
* be called once, so it is safe to implement a time-consuming computation in it.
*/
protected def getDependencies: Seq[Dependency[_]] = deps

/** Optionally overridden by subclasses to specify placement preferences. */
protected def getPreferredLocations(split: Partition): Seq[String] = Nil

/** Optionally overridden by subclasses to specify how they are partitioned. */
val partitioner: Option[Partitioner] = None

// =======================================================================
// Methods and fields available on all RDDs
// =======================================================================

/** The SparkContext that created this RDD. */
def sparkContext: SparkContext = sc

/** A unique ID for this RDD (within its SparkContext). */
val id: Int = sc.newRddId()

/** A friendly name for this RDD */
var name: String = null

/**
* Set this RDD's storage level to persist its values across operations after the first time
* it is computed. This can only be used to assign a new storage level if the RDD does not
* have a storage level set yet..
*/
def persist(newLevel: StorageLevel): RDD[T] = {
// TODO: Handle changes of StorageLevel
if (storageLevel != StorageLevel.NONE && newLevel != storageLevel) {
throw new UnsupportedOperationException(
"Cannot change storage level of an RDD after it was already assigned a level")
}
storageLevel = newLevel
// Register the RDD with the SparkContext
sc.persistentRdds(id) = this
this
}

/** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
def persist(): RDD[T] = persist(StorageLevel.MEMORY_ONLY)

/** Persist this RDD with the default storage level (`MEMORY_ONLY`). */
def cache(): RDD[T] = persist()

/** Get the RDD's current storage level, or StorageLevel.NONE if none is set. */
def getStorageLevel = storageLevel

// Our dependencies and partitions will be gotten by calling subclass's methods below, and will
// be overwritten when we're checkpointed
private var dependencies_ : Seq[Dependency[_]] = null
@transient private var partitions_ : Array[Partition] = null

/** An Option holding our checkpoint RDD, if we are checkpointed
* checkpoint就是把RDD存到磁盘文件中, 以提高failover的效率, 虽然也可以选择replay
* 并且在RDD的实现中, 如果存在checkpointRDD, 则可以直接从中读到RDD数据, 而不需要compute */
private def checkpointRDD: Option[RDD[T]] = checkpointData.flatMap(_.checkpointRDD)

/**
* Internal method to this RDD; will read from cache if applicable, or otherwise compute it.
* This should ''not'' be called by users directly, but is available for implementors of custom
* subclasses of RDD.
*/

/** 这是RDD访问数据的核心, 在RDD中的Partition中只包含id而没有真正数据
* 那么如果获取RDD的数据? 参考storage模块
* 在cacheManager.getOrCompute中, 会将RDD和Partition id对应到相应的block, 并从中读出数据*/
final def iterator(split: Partition, context: TaskContext): Iterator[T] = {
if (storageLevel != StorageLevel.NONE) {//StorageLevel不为None,说明这个RDD persist过, 可以直接读出来
SparkEnv.get.cacheManager.getOrCompute(this, split, context, storageLevel)
} else {
computeOrReadCheckpoint(split, context) //如果没有persisit过, 只有从新计算出, 或从checkpoint中读出
}
}

// Transformations (return a new RDD)
//...... 各种transformations的接口,map, union...

/**
* Return a new RDD by applying a function to all elements of this RDD.
*/
def map[U: ClassManifest](f: T => U): RDD[U] = new MappedRDD(this, sc.clean(f))

// Actions (launch a job to return a value to the user program)
//......各种actions的接口,count, collect...

/**
* Return the number of elements in the RDD.
*/
def count(): Long = {// 只有在action中才会真正调用runJob, 所以transform都是lazy的
sc.runJob(this, (iter: Iterator[T]) => {
var result = 0L
while (iter.hasNext) {
result += 1L
iter.next()
}
result
}).sum
}

// =======================================================================
// Other internal methods and fields
// =======================================================================

/** Returns the first parent RDD
返回第一个parent RDD*/
protected[spark] def firstParent[U: ClassManifest] = {
dependencies.head.rdd.asInstanceOf[RDD[U]]
}

//................
}

FilteredRDD

/**
* Return a new RDD containing only the elements that satisfy a predicate.
*/
def filter(f: T => Boolean): RDD[T] = new FilteredRDD(this, sc.clean(f))

private[spark] class FilteredRDD[T: ClassManifest]( //filter是典型的one-to-one dependency, 使用辅助构造函数
prev: RDD[T], //parent RDD
f: T => Boolean) //f,过滤函数
extends RDD[T](prev) {
//firstParent会从deps中取出第一个RDD对象, 就是传入的prev RDD, 在One-to-one Dependency中,parent和child的partition信息相同
override def getPartitions: Array[Partition] = firstParent[T].partitions

override val partitioner = prev.partitioner// Since filter cannot change a partition's keys

override def compute(split: Partition, context: TaskContext) =
firstParent[T].iterator(split, context).filter(f) //compute就是真正产生RDD的逻辑
}

UnionRDD

/**
* Return the union of this RDD and another one. Any identical elements will appear multiple
* times (use `.distinct()` to eliminate them).
*/
def union(other: RDD[T]): RDD[T] = new UnionRDD(sc, Array(this, other))

private[spark] class UnionPartition[T: ClassManifest](idx: Int, rdd: RDD[T], splitIndex: Int)
extends Partition {

var split: Partition = rdd.partitions(splitIndex)//从parent RDD中取出相应的partition, 重用

def iterator(context: TaskContext) = rdd.iterator(split, context)//Iterator也可以重用

def preferredLocations() = rdd.preferredLocations(split)

override val index: Int = idx//partition id是新的, 因为多个合并后, 序号肯定会发生变化
}

class UnionRDD[T: ClassManifest](
sc: SparkContext,
@transient var rdds: Seq[RDD[T]])//parent RDD Seq
extends RDD[T](sc, Nil) {// Nil since we implement getDependencies

override def getPartitions: Array[Partition] = {
val array = new Array[Partition](rdds.map(_.partitions.size).sum) //UnionRDD的partition数,是所有parent RDD中的partition数目的和
var pos = 0
for (rdd <- rdds; split <- rdd.partitions) {
array(pos) = new UnionPartition(pos, rdd, split.index) //创建所有的UnionPartition
pos += 1
}
array
}

override def getDependencies: Seq[Dependency[_]] = {
val deps = new ArrayBuffer[Dependency[_]]
var pos = 0
for (rdd <- rdds) {
deps += new RangeDependency(rdd, 0, pos, rdd.partitions.size)//创建RangeDependency
pos += rdd.partitions.size)//由于是RangeDependency, 所以pos的递增是加上整个区间size
}
deps
}

override def compute(s: Partition, context: TaskContext): Iterator[T] =
s.asInstanceOf[UnionPartition[T]].iterator(context)//Union的compute非常简单,什么都不需要做

override def getPreferredLocations(s: Partition): Seq[String] =
s.asInstanceOf[UnionPartition[T]].preferredLocations()
}