Spark SQL模塊,主要就是處理跟SQL解析相關的一些內容,說得更通俗點就是怎麼把一個SQL語句解析成Dataframe或者說RDD的任務。以Spark 2.4.3為例,Spark SQL這個大模塊分為三個子模塊,如下圖所示 其中Catalyst可以說是Spark內部專門用來解析SQL的一個框架 ...
Spark SQL模塊,主要就是處理跟SQL解析相關的一些內容,說得更通俗點就是怎麼把一個SQL語句解析成Dataframe或者說RDD的任務。以Spark 2.4.3為例,Spark SQL這個大模塊分為三個子模塊,如下圖所示
其中Catalyst可以說是Spark內部專門用來解析SQL的一個框架,在Hive中類似的框架是Calcite(將SQL解析成MapReduce任務)。Catalyst將SQL解析任務分成好幾個階段,這個在對應的論文中講述得比較清楚,本系列很多內容也會參考論文,有興趣閱讀原論文的可以到這裡看:Spark SQL: Relational Data Processing in Spark。
而Core模塊其實就是Spark SQL主要解析的流程,當然這個過程中會去調用Catalyst的一些內容。這模塊裡面比較常用的類包括SparkSession,DataSet等。
至於hive模塊,這個不用說,肯定跟hive有關的。這個模塊在本系列基本不會涉及到,就不多介紹了。
值得一提的是,論文發表的時候還是在Spark1.x階段,那個時候SQL解析成詞法樹用的是scala寫的一個解析工具,到2.x階段改為使用antlr4來做這部分工作(這應該算是最大的改變)。至於為什麼要改,我猜是出於可讀性和易用性方面的考慮,當然這個僅是個人猜測。
另外,這一系列會簡單介紹一條SQL語句的處理流程,基於spark 2.4.3(sql這個模塊在spark2.1後變化不大)。這一篇先從整體介紹Spark SQL出現的背景及解決問題,Dataframe API以及Catalyst的流程大概是怎麼樣,後面分階段細說Catalyst的流程。
Spark SQL出現的背景及解決的問題
在最早的時候,大規模處理數據的技術是MapReduce,但這種框架執行效率太慢,進行一些關係型處理(如join)需要編寫大量代碼。後來hive這種框架可以讓用戶輸入sql語句,自動進行優化並執行。
但在大型系統中,任然有兩個主要問題,一個是ETL操作需要對接多個數據源。另一個是用戶需要執行複雜分析,比如機器學習和圖計算等。但傳統的關係型處理系統中較難實現。
Spark SQL提供了兩個子模塊來解決這個問題,DataFrame API和Catalyst。
相比於RDD,Dataframe api提供了更加豐富的關係型api,並且能和RDD相互轉換,後面Spark機器學習方面的工作重心,也從以RDD為基礎的mllib轉移到以Dataframe為基礎的Spark ML(雖然Dataframe底層也是RDD)。
另一個就是Catalyst,通過它可以輕鬆為諸如機器學習之類的域添加數據源(比如json或通過case class自定義的類型),優化規則和數據類型。
通過這兩個模塊,Spark SQL主要實現以下目標:
- 提供方便易用好的API,包括讀取外部數據源,以及關係數據處理(用過的都知道)
- 使用已建立的DBMS技術提供高性能。
- 輕鬆支持新數據源,包括半結構化數據和外部資料庫(比如MYSQL)。
- 圖計算和機器學習方面的拓展
那下麵就介紹Dataframe和Catalyst的流程,當然主要討論的還是Catalyst。
統一API Dataframe
先來看看論文裡面提供的一張圖:
這張圖可以說明很多,首先Spark的Dataframe API底層也是基於Spark的RDD。但與RDD不同的在於,Dataframe會持有schema(這個實在不好翻譯,可以理解為數據的結構吧),以及可以執行各種各樣的關係型操作,比如Select,Filter,Join,Groupby等。從操作上來說,和pandas的Dataframe有點像(連名字都是一樣的)。
同時因為是基於RDD的,所以很多RDD的特性Dataframe都能夠享受到,比如說分散式計算中一致性,可靠性方面的保證,以及可以通過cache緩存數據,提高計算性能啊等等。
同時圖中頁展示了Dataframe可以通過JDBC鏈接外部資料庫,通過控制台操作(spark-shell),或者用戶程式。說白了,就是Dataframe可以通過RDD轉換而來,也可以通過外部數據表生成。
對了,這裡順便說一句,很多初次接觸Spark SQL的童鞋可能會對Dataset和Dataframe這兩個東西感到疑惑,在1.x時代它們確實有些差別,不過在spark2.x的時候,這兩個API已經統一了。所以基本上Dataset和Dataframe可以看成是等價的東西。
最後還是結合代碼做一下實際的展示吧,如下展示生成一個RDD,並且根據這個RDD生成對應的Dataframe,從中可以看出RDD和Dataframe的區別:
//生成RDD
scala> val data = sc.parallelize(Array((1,2),(3,4)))
data: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[0] at parallelize at <console>:24
scala> data.foreach(println)
(1,2)
(3,4)
scala> val df = data.toDF("fir","sec")
df: org.apache.spark.sql.DataFrame = [fir: int, sec: int]
scala> df.show()
+---+---+
|fir|sec|
+---+---+
| 1| 2|
| 3| 4|
+---+---+
//跟RDD相比,多了schema
scala> df.printSchema()
root
|-- fir: integer (nullable = false)
|-- sec: integer (nullable = false)
Catalyst流程解析
Catalyst在論文中被叫做優化器(Optimizer),這部分是論文裡面較為核心的內容,不過其實流程還是蠻好理解的,依舊貼下論文裡面的圖。
主要流程大概可以分為以下幾步:
- Sql語句經過Antlr4解析,生成Unresolved Logical Plan(有使用過Antlr4的童鞋肯定對這一過程不陌生)
- analyzer與catalog進行綁定(catlog存儲元數據),生成Logical Plan;
- optimizer對Logical Plan優化,生成Optimized LogicalPlan;
- SparkPlan將Optimized LogicalPlan轉換成 Physical Plan;
- prepareForExecution()將 Physical Plan 轉換成 executed Physical Plan;
- execute()執行可執行物理計劃,得到RDD;
提前說一下吧,上述流程多數是在org.apache.spark.sql.execution.QueryExecution這個類裡面,這個貼一下簡單的代碼,看看就好,先不多做深究。後面的文章會詳細介紹這裡的內容。
class QueryExecution(val sparkSession: SparkSession, val logical: LogicalPlan) {
......其他代碼
//analyzer階段
lazy val analyzed: LogicalPlan = {
SparkSession.setActiveSession(sparkSession)
sparkSession.sessionState.analyzer.executeAndCheck(logical)
}
//optimizer階段
lazy val optimizedPlan: LogicalPlan = sparkSession.sessionState.optimizer.execute(withCachedData)
//SparkPlan階段
lazy val sparkPlan: SparkPlan = {
SparkSession.setActiveSession(sparkSession)
// TODO: We use next(), i.e. take the first plan returned by the planner, here for now,
// but we will implement to choose the best plan.
planner.plan(ReturnAnswer(optimizedPlan)).next()
}
//prepareForExecution階段
// executedPlan should not be used to initialize any SparkPlan. It should be
// only used for execution.
lazy val executedPlan: SparkPlan = prepareForExecution(sparkPlan)
//execute階段
/** Internal version of the RDD. Avoids copies and has no schema */
lazy val toRdd: RDD[InternalRow] = executedPlan.execute()
......其他代碼
}
值得一提的是每個階段都使用了lazy懶載入,對這塊感興趣可以看看我之前的文章Scala函數式編程(六) 懶載入與Stream。
上述主要介紹Spark SQL模塊內容,其出現的背景以及主要解決問題。而後簡單介紹下Dataframe API的內容,以及Spark SQL解析SQL的內部框架Catalyst。後續主要會介紹Catalyst中各個步驟的流程,結合源碼來做一些分析。
以上~