背景 SQL作為一門標準的、通用的、簡單的DSL，在大數據分析中有著越來越重要的地位;Spark在批處理引擎領域當前也是處於絕對的地位，而Spark2.0中的SparkSQL也支持ANSI SQL 2003標準。因此SparkSQL在大數據分析中的地位不言而喻。本文將通過分析一條SQL在Spa ...

背景

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 發佈！

一條Sql的Spark之旅

背景