1.1RDD解读

Posted 2020-06-20 一路向前走

1.RDD（Resilient Distributed DataSet）是Spark生态系统中最基本的抽象，代表不可变的、可并行操作的分区元素集合。RDD这个类有RDD系列所有基本的操作，比如map、filter、persist.另外，org.apache.spark.rdd.PairRDDFunctions含有key-value类型的RDD的基本操作，比如groupby、join；org.apache.spark.rdd.DoubleRDDFunctions含有Double类型的RDD的基本操作;org.apache.spark.rdd.SequenceFileRDDFunctions含有可以将RDD保存SequenceFiles的基本操作。所有的操作会通过有隐式转换适用于任何RDD。

       每个RDD的5个主要属性：

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

2.重要方法解读

（1）//注册一个新的RDD，并根据当前值加1返回它的RDD的ID
private[spark] def newRddId(): Int = nextRddId.getAndIncrement()

（2）缓存相关

             a）persist、cache

/**
 * 指定RDD缓存的Level，详见StorageLevel object
 *
 * @param newLevel 缓存Level

 * @param allowOverride 是否重写缓存
 */
private def persist(newLevel: StorageLevel, allowOverride: Boolean): this.type = {
  if (storageLevel != StorageLevel.NONE && newLevel != storageLevel && !allowOverride) {
    throw new UnsupportedOperationException(
      "Cannot change storage level of an RDD after it was already assigned a level")
  }
  if (storageLevel == StorageLevel.NONE) {
    sc.cleaner.foreach(_.registerRDDForCleanup(this))
    sc.persistRDD(this)
  }
  storageLevel = newLevel
  this
}

//可见cache其实是调用的persist方法，RDD默认的缓存策略是MEMORY_ONLY

def cache(): this.type = persist()

def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

            b）unpersist

//将RDD设置为不缓存，并且把内存或磁盘上的blocks都删除

def unpersist(blocking: Boolean = true): this.type = {
  logInfo("Removing RDD " + id + " from persistence list")
  sc.unpersistRDD(id, blocking)

  //将缓存Level设置为NONE
  storageLevel = StorageLevel.NONE
  this
}

unpersistRDD（id，blocking）的源码如下所示：

/**
 * 将内存或磁盘中缓存的RDD删除

 */
private[spark] def unpersistRDD(rddId: Int, blocking: Boolean = true) {
  env.blockManager.master.removeRdd(rddId, blocking)

  //persistentRdds是一个弱引用得HashMap，key为rddId，value为对应的RDD
  persistentRdds.remove(rddId)
  listenerBus.post(SparkListenerUnpersistRDD(rddId))
}

（3）分区partitions

//得到RDD的所有分区，并以数组形式返回

final def partitions: Array[Partition] = {
  checkpointRDD.map(_.partitions).getOrElse {
    if (partitions_ == null) {
      partitions_ = getPartitions
    }
    partitions_
  }
}

（4）

//得到分区预先存放的位置

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

（5）依赖

//得到窄依赖的祖先节点

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)
    }
  }