關(guān)于RDD, 詳細(xì)可以參考Spark的論文, 下面看下源碼
A Resilient Distributed Dataset (RDD), the basic abstraction in Spark.
Represents an immutable, partitioned collection of elements that can be operated on in parallel.

* Internally, each RDD is characterized by five main properties:
*? - A list of partitions
*? - A function for computing each split
*? - A list of dependencies on other RDDs
*? - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
*? - Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

RDD分為一下幾類,

basic(org.apache.spark.rdd.RDD): This class contains the basic operations available on all RDDs, such as `map`, `filter`, and `persist`.

org.apache.spark.rdd.PairRDDFunctions: contains operations available only on RDDs of key-value pairs, such as `groupByKey` and `join`

org.apache.spark.rdd.DoubleRDDFunctions: contains operations available only on RDDs of Doubles

org.apache.spark.rdd.SequenceFileRDDFunctions: contains operations available on RDDs that can be saved as SequenceFiles

?

RDD首先是泛型類, T表示存放數(shù)據(jù)的類型, 在處理數(shù)據(jù)是都是基于Iterator[T]
以SparkContext和依賴關(guān)系Seq deps為初始化參數(shù)
從RDD提供的這些接口大致就可以知道, 什么是RDD
1. RDD是一塊數(shù)據(jù), 可能比較大的數(shù)據(jù), 所以不能保證可以放在一個(gè)機(jī)器的memory中, 所以需要分成partitions, 分布在集群的機(jī)器的memory
所以自然需要getPartitions, partitioner如果分區(qū), getPreferredLocations分區(qū)如何考慮locality

Partition的定義很簡(jiǎn)單, 只有id, 不包含data

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

2. RDD之間是有關(guān)聯(lián)的, 一個(gè)RDD可以通過(guò)compute邏輯把父RDD的數(shù)據(jù)轉(zhuǎn)化成當(dāng)前RDD的數(shù)據(jù), 所以RDD之間有因果關(guān)系
并且通過(guò)getDependencies, 可以取到所有的dependencies

3. RDD是可以被persisit的, 常用的是cache, 即StorageLevel.MEMORY_ONLY

4. RDD是可以被checkpoint的, 以提高failover的效率, 當(dāng)有很長(zhǎng)的RDD鏈時(shí), 單純的依賴replay會(huì)比較低效

5. RDD.iterator可以產(chǎn)生用于迭代真正數(shù)據(jù)的Iterator[T]

6. 在RDD上可以做各種transforms和actions

