前言
上一节 我们看完了集群资源的分配及Executor启动,今天我们就来看下RDD的实际执行流程吧。默认大家已经知道spark中action和transformation两类算子的概念。
样例
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("count")
val sc = new SparkContext(conf)
val lines = sc.textFile("test.txt")
// 随便一些transformation算子的rdd转换
val words = lines.flatMap(line =>line.split(" ")).map{(_,1)}.reduceByKey(_+_)
// 最后加上个action算子
words.count()
words.saveAsTextFile("tmp.txt")
}
lines经过一系列的RDD转换形成dagScheduler,在没有遇到action算子时其实都不会被真正执行。
runJob
我们先以最简单的count()为例
// 真正执行就是runJob() 方法
sc.runJob(this, Utils.getIteratorSize _).sum
// 重载
-> runJob(rdd, func, 0 until rdd.partitions.length)
// 重载
-> runJob(rdd, (ctx: TaskContext, it: Iterator[T]) => cleanedFunc(it), partitions)
// 重载
-> runJob[T, U](rdd, func, partitions, (index, res) => results(index) = res)
// 调用dagScheduler的runJob开始提交,所有的job都是通过这个方法执行的。
dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
progressBar.foreach(_.finishAll())
// 保存运行状态
rdd.doCheckpoint()
submitJob有任务的执行的具体过程,下一节我们再来看看spark源码是如何划分stage任务的。
// 调用submitJob方法,并返回一个回调器
val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
现在先回到样例中的words.saveAsTextFile("tmp.txt")
,这个save方法也是一个action算子,我们同样跟踪下他的源码。
RDD.rddToPairRDDFunctions(r)(nullWritableClassTag, textClassTag, null)
.saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path)
-> // 重载saveAsHadoopFile
saveAsHadoopFile(path, keyClass, valueClass, fm.runtimeClass.asInstanceOf[Class[F]])
-> // 继续重载saveAsHadoopFile
saveAsHadoopDataset(hadoopConf)
-> // writeToFile是一系列写文件的定义,将runJob返回后的结果写入。
self.context.runJob(self, writeToFile)
writer.commitJob()
-> // runJob又到了实际执行任务的方法
runJob(rdd, func, 0 until rdd.partitions.length)
-> ...
清楚RDD的执行流程后,下一节我们就可以看看spark执行时具体是如何切分stage的了。