[1]參考文章：高彥傑，陳冠誠 Spark SQL : 基於記憶體的大資料分析引擎《程式設計師》2014 . 8

AMPLab將大資料分析負載分為三大型別：批量資料處理、互動式查詢、實時流處理。而其中很重要的一環便是互動式查詢。大資料分析棧中需要滿足使用者ad-hoc、reporting、iterative等型別的查詢需求，也需要提供SQL介面來相容原有資料庫使用者的使用習慣，同時也需要SQL能夠進行關係模式的重組。完成這些重要的SQL任務的便是 Spark SQL和Shark這兩個開源分散式大資料查詢引擎，它們可以理解為輕量級Hive SQL在Spark上的實現，業界將該類技術統稱為SQL on Hadoop。

在Spark 峰會2014上，Databricks宣佈不再支援Shark的開發，全力以赴開發Shark的下一代技術Spark SQL，同時Hive社群也啟動了Hive onSpark專案，將Spark作為Hive（除MapReduce和Tez之外的）新執行引擎。根據伯克利的BigData Benchmark測試對比資料，Shark的In Memory 效能可以達到Hive的100倍，即使是On Disk 也能達到10倍的效能提升，是Hive的強有力的替代解決方案。而作為Shark的進化版本的Spark SQL，在AMPLab最新的測試中的效能已經超過Shark。圖3-1展示了Spark SQL和Hive on Spark是新的發展方向。

Development……to Spark SQL：Shark開發終止，轉向Spark SQL

A new……for Spark：基於Spark的新的SQL查詢引擎

Help……to Spark：幫助現有Hive使用者遷移到Spark

圖3-1  Spark SQL和Hive on Spark是新的發展方向

.1.1為什麼使用Spark SQL

由於Shark底層依賴於Hive，這個架構的優勢是對傳統Hive使用者可以將Shark無縫整合進現有系統執行查詢負載。但是我們也看到一些問題：隨著版本升級，查詢優化器依賴於Hive，不方便新增新的優化策略，需要進行另一套系統的學習和二次開發，學習成本很高。另一方面，MapReduce是程序級並行，例如：Hive在不同的程序空間會使用一些靜態變數，當在同一程序空間進行多執行緒並行執行，多執行緒同時寫同名稱的靜態變數會產生一致性問題，所以Shark需要使用另外一套獨立維護的Hive原始碼分支。而為了解決這個問題AMPLab和Databricks利用Catalyst開發了SparkSQL。

Spark的全棧解決方案為使用者提供了多樣的資料分析框架，機器學習、圖計算、流計算如火如荼的發展和流行吸引了大批的學習者，為什麼我們今天還是要重視的在大資料環境下使用SQL呢？筆者認為主要有以下幾點原因：

1）易用性與使用者慣性。在過去的很多年中，有大批的程式設計師的工作是圍繞著資料庫+應用的架構來做的，因為SQL的易用性提升了應用的開發效率。程式設計師已經習慣了業務邏輯程式碼呼叫SQL的模式去寫程式，慣性的力量是強大的，如果還能用原有的方式解決現有的大資料問題，何樂而不為呢？提供SQL和JDBC的支援會讓傳統使用者像以前一樣的書寫程式，大大減少遷移成本。

2）生態系統的力量。很多系統軟體效能好，但是未取得成功和沒落，很大程度上因為生態系統問題。傳統的SQL在JDBC、ODBC、SQL的各種標準下形成了一整套成熟的生態系統，很多應用元件和工具可以遷移使用，像一些視覺化的工具、資料分析工具等，原有企業的IT工具可以無縫過渡。

3）資料解耦，Spark SQL正在擴充套件支援多種持久化層，使用者可以使用原有的持久化層儲存資料，但是也可以體驗和遷移到Spark SQL提供的資料分析環境下進行Big Data的分析。

2Spark SQL架構分析

Spark SQL與傳統的DBMS的查詢優化器+執行器的架構較為類似，只不過其執行器是在分散式環境中實現，並採用的Spark作為執行引擎。SparkSQL的查詢優化是Catalyst，其基於Scala語言開發，可以靈活利用Scala原生的語言特性很方便進行功能擴充套件，奠定了Spark SQL的發展空間。Catalyst將SQL語言翻譯成最終的執行計劃，並在這個過程中進行查詢優化。這裡和傳統不太一樣的地方就在於，SQL經過查詢優化器最終轉換為可執行的查詢計劃是一個查詢樹，傳統DB就可以執行這個查詢計劃了。而Spark SQL最後執行還是會在Spark內將這棵執行計劃樹轉換為Spark的有向無環圖DAG再進行執行。

1.Catalyst架構及執行流程分析

下面我們可以看到Catalyst的整體架構：

Phases……relational queries：規則分析，優化，查詢計劃的各個階段

Unresolved Logical Plan :未解析的邏輯查詢計劃

Logical Plan:邏輯查詢計劃

Optimized Logical Plan:優化的邏輯查詢計劃

Physical Plans:物理查詢計劃

Analysis Rules:分析規則

Optimization Rules:優化規則

Planning Rules:計劃規則

圖3-2  Spark SQL查詢引擎Catalyst的架構

