3.4 RDD的计算

3.4 RDD的计算

3.4.1 Ta s k简介

原始的RDD经过一系列转换后，会在最后一个RDD上触发一个动作，这个动作会生成一个Job。在Job被划分为一批计算任务（Task）后，这批Task会被提交到集群上的计算节点去计算。计算节点执行计算逻辑的部分称为Executor。Executor在准备好Task的运行时环境后，会通过调用org.apache.spark.scheduler.Task#run来执行计算。Spark的Task分为两种：

1）org.apache.spark.scheduler.ShuffleMapTask

2）org.apache.spark.scheduler.ResultTask

简单来说，DAG的最后一个阶段会为每个结果的Partition生成一个ResultTask，其余所有的阶段都会生成ShuffleMapTask。生成的Task会被发送到已经启动的Executor上，由Executor来完成计算任务的执行，执行过程的实现在org.apache. spark.executor.Executor.TaskRunner#run。第6章会介绍这一部分的实现原理和设计思想。

3.4.2 Task的执行起点

org.apache.spark.scheduler.Task#run会 调 用ShuffleMapTask或 者ResultTask的runTask；runTask会调用RDD的org.apache.spark.rdd.RDD#iterator。计算由此开始。

final def iterator(split: Partition, context: TaskContext): Iterator[T] = {

if(storageLevel != StorageLevel.NONE) {

//如果存储级别不是NONE，那么先检查是否有缓存；没有缓存则要进行计算

SparkEnv.get.cacheManager.getOrCompute(this, split, context,storageLevel)

} else {

//如果有checkpoint，那么直接读取结果；否则直接进行计算

computeOrReadCheckpoint(split, context)

}

}

其中，SparkEnv中包含了一个运行时节点所需要的所有的环境信息。cache-Manager是org.apache.spark.CacheManager，它负责调用BlockManager来管理RDD的缓存，如果当前RDD原来计算过并且把结果缓存起来，那么接下来的运算都可以通过BlockManager来直接读取缓存后返回。SparkEnv除了cacheManager，还包括以下重要的成员变量：

1）akka.actor.ActorSystem：运行在该节点的Actor System，其中运行在Driver上的名字是sparkDriver；运行在Executor上的是sparkExecutor。

2）org.apache.spark.serializer.Serializer：序列化和发序列化的工具。

3）org.apache.spark.MapOutputTracker；保存Shuffle Map Task输出的位置信息。其中在Driver上的Tracer是org.apache.spark.MapOutputTrackerMaster；而在Executor上的Tracker是org.apache.spark.MapOutputTrackerWorker，它会从org.apache.spark. MapOutputTrackerMaster获取信息。

4）org.apache.spark.shuffle.ShuffleManager：Shuffle的管理者，其中Driver端会注册Shuffle的信息，而Executor端会上报和获取Shuffle的信息。现阶段内置支持Hash Based Shuffle和Sort Based Shuffle，具体实现细节请参阅第7章。

5）org.apache.spark.broadcast.BroadcastManager：广播变量的管理者。

6）org.apache.spark.network.BlockTransferService：Executor读取Shuffle数据的Client。当前支持netty和nio，可以通过spark.shuffle.blockTransferService来设置。具体详情可以参阅第7章。

7）org.apache.spark.storage.BlockManager：提供了Storage模块与其他模块的交互接口，管理Storage模块。

8）org.apache.spark.SecurityManager：Spark对于认证授权的实现。

9）org.apache.spark.HttpFileServer：可以提供HTTP服务的Server。当前主要用于Executor端下载依赖。

10）org.apache.spark.metrics.MetricsSystem：用于搜集统计信息。

11）org.apache.spark.shuffle.ShuffleMemoryManager：管理Shuffle过程中使用的内存。ExternalAppendOnlyMap 和ExternalSorter都会从ShuffleMemoryManager中申请内存，在数据spill到Disk后会释放内存。当然了，当Task退出时这个内存也会被回收。为了使得每个thread都会比较公平地获取内存资源，避免一个thread申请了大量内存后造成其他的thread需要频繁地进行spill操作，它采取的内存分配策略是：对于N个thread，每个thread可以至少申请1/（2*N）的内存，但是至多申请1/N。这个N是动态变化的，感兴趣的读者可以查阅这个类的具体实现。

