一條Sql的Spark之旅
背景
SQL作為一門標準的、通用的、簡單的DSL，在大資料分析中有著越來越重要的地位;Spark在批處理引擎領域當前也是處於絕對的地位，而Spark2.0中的SparkSQL也支援ANSI-SQL 2003標準。因此SparkSQL在大資料分析中的地位不言而喻。
本文將通過分析一條SQL在Spark中的解析執行過程來梳理SparkSQL執行的一個流程。
案例分析
程式碼
val spark = SparkSession.builder().appName("TestSql").master("local[*]").enableHiveSupport().getOrCreate() val df = spark.sql("select sepal_length,class from origin_csvload.csv_iris_qx order by sepal_length limit 10 ") df.show(3)
我們在數倉中新建了一張表origin_csvload.csv_iris_qx，然後通過SparkSQL執行了一條SQL，由於整個過程由於是懶載入的，需要通過Terminal方法觸發，此處我們選擇show方法來觸發。
原始碼分析
詞法解析、語法解析以及分析
sql方法會執行以下3個重點：
sessionState.sqlParser.parsePlan(sqlText)：將SQL字串通過ANTLR解析成邏輯計劃（Parsed Logical Plan）sparkSession.sessionState.executePlan(logicalPlan)：執行邏輯計劃，此處為懶載入，只新建QueryExecution例項，並不會觸發實際動作。需要注意的是QueryExecution其實是包含了SQL解析執行的4個階段計劃（解析、分析、優化、執行）QueryExecution.assertAnalyzed()：觸發語法分析，得到分析計劃（Analyzed Logical Plan）
def sql(sqlText: String): DataFrame = { //1:Parsed Logical Plan Dataset.ofRows(self, sessionState.sqlParser.parsePlan(sqlText)) } def ofRows(sparkSession: SparkSession, logicalPlan: LogicalPlan): DataFrame = { val qe = sparkSession.sessionState.executePlan(logicalPlan)//d-1 qe.assertAnalyzed()//d-2 new Dataset[Row](sparkSession, qe, RowEncoder(qe.analyzed.schema)) } //d-1 def executePlan(plan: LogicalPlan): QueryExecution = new QueryExecution(sparkSession, plan) //2:Analyzed Logical Plan lazy val analyzed: LogicalPlansparkSession.sessionState.analyzer.executeAndCheck(logical)
解析計劃和分析計劃
sql解析後計劃如下：
== Parsed Logical Plan ==
'GlobalLimit 10
+- 'LocalLimit 10
+- 'Sort ['sepal_length ASC NULLS FIRST], true
+- 'Project ['sepal_length, 'class]
+- 'UnresolvedRelation `origin_csvload`.`csv_iris_qx`主要是將SQL一一對應地翻譯成了catalyst的操作，此時資料表並沒有被解析，只是簡單地識別為表。而分析後的計劃則包含了欄位的位置、型別，表的具體型別（parquet）等資訊。
== Analyzed Logical Plan ==
sepal_length: double, class: string
GlobalLimit 10
+- LocalLimit 10
+- Sort [sepal_length#0 ASC NULLS FIRST], true
+- Project [sepal_length#0, class#4]
+- SubqueryAlias `origin_csvload`.`csv_iris_qx`
+- Relation[sepal_length#0,sepal_width#1,petal_length#2,petal_width#3,class#4] parquet此處有個比較有意思的點，UnresolvedRelation origin_csvload.csv_iris_qx被翻譯成了一個子查詢別名，讀取檔案出來的資料註冊成了一個表，這個是不必要的，後續的優化會消除這個子查詢別名。
優化以及執行
以DataSet的show方法為例，show的方法呼叫鏈為showString->getRows->take->head->withAction，我們先來看看withAction方法：
def head(n: Int): Array[T] = withAction("head", limit(n).queryExecution)(collectFromPlan)
private def withAction[U](name: String, qe: QueryExecution)(action: SparkPlan => U) = {
val
result= SQLExecution.withNewExecutionId(sparkSession, qe) {
action(qe.executedPlan)
}
result
}withAction方法主要執行如下邏輯：
1. 拿到快取的解析計劃，使用遍歷優化器執行解析計劃，得到若干優化計劃。
2. 獲取第一個優化計劃，遍歷執行前優化獲得物理執行計劃，這是已經可以執行的計劃了。
3. 執行物理計劃，返回實際結果。至此，這條SQL之旅就結束了。
//3:Optimized Logical Plan,withCachedData為Analyzed Logical Plan，即快取的變數analyzed
lazy val optimizedPlan: LogicalPlan = sparkSession.sessionState.optimizer.execute(withCachedData)
lazy val sparkPlan: SparkPlan = planner.plan(ReturnAnswer(optimizedPlan)).next()
//4:Physical Plan
lazy val executedPlan: SparkPlan = prepareForExecution(sparkPlan)優化計劃及物理計劃
優化後的計劃如下，可以看到SubqueryAliases已經沒有了。
== Optimized Logical Plan ==
GlobalLimit 10
+- LocalLimit 10
+- Sort [sepal_length#0 ASC NULLS FIRST], true
+- Project [sepal_length#0, class#4]
+- Relation[sepal_length#0,sepal_width#1,petal_length#2,petal_width#3,class#4] parquet具體的優化點如下圖所示，行首有!表示優化的地方。
其中"=== Result of Batch Finish Analysis ==="表示"Finish Analysis"的規則簇（參見附錄一）被應用成功，可以看到該規則簇中有一個消除子查詢別名的規則EliminateSubqueryAliases
Batch("Finish Analysis", Once,
EliminateSubqueryAliases,
ReplaceExpressions,
ComputeCurrentTime,
GetCurrentDatabase(sessionCatalog),
RewriteDistinctAggregates)最後根據物理計劃生成規則（附錄二）可以得到物理計劃，這就是已經可以執行的計劃了。具體如下：
== Physical Plan ==
TakeOrderedAndProject(limit=10, orderBy=[sepal_length#0 ASC NULLS FIRST], output=[sepal_length#0,class#4])
+- *(1) Project [sepal_length#0, class#4]
+- *(1) FileScan parquet origin_csvload.csv_iris_qx[sepal_length#0,class#4] Batched: true, Format: Parquet, Location: CatalogFileIndex[hdfs://di124:8020/user/hive/warehouse/origin_csvload.db/csv_iris_qx], PartitionCount: 1, PartitionFilters: [], PushedFilters: [], ReadSchema: struct<sepal_length:double,class:string>總結
本文簡述了一條SQL是如何從字串經過詞法解析、語法解析、規則優化等步驟轉化成可執行的物理計劃，最後以一個Terminal方法觸發邏輯返回結果。本文可為後續SQL優化提供一定思路，之後可再詳述具體的SQL優化原則。
附錄一：優化方法
分析計劃會依次應用如下優化：
- 前置優化。當前為空。
- 預設優化。主要有如下類別，每個類別分別有若干優化規則。
- Optimize Metadata Only Query
- Extract Python UDFs
- Prune File Source Table Partitions
- Parquet Schema Pruning
- Finish Analysis
- Union
- Subquery
- Replace Operators
- Aggregate
- Operator Optimizations
- Check Cartesian Products
- Decimal Optimizations
- Typed Filter Optimization
- LocalRelation
- OptimizeCodegen
- RewriteSubquery
- 後置優化。當前為空。
- 使用者提供的優化。來自
experimentalMethods.extraOptimizations，當前也沒有。
附錄二：物理計劃生成規則
生成物理執行計劃的規則如下：
- PlanSubqueries
- EnsureRequirements
- CollapseCodegenStages
- ReuseExchange
- ReuseSubquery
本文由部落格一文多發平臺 OpenWrite 釋出！