目前网络上以及书籍里，对RDD的分析停留在RDD本身，对于Task的分析依然停留在DAG Scheduler, Task Scheduler这样高级调度器，低级调度器的层面。但是很少有把Task的运行和RDD的结构结合起来进行分析的。这样在宏观上确实是可以说清楚例如宽依赖，窄依赖如何区别，DAG中的Stage如何划分。但是对于划分好的Stage中TaskSet如何运行，也就是说如何通过一个Stage内窄依赖，由输入的RDD一层一层得到输出的RDD这块的运算流转逻辑却很少提及。从而造成宏观上的运行机制一想就通，微观上的运行链条总是觉得哪里有问题。[注： DAG Stage是依据RDD的宽依赖为划分标准，通过广度优先搜索以及嵌套递归的方式来划分的]
本文结合网络上看的一些技术帖子以及部分Spark源代码，从SparkRDD的角度出发，简单分析一下Spark Task的运行。
（1）首先还是从DAGScheduler, TaskScheduler出发，看一下每一个Stage中任务是如何触发的。[注： 这里只是从逻辑上阐述一下Stage的划分，没过去深入探究Stage划分中的广度优先以及递归等算法.
先是根据RDD之间的依赖关系划分Stage, 以宽依赖作为划分的依据。如下图所示：
而这三个Stage是如何描述的呢？
private[scheduler] abstract class Stage(
val id: Int,
val rdd: RDD[_],
val numTasks: Int,
val parents: List[Stage],
val firstJobId: Int,
val callSite: CallSite)
extends Logging {...
我们看一下Stage的抽象类，发现其中的参数有rdd, numTasks, parent，这里的rdd就是每一个stage最后的一个RDD, 为什么呢？ 这个是由Stage切分算法决定的，从后往前推，一个宽依赖的父RDD就是是新的Stage中最后的RDD。
如果图中Stage2的RDD就是RDD F
在spark调用action的时候DAG图从前往后进行计算，例如要计算Stage3则先要计算Stage1和Stage2.
以Stage2的提交计算举例：
而在提交计算Stage2的时候，其对应的RDD就是RDD F,其中还有4个partition, 所以其对应的TaskSet中有4个Task. 在给每个task分配具体的core之后，会将每个task对应的TaskDescription发送到Executor上，然后被反序列之后进行执行（注意提交的时候是一个task一个task的提交，所以executor上反序列化后的也是一个task一个task的执行）
在ShuffleMapTask的runTask中有如下代码：
override def runTask(context: TaskContext): MapStatus = {
// Deserialize the RDD using the broadcast variable.
val deserializeStartTime = System.currentTimeMillis()
val ser = SparkEnv.get.closureSerializer.newInstance()
val (rdd, dep) = ser.deserialize[(RDD[_], ShuffleDependency[_, _, _])](
ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)
_executorDeserializeTime = System.currentTimeMillis() - deserializeStartTime
var writer: ShuffleWriter[Any, Any] = null
try {
val manager = SparkEnv.get.shuffleManager
writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context)
writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])
writer.stop(success = true).get
} catch {
case e: Exception =>
try {
if (writer != null) {
writer.stop(success = false)
}
} catch {
case e: Exception =>
log.debug("Could not stop writer", e)
}
throw e
}
}
如上图，高亮部分，通过SparkEnv拿到ShuffleManager，从而拿到ShuffleWriter（例如SortShuffleWriter）。writer.write负责计算结果的缓存处理以及持久化，所以其参数
rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]] 就是某个Task计算的结果。
下面我们就从SparkRDD的角度来看一下，task的计算:
注意writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])中的参数rdd， 这个rdd是，SparkEnv.get.closureSerializer.newInstance()反序列化后得到的。其中的dep是本stage对应的shuffle依赖，也就是是否需要从上一个stage的持久化的shuffle文件中读数据。
（2）
通过上面的图我们可以看到，一个Stage里的task可能会跨越好几个窄依赖的RDD，例如从hdfs中读取文件，然后经过map操作，再经过filter操作，例如：
var textRDD = sc.textFile(“xxx”)
var result = textRDD.map(word = > (word, 1)).filter(xxxxxx)
这样一个特定task，例如ShuffleMapTask 是和一个特定的数据分片想对应的，有就是一个partition。
一个Stage的TaskSets对应着一系列数据分片不同但是数据流转算法相同的的task
taskSets中的每一个task是与一个RDD的一个partition相对应的。
换句话说Driver会把一个stage中的TaskSet分配好资源之后全部发送给Executor， 每个Task对应的partiton不同。在Executor上反序列化之后的Task是包含有对应的partition的信息的。
rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]] 就是某个Task计算的结果。
下来我们看一下RDD里相关的方法。
在抽象类RDD中(参见RDD.scala)
这里需要说明一下，RDD是一种抽象的概念，是一个对分布式数据的操作管理的集合，其本身并不代表数据本身。数据本身是在RDD映射下的流转。
有如下的虚函数：
/**
* :: DeveloperApi ::
* Implemented by subclasses to compute a given partition.
*/
@DeveloperApi
def compute(split: Partition, context: TaskContext): Iterator[T]
具体特定的RDD会覆写这个方法。例如MapPartitionsRDD，HadoopRDD, UnionRDD
等等（参见package org.apache.spark.rdd）
通过上面论述可以得知在提交Stage2中高亮部分的代码对应的是RDD F（stage2通过宽依赖划分的最后一个RDD）.
那接着我们来看
rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]]。
我们假设RDD F是一个UnionRDD
则rdd.iterator先调用其父类RDD中的
/**
* 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.
*/
final def iterator(split: Partition, context: TaskContext): Iterator[T] = {
if (storageLevel != StorageLevel.NONE) {
getOrCompute(split, context)
} else {
computeOrReadCheckpoint(split, context)
}
}
我们先不考虑有Checkpoint的情况，所以直接来看getOrCompute
/**
* Gets or computes an RDD partition. Used by RDD.iterator() when an RDD is cached.
*/
private[spark] def getOrCompute(partition: Partition, context: TaskContext): Iterator[T] = {
val blockId = RDDBlockId(id, partition.index)
var readCachedBlock = true
// This method is called on executors, so we need call SparkEnv.get instead of sc.env.
SparkEnv.get.blockManager.getOrElseUpdate(blockId, storageLevel, elementClassTag, () => {
readCachedBlock = false
computeOrReadCheckpoint(partition, context)
}) match {
case Left(blockResult) =>
if (readCachedBlock) {
val existingMetrics = context.taskMetrics().inputMetrics
existingMetrics.incBytesRead(blockResult.bytes)
new InterruptibleIterator[T](context, blockResult.data.asInstanceOf[Iterator[T]]) {
override def next(): T = {
existingMetrics.incRecordsRead(1)
delegate.next()
}
}
} else {
new InterruptibleIterator(context, blockResult.data.asInstanceOf[Iterator[T]])
}
case Right(iter) =>
new InterruptibleIterator(context, iter.asInstanceOf[Iterator[T]])
}
}
看高亮部分
/**
* Compute an RDD partition or read it from a checkpoint if the RDD is checkpointing.
*/
private[spark] def computeOrReadCheckpoint(split: Partition, context: TaskContext): Iterator[T] =
{
if (isCheckpointedAndMaterialized) {
firstParent[T].iterator(split, context)
} else {
compute(split, context)
}
}
最终会调用compute(split, context)，而compute是子类UnionRDD提供的，
UnionRDD.scala中
override def compute(s: Partition, context: TaskContext): Iterator[T] = {
val part = s.asInstanceOf[UnionPartition[T]]
parent[T](part.parentRddIndex).iterator(part.parentPartition, context)
}
这里可以看到F1的4个task任务，前两个会调用RDD D的iterator， 后两个会调用RDD E的iterator, 然后没有做其他任何工作
RDD D是一个MapPartitionsRDD， RDD E例如是一个HadoopRDD。整个方法调用顺序和上面类似，实时HadoopRDD的compute方法不需要再调用其依赖的RDD的iterator方法了，因为其就是数据的开始来源。而RDD D
的compute是MapPartitionsRDD提供的，
MapPartitionsRDD.scala中
override def compute(split: Partition, context: TaskContext): Iterator[U] =
f(context, split.index, firstParent[T].iterator(split, context))
类的这RDD D的compute方法会调用RDD C的iterator方法，同理一个task映射一个partition。得到结果后用f来做一个map映射。
RDD C的iterator方法，如果C也是一个数据的源头则其肯定自己提供了一个读数据的compute方法。如果其实一个shuffle过程的mapper开始的数据，则其实一个ShuffledRDD,其compute方法如下
override def compute(split: Partition, context: TaskContext): Iterator[(K, C)] = {
val dep = dependencies.head.asInstanceOf[ShuffleDependency[K, V, C]]
SparkEnv.get.shuffleManager.getReader(dep.shuffleHandle, split.index, split.index + 1, context)
.read()
.asInstanceOf[Iterator[(K, C)]]
}
调用了shuffleManager.getReader从shuffle文件中读取数据。
通过上述分析，可以看到
（1）一个Stage中的task的计算过程实质是RDD的compute方法的链式调用的过程，是从按照宽依赖划分的最后一个RDD开始一直向父RDD的compute进行调用的过程。
（2）一个Stage里的Task Set中的一个task是个这个Stage中所有的RDD的算子相对应的，一个task需要串型调用这些RDD的compute方法。
（3）Spark中最细粒度，也就是实质性的计算都是定义在RDD中的，通过外界传入的参数，在集群的Executor中会去对特定的数据分片（对应特定的Task）执行RDD中的算子。
加入技术讨论群
《大数据和云计算技术》社区群人数已经3000+，欢迎大家加下面助手微信，拉大家进群，自由交流。
喜欢QQ群的，可以扫描下面二维码：
欢迎大家通过二维码打赏支持技术社区（英雄请留名，社区感谢您，打赏次数超过108+）：