在用户创建org.apache.spark.SparkContext时会创建org.apache.spark.SparkEnv。

3.4.3 缓存的处理

如果存储级别不是NONE，那么先检查是否有缓存；没有缓存则要进行计算。什么是存储级别？从用户的角度来看就是缓存保存到不同的存储位置，比如内存、硬盘、Tachyon；还有缓存的数据是否需要序列化等。详细的存储级别的介绍可以参阅第8章。

cacheManager对Storage模块进行了封装，使得RDD可以更加简单地从Storage模块读取或者写入数据。RDD的每个Partition对应Storage模块的一个Block，只不过Block是Partition经过处理后的数据。在系统实现的层面上，可以认为Partition和Block是一一对应的。cacheManager会通过getOrCompute来判断当前的RDD是否需要进行计算。

首先，cacheManager会通过RDD的ID和当前计算的Partition的ID向Storage模块的BlockManager发起查询请求，如果能够获得Block的信息，会直接返回Block的信息。否则，代表该RDD是需要计算的。这个RDD以前可能计算过并且被存储到了内存中，但是后来由于内存紧张，这部分内存被清理了。在计算结束后，计算结果会根据用户定义的存储级别，写入BlockManager中。这样，下次就可以不经过计算而直接读取该RDD的计算结果了。核心实现逻辑如下：

def getOrCompute[T](

rdd: RDD[T],

partition: Partition,

context: TaskContext,

storageLevel: StorageLevel): Iterator[T] = {

//获取RDD的BlockId

val key = RDDBlockId(rdd.id, partition.index)

logDebug(s"Looking for partition $key")

blockManager.get(key) match { //向BlockManager查询是否有缓存

case Some(blockResult) => //缓存命中

//更新统计信息，将缓存作为结果返回

context.taskMetrics.inputMetrics = Some(blockResult.inputMetrics)

new InterruptibleIterator(context, blockResult.data.asInstanceOf[Iterator[T]])

case None => //没有缓存命中，需要计算

// 判断当前是否有线程在处理当前的Partition，如果有那么等待它结束后，直接从Block

// Manager中读取处理结果如果没有线程在计算，那么storedValues就是None，否则

// 就是计算的结果

val storedValues = acquireLockForPartition[T](key)

if (storedValues.isDefined) { // 已经被其他线程处理了，直接返回结果

return new InterruptibleIterator[T](context, storedValues.get)

}

// 需要计算

try {

// 如果被checkpoint过，那么读取checkpoint的数据；否则调用rdd的compute()开始

  // 计算

        val computedValues = rdd.computeOrReadCheckpoint(partition,context)

// Task是在Driver端执行的话就不需要缓存结果，这个主要是为了first() 或者take()

// 这种仅仅有一个执行阶段的任务的快速执行。这类任务由于没有Shuffle阶段，直接运行

// 在Driver端可能会更省时间

if (context.isRunningLocally) {

return computedValues

}

// 将计算结果写入到BlockManager

val updatedBlocks = new ArrayBuffer[(BlockId, BlockStatus)]

val cachedValues =

putInBlockManager(key, computedValues, storageLevel, updatedBlocks)

// 更新任务的统计信息

val metrics = context.taskMetrics

val lastUpdatedBlocks = metrics.updatedBlocks.getOrElse(

Seq[(BlockId, BlockStatus)]())

metrics.updatedBlocks = Some(lastUpdatedBlocks ++ updatedBlocks.toSeq)

new InterruptibleIterator(context, cachedValues)

} finally {

loading.synchronized {

loading.remove(key)

// 如果有其他的线程在等待该Partition的处理结果，那么通知它们计算已经完成，结果已

// 经存到BlockManager中（注意前面那类不会写入BlockManager的本地任务）

// loading.notifyAll()

}

}

}

}

3.4.4 checkpoint的处理

在缓存没有命中的情况下，首先会判断是否保存了RDD的checkpoint，如果有，则读取checkpoint。为了理解checkpoint的RDD是如何读取计算结果的，需要先看一下checkpoint的数据是如何写入的。

