背景

? 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 發布！

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/226116.html

標籤：其他

上一篇：什么是spark？

下一篇：mysql隨機抽取資料

一條Sql的Spark之旅

背景