abstract class RDD[T: ClassManifest](@transient private var sc: SparkContext, //@transient, 不需要序列化@transient private var deps: Seq[Dependency[_]]) extends Serializable with Logging { /**輔助構(gòu)造函數(shù), 專門用于初始化1對(duì)1依賴關(guān)系的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, 這種情況因?yàn)橹挥幸粋€(gè)parent, 所以直接傳入parent RDD對(duì)象即可
// =======================================================================// 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 StorageLevelif (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 SparkContextsc.persistentRdds(id) = thisthis}/** 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 checkpointedprivate 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的實(shí)現(xiàn)中, 如果存在checkpointRDD, 則可以直接從中讀到RDD數(shù)據(jù), 而不需要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訪問(wèn)數(shù)據(jù)的核心, 在RDD中的Partition中只包含id而沒(méi)有真正數(shù)據(jù)
* 那么如果獲取RDD的數(shù)據(jù)? 參考storage模塊
* 在cacheManager.getOrCompute中, 會(huì)將RDD和Partition id對(duì)應(yīng)到相應(yīng)的block, 并從中讀出數(shù)據(jù)*/ final def iterator(split: Partition, context: TaskContext): Iterator[T] = {if (storageLevel != StorageLevel.NONE) {//StorageLevel不為None,說(shuō)明這個(gè)RDD persist過(guò), 可以直接讀出來(lái)SparkEnv.get.cacheManager.getOrCompute(this, split, context, storageLevel)} else {computeOrReadCheckpoint(split, context) //如果沒(méi)有persisit過(guò), 只有從新計(jì)算出, 或從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中才會(huì)真正調(diào)用runJob, 所以transform都是lazy的sc.runJob(this, (iter: Iterator[T]) => {var result = 0Lwhile (iter.hasNext) {result += 1Liter.next()}result}).sum} // =======================================================================// Other internal methods and fields// ======================================================================= /** Returns the first parent RDD
返回第一個(gè)parent RDD*/protected[spark] def firstParent[U: ClassManifest] = {dependencies.head.rdd.asInstanceOf[RDD[U]]} //................ }

?

這里先只討論一些basic的RDD, pairRDD會(huì)單獨(dú)討論

FilteredRDD

One-to-one Dependency, FilteredRDD

使用FilteredRDD, 將當(dāng)前RDD作為第一個(gè)參數(shù), f函數(shù)作為第二個(gè)參數(shù), 返回值是filter過(guò)后的RDD

/*** 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))

在compute中, 對(duì)parent RDD的Iterator[T]進(jìn)行filter操作

private[spark] class FilteredRDD[T: ClassManifest]( //filter是典型的one-to-one dependency, 使用輔助構(gòu)造函數(shù) prev: RDD[T], //parent RDDf: T => Boolean) //f,過(guò)濾函數(shù)extends RDD[T](prev) {//firstParent會(huì)從deps中取出第一個(gè)RDD對(duì)象, 就是傳入的prev RDD, 在One-to-one Dependency中,parent和child的partition信息相同override def getPartitions: Array[Partition] = firstParent[T].partitionsoverride val partitioner = prev.partitioner // Since filter cannot change a partition's keysoverride def compute(split: Partition, context: TaskContext) =firstParent[T].iterator(split, context).filter(f) //compute就是真正產(chǎn)生RDD的邏輯 }

?

UnionRDD

Range Dependency, 仍然是narrow的

先看看如果使用union的, 第二個(gè)參數(shù)是, 兩個(gè)RDD的array, 返回值就是把這兩個(gè)RDD union后產(chǎn)生的新的RDD

/*** 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))

?

先定義UnionPartition, Union操作的特點(diǎn)是, 只是把多個(gè)RDD的partition合并到一個(gè)RDD中, 而partition本身沒(méi)有變化, 所以可以直接重用parent partition

3個(gè)參數(shù)
idx, partition id, 在當(dāng)前UnionRDD中的序號(hào)
rdd, parent RDD
splitIndex, parent partition的id

private[spark] class UnionPartition[T: ClassManifest](idx: Int, rdd: RDD[T], splitIndex: Int)extends Partition {var split: Partition = rdd.partitions(splitIndex)//從parent RDD中取出相應(yīng)的partition, 重用def iterator(context: TaskContext) = rdd.iterator(split, context)//Iterator也可以重用def preferredLocations() = rdd.preferredLocations(split)override val index: Int = idx//partition id是新的, 因?yàn)槎鄠€(gè)合并后, 序號(hào)肯定會(huì)發(fā)生變化 }

定義UnionRDD

class UnionRDD[T: ClassManifest](sc: SparkContext,@transient var rdds: Seq[RDD[T]]) //parent RDD Seqextends RDD[T](sc, Nil) { // Nil since we implement getDependenciesoverride def getPartitions: Array[Partition] = {val array = new Array[Partition](rdds.map(_.partitions.size).sum) //UnionRDD的partition數(shù),是所有parent RDD中的partition數(shù)目的和var pos = 0for (rdd <- rdds; split <- rdd.partitions) {array(pos) = new UnionPartition(pos, rdd, split.index) //創(chuàng)建所有的UnionPartitionpos += 1}array}override def getDependencies: Seq[Dependency[_]] = {val deps = new ArrayBuffer[Dependency[_]]var pos = 0for (rdd <- rdds) { deps += new RangeDependency(rdd, 0, pos, rdd.partitions.size)//創(chuàng)建RangeDependencypos += rdd.partitions.size)//由于是RangeDependency, 所以pos的遞增是加上整個(gè)區(qū)間size}deps}override def compute(s: Partition, context: TaskContext): Iterator[T] =s.asInstanceOf[UnionPartition[T]].iterator(context)//Union的compute非常簡(jiǎn)單,什么都不需要做override def getPreferredLocations(s: Partition): Seq[String] =s.asInstanceOf[UnionPartition[T]].preferredLocations() }

轉(zhuǎn)載于:https://www.cnblogs.com/fxjwind/p/3489107.html

總結(jié)

以上是生活随笔為你收集整理的Spark 源码分析 -- RDD的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。