本文基于 Spark 2.1,其他版本实现可能会有所不同
之前写过不少 Spark Core、Spark Streaming 相关的文章,但使用更广泛的 Spark Sql 倒是极少,恰好最近工作中使用到了,便开始研读相关的源码以及写相应的文章,这篇便作为 Spark Sql 系列文章的第一篇。
既然是第一篇,那么就来说说在 Spark Sql 中一条 sql 语句的主要执行流程,来看看下面这个简单的例子:
val spark = SparkSession
.builder()
.appName("Spark SQL basic example")
.config("spark.some.config.option", "some-value")
.getOrCreate()
// For implicit conversions like converting RDDs to DataFrames
import spark.implicits._
val df = spark.read.json("examples/src/main/resources/people.json")
// Register the DataFrame as a SQL temporary view
df.createOrReplaceTempView("people")
val sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
上面这段代码主要做了这么几件事:
SELECT * FROM people
,得到 sqlDF在这里,主要关注第 3、4 步。第3步是从 sql 语句转化为 DataFrame 的过程,该过程尚未执行 action 操作,并没有执行计算任务;第4步是一个 action 操作,会触发计算任务的调度、执行。下面,我们分别来看看这两大块
这个过程的 uml 时序图如下:
根据该时序图,我们对该过程进一步细分:
需要注意的是,到这里为止,虽然 SparkSession#sql
已经返回,并生成了 sqlDataFrame,但由于该 sqlDataFrame 并没有执行任何 action 操作,所以到这里为止,除了在 driver 端执行了上述分析的操作外,其实并没有触发或执行其他的计算任务。
这个过程最重要的产物 unresolved logical plan 被存放在 sqlDataFrame.queryExecution
中,即 sqlDataFrame.queryExecution.logical
同样可以将上面这个过程进行细分(忽略第1、2步):
QueryExecution#prepareForExecution
方法,将 physical plan 转化为 executable physical plan,主要是插入 shuffle 操作和 internal row 的格式转换如果将 sql 到 dataFrame 及 dataFrame action 串起来,简化上文的分析,最核心的流程应该如下图所示:
这篇文章是一片相对宏观的整体流程的分析,目的有二:
更多关于各个流程的进一步实现分析请见之后的文章