首先在Job结束后，会判断是否需要checkpoint。如果需要，就调用org.apache. spark.rdd.RDDCheckpointData#doCheckpoint。doCheckpoint首先为数据创建一个目录；然后启动一个新的Job来计算，并且将计算结果写入新创建的目录；接着创建一个org.apache.spark.rdd.CheckpointRDD；最后，原始RDD的所有依赖被清除，这就意味着RDD的转换的计算链（compute chain）等信息都被清除。这个处理逻辑中，数据写入的实现在org.apache.spark.rdd.CheckpointRDD$#writeToFile。简要的核心逻辑如下：

// 创建一个保存checkpoint数据的目录

val path = new Path(rdd.context.checkpointDir.get, "rdd-" + rdd.id)

val fs = path.getFileSystem(rdd.context.hadoopConfiguration)

if (!fs.mkdirs(path)) {

throw new SparkException("Failed to create checkpoint path " + path)

}

// 创建广播变量

val broadcastedConf = rdd.context.broadcast(

new SerializableWritable(rdd.context.hadoopConfiguration))

//开始一个新的Job进行计算，计算结果存入路径path中

rdd.context.runJob(rdd, CheckpointRDD.writeToFile[T](path.toString, broadcastedConf) _)

//根据结果的路径path来创建CheckpointRDD

val newRDD = new CheckpointRDD[T](rdd.context, path.toString)

//保存结果，清除原始RDD的依赖、Partition信息等

RDDCheckpointData.synchronized {

cpFile = Some(path.toString)

cpRDD = Some(newRDD) // RDDCheckpointData对应的CheckpointRDD

rdd.markCheckpointed(newRDD)      // 清除原始RDD的依赖，Partition

cpState = Checkpointed            //标记checkpoint的状态为完成

}

至此，RDD的checkpoint完成，其中checkpoint的数据可以通过checkpointRDD的readFromFile读取。但是，上述逻辑在清除了RDD的依赖后，并没有和check-pointRDD建立联系，那么Spark是如何确定一个RDD是否被checkpoint了，而且正确读取checkpoint的数据呢？

答案就在org.apache.spark.rdd.RDD#dependencies的实现，它会首先判断当前的RDD是否已经Checkpoint过，如果有，那么RDD的依赖就变成了对应的CheckpointRDD：

privatedefcheckpointRDD: Option[RDD[T]]=checkpointData.flatMap(_.checkpointRDD)

final def dependencies: Seq[Dependency[_]] = {

checkpointRDD.map(r => List(new OneToOneDependency(r))).getOrElse {

if (dependencies_ == null) { //没有checkpoint

dependencies_ = getDependencies

}

dependencies_

}

}

理解了Checkpoint的实现过程，接下来看一下computeOrReadCheckpoint的实现。前面提到了，它一共在两个地方被调用，org.apache.spark.rdd.RDD#iterator和org.apache. spark.CacheManager#getOrCompute。它实现的逻辑比较简单，首先检查当前RDD是否被Checkpoint过，如果有，读取Checkpoint的数据；否则开始计算。实现如下：

private[spark] def computeOrReadCheckpoint(split: Partition, context: TaskContext)

: Iterator[T] =

{

if (isCheckpointed) firstParent[T].iterator(split, context) else compute(split, context)

}

firstParent[T].iterator(split，context)会调用对应CheckpointRDD的iterator，最终调用到它的compute：

override def compute(split: Partition, context: TaskContext): Iterator[T] = {

val file=new Path(checkpointPath, CheckpointRDD.splitIdToFile(split.index))

CheckpointRDD.readFromFile(file, broadcastedConf, context) //读取Checkpoint的数据

}

3.4.5 RDD的计算逻辑

RDD的计算逻辑在org.apache.spark.rdd.RDD#compute中实现。每个特定的RDD都会实现compute。比如前面提到的CheckpointRDD的compute就是直接读取checkpoint数据。HadoopRDD就是读取指定Partition的数据。MapPartitionsRDD就是将用户的转换逻辑作用到指定的Partition上。