一點資訊 SparkSQL 查詢引擎實踐

02

—

SparkSQL ThriftServer工作原理剖析

圖1: SparkSQL ThriftServer工作的5個步驟

從上圖可知，SparkSQL ThriftServer的啟動過程分 5個步驟。 在步驟3 ExecuteStatement時，SparkSQL 為每一個查詢建立Spark的Job，並隨機生成UUID。

一點大資料團隊修改SparkSQL原始碼實現定製化的JobID生成，並跟蹤每個SparkJob的執行過程實現SQL執行進度的計算。

1:SparkSQL初始化

圖2: SparkSQL ThriftServer 初始化

如上圖所知，SparkSQL ThriftServer初始化包含以下幾個步驟：

1. Java Main函式啟動，初始化HiveServer2，建立 ThriftBinaryCLIService

2. 建立Hive模組內的 ThriftBinaryCLIService 並傳入 SparkSQL模組內的SparkSqlCliService

3. 建立SparkSQLSessionManager 通過反射方式將CliService中的SessionManager設定為自身

1. 使用反射是 因為Hive程式碼中的CLiService的SessonManager為 Private方法

4. Thrift BinaryService啟動，並根據配置的ThriftSever埠、地址等資訊建立TThreadPoolServer。

2:建立連結OpenSession：

客戶端(Beeline CLI) 向伺服器ThriftBinaryServer傳送OpenSession的Thrift請求：

1. SparkSQLCLIService 呼叫初始化過程中建立的SessionManager 建立Session

2. 此時根據配置的不同為Thrift的會話繫結一個SparkSession

3. 每個Session設定自己的SessionManager和OperationManager

1. SessonManager用於管理整個Hive的查詢週期，比如建立連結、關閉連結等。

2. OperationManager 完成具體任務：用於執行具體的任務邏輯比如查看錶資訊、檢視元資料

4. SparkSQL重寫OperationManager的 newExecuteStatementOperation()方法,  轉換為Spark作業

3:發起SQL查詢ExecuteStatement(重點)

圖3: Beeline 客戶端發起查詢SQL請求到 SparkSQL Server

1. 客戶端向伺服器ThriftBinaryServer傳送ExecuteStatement的Thrift請求，伺服器接到請求後獲取當且Thrift會話中的 HiveSession ，並呼叫 ExecuteStatement 方法。

2. 繼續呼叫 executeStatementInternal 函式。此方法呼叫 步驟2:建立鏈 接OpenSession : 中初始化好的 OperationManager執行 newExecuteStatementOperation ()

3. HiveSession繼續呼叫重寫的 runInternal ()方法，將Hive請求進一步呼叫 execute() 來完成Spark的計算邏輯。

4. Spark收到查詢請求後，在 SparkExecuteStatementOperation 中為每一個查詢語句隨機生成UUID作為JobId:

1. 此處是關鍵點即Spark通過UUID來識別每一個的Job,並在Spark UI顯示

2. 而UUID和SQL語句間沒有直接的關聯

5. Spark執行SparkContext.Sql() 函式直接計算獲取結果的DataFrame，結果計算完成將statement標記為完成狀態。返回給客戶端 OperationHandle 控制代碼作為查詢結果的依據。

4: 客戶端獲取結果FetchResults

客戶端檢查Opeartion的狀態，當發現是FINISHED狀態時候，請求結果的元資料Meta資訊和結果內容資訊，分別由客戶端的  FetchResults 請求和  GetResultSetMetadata 請求

1. Spark 將DataFrame的每一行轉換為Thrift結果的Row

2. Spark 將DataFrame的 DataFrame的行列資訊轉換為 Hive的元資料資訊。

5:關閉並釋放連結

03

—

設計原理與程式碼修改

步驟1: 一點查詢客戶端根據SQL生成UUID並儲存對應關係：

• 我們根據Hive JDBC介面，封裝了自己的查詢客戶端，並生成唯一的查詢ID

• QueryID 樣例: 

• 20181014_101745_57dbf79bc1e1f27f82911b00b91ddcde

• 客戶端通過 jdbc:hive2://IP:Port/dw? shark.sparksql.queryid=$QueryID  這樣的連線來訪問後端的ThriftServer。

步驟2: 伺服器接受SQL和QueryID，將Spark的JobID設定為QueryID：

圖4: 如何實現HiveJDBC客戶端傳遞UUID到伺服器端

1. Spark ThriftServer 建立連線請求的時候，根據Url中的引數資訊，HiveSession在建立連結的時候解析 JDBC:hive2 url引數 並且將引數 “shark.sparksql.queryid“ , 以Key-Value 的形式存放於 HiveConf

2. 在OperationManager中，當建立ExecuteStatementOperation時候，讀取所有HiveConf中的配置資訊，寫入到SparkContxt的配置中。

3. 在 執行 ExecuteStatementOperation 時，取上下文  “shark.sparksql.quer yid" 對應的值，並將其設定為JobID, 此ID在SparkUI中可以查詢到

1. statementId=sqlContext.getConf("shark.sparksql.queryid", statementId)

修改後的SparkSQL執行頁面中JobID已經生效

步驟3: 客戶端讀取 Spark RESTful API 獲取查詢進度：

我們採用Spark的Restful API來獲取每一個查詢ID的進展情況

Spark RESTful API的訪問地址為 

http://HOST:4041/api/v1/applications 

• Step1:查詢當前Spark的ApplicationID：http://HOST:4041/api/v1/applications 

• Step2:查詢當前Spark的所有Job列表：http://HOST:4041/api/v1/applications/application_id/jobs application_id為 Step_1中 取到的application_id

• Step3: 根據QueryID，可以獲取到當前QueryID下所有任務數（numTasks),已經完成的任務數（numCompletedTasks），跳過的Task數（numSkippedTasks)

• 計算公式為：完成任務數+跳過任務數)/(總任務數)

圖6: Spark 的Restful API獲取任務狀態

步驟4：效果展示

通過以上對SparkSQL的改進，我們的Shark 一點大資料查詢系統可以支援SparkSQL的進度查詢。

圖7:  藍色為SparkSQL 查詢進度條 當前進度9％

另外我們通過SQL語法分析工具、LeaderLatch（ZK Leader選擇）等技術實現了SQL格式化、高可用的Presto等，使得我們的Shark查詢引擎達到了生產系統的要求。

圖8:  一點大資料查詢系統與Redash、Zeppelin對比分析

04

—

未來規劃

1: SparkSQL ThriftServer的分散式化：

1. 智慧的客戶端根據查詢型別、資料分割槽大小提交至不同的SparkSQL ThriftServer。

2. 對資料取樣、小查詢等建立單獨的SparkSQL查詢後端服務，提高查詢的響應速度對大資料量、高記憶體佔用查詢分配大記憶體的Executor保障查詢的穩定性。

3. 統一的Web客戶端維護查詢ID和後端SparkThriftServer的關係，支援失敗重試、同時提交等功能。