我們從圖3-2中可以看到整個的Catalyst是Spark SQL的排程核心，遵循傳統資料庫的查詢解析步驟，對SQL進行解析，轉換為邏輯查詢計劃，物理查詢計劃，最終轉換為Spark的DAG進行執行。圖3-3為Catalyst的執行流程。

圖3-3  Catalyst的執行流程

SQlParser將SQL語句轉換為邏輯查詢計劃，Analyzer對邏輯查詢計劃進行屬性和關係關聯檢驗，之後Optimizer通過邏輯查詢優化將邏輯查詢計劃轉換為優化的邏輯查詢計劃，QueryPlanner將優化的邏輯查詢計劃轉換為物理查詢計劃，prepareForExecution調整資料分佈，最後將物理查詢計劃轉換為執行計劃進入Spark執行任務。

2.Spark SQL優化策略

查詢優化是傳統資料庫中最為重要的一環，這項技術在傳統資料庫中已經很成熟。除了查詢優化，Spark SQL在儲存上也是進行了優化，下面我們從以下幾點看看Spark SQL的一些優化策略：

（1）記憶體列式儲存與記憶體快取表

Spark SQL可以通過cacheTable將資料儲存轉換為列式儲存，同時將資料載入到記憶體進行快取。cacheTable相當於在分散式叢集的記憶體物化檢視，將資料進行快取，這樣迭代的或者互動式的查詢不用再從HDFS讀資料，直接從記憶體讀取資料大大減少了I/O開銷。列式儲存的優勢在於Spark SQL只需要讀出使用者需要的列，而不需要像行儲存那樣需要每次將所有列讀出，從而大大減少記憶體快取資料量，更高效的利用記憶體資料快取，同時減少網路傳輸和IO開銷。資料按照列式儲存，由於是資料型別相同的資料連續儲存，能夠利用序列化和壓縮減少記憶體空間的佔用。

（2）列儲存壓縮

為了減少記憶體和硬碟空間佔用，Spark SQL採用了一些壓縮策略對記憶體列儲存資料進行壓縮。Spark SQL的壓縮方式要比Shark豐富很多，例如它支援PassThrough,RunLengthEncoding, DictionaryEncoding, BooleanBitSet, IntDelta, LongDelta等多種壓縮方式。這樣能夠大幅度減少記憶體空間佔用和網路傳輸開銷和I/O開銷。

（3）邏輯查詢優化

SparkSQL在邏輯查詢優化（見圖3-4）上支援列剪枝、謂詞下壓、屬性合併等邏輯查詢優化方法。列剪枝為了減少讀取不必要的屬性列，減少資料傳輸和計算開銷，在查詢優化器進行轉換的過程中會進行列剪枝的優化。

下面我們介紹一個邏輯優化例子：

SELECT ClassFROM (SELECT ID,Name,Class  FROM STUDENT) S WHERE S.ID=1

圖3-4  邏輯查詢優化

Catalyst將原有查詢通過謂詞下壓，將選擇操作ID=1優先執行，這樣過濾大部分資料，通過屬性合併將最後的投影只做一次最終保留Class屬性列。

（4）Join優化

Spark SQL深度借鑑傳統資料庫的查詢優化技術的精髓，同時也在分散式環境下進行特定的優化策略調整和創新。現在Spark SQL對Join進行了優化支援多種連線演算法，現在的連線演算法已經比Shark豐富，而且很多原來Shark的元素也逐步遷移過來。例如：BroadcastHashJoin、BroadcastNestedLoopJoin、HashJoin、LeftSemiJoin，等等。

下面我們介紹一個其中的BroadcastHashJoin演算法思想:BroadcastHashJoin將小錶轉化為廣播變數進行廣播，這樣避免Shuffle開銷，最後在分割槽內做Hash連線。這裡用的就是Hive中Map Side Join的思想。同時用了DBMS中的Hash連線演算法做連線。

隨著Spark SQL發展，未來會有更多的查詢優化策略加入進來。同時後續Spark SQL會支援像Shark Server一樣的服務端，JDBC介面，相容更多的持久化層例如NoSQL，傳統的DBMS等。一個強有力的結構化大資料查詢引擎正在崛起。

3.如何使用Spark SQL

val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// 在這裡引入sqlContext下所有的方法我們就可以直接用sql方法進行查詢。

import sqlContext._

case class Person(name: String, age: Int)

// 下面的people是含有case型別資料的RDD，會預設由Scala的implicit機制將RDD轉換為SchemaRDD，SchemaRDD是SparkSQL中的核心RDD。

val people =sc.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(p=> Person(p(0), p(1).trim.toInt))

// 在記憶體的元資料中登錄檔資訊，這樣一個Spark SQL表就建立完成了。

people.registerAsTable("people")

// sql語句就會觸發上面分析的SparkSQL的執行過程，讀者可以參考上面的圖示。

val teenagers =sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")

// 最後生成teenagers也是一個RDD

teenagers.map(t=>"Name: " + t(0)).collect().foreach(println)

通過之前的介紹，讀者對支撐結構化資料分析任務的Spark SQL的原理與使用有了一定的瞭解。在生產環境中，有一類資料分析任務對響應延遲要求高，需要實時處理流資料，在BDAS中，Spark Streaming用於支撐大規模流式處理分析任務。

友情推薦：ABC技術研習社

為技術人打造的專屬A(AI),B(Big Data),C(Cloud)技術公眾號和技術交流社群。