Spark源码分析之Job触发原理
一 Job的执行流程
1.1 从数据源加载数据,数据源可以是本地数据文件和HDFS文件,也可以你是内存里的数据结构或者HBase等,创建初始的RDD
1.2 对RDD进行一系列的transformation操作,每一个transformation可能产生一个或者多个RDD
1.3 对最后的final RDD进行action操作,触发job操作,将最后每一个分区计算后得到结果
1.4 对每一个分区的结果返回到Driver端,进行最后的计算。比如count实际上包含了action和sum两个步骤的计算。RDD可以被cache到内存,也可以checkpoint到磁盘。
二 Job触发流程源码
2.1 调用action操作,运行job
我们以count这个action操作为例子,它首先会调用SparkContext的runJob方法
def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum
2.2runJob
在一个RDD的所有分区上运行job,并且返回结果
def runJob[T, U: ClassTag](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int], resultHandler: (Int, U) => Unit): Unit = {
// 判断SparkContext是否停止或者关闭
if (stopped.get()) {
throw new IllegalStateException("SparkContext has been shutdown")
}
val callSite = getCallSite
// 清除闭包
val cleanedFunc = clean(func)
logInfo("Starting job: " + callSite.shortForm)
if (conf.getBoolean("spark.logLineage", false)) {
logInfo("RDD's recursive dependencies:\n" + rdd.toDebugString)
}
// 委托给DAGScheduler的runJob方法,提交Job
dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
progressBar.foreach(_.finishAll())
// RDD执行checkpoint操作
rdd.doCheckpoint()
}
2.3submitJob
def submitJob[T, U](rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
callSite: CallSite,
resultHandler: (Int, U) => Unit,
properties: Properties): JobWaiter[U] = {
// 判断任务处理的分区是否存在
val maxPartitions = rdd.partitions.length
partitions.find(p => p >= maxPartitions || p < 0).foreach { p =>
throw new IllegalArgumentException(
"Attempting to access a non-existent partition: " + p + ". " +
"Total number of partitions: " + maxPartitions)
}
// 获取jobId,如果作业只包含0个任务,则立即返回JobWaiter
val jobId = nextJobId.getAndIncrement()
if (partitions.size == 0) {
return new JobWaiter[U](this, jobId, 0, resultHandler)
}
assert(partitions.size > 0)
val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
// 创建JobWaiter对象
val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
// 创建JobSubmitted对象,放入队列eventProcessLoop
eventProcessLoop.post(JobSubmitted(
jobId, rdd, func2, partitions.toArray, callSite, waiter,
SerializationUtils.clone(properties)))
waiter
}