百度分佈式交互查詢平臺——PINGO架構迭代

PINGO是一個由百度大數據部與百度美國研發中心合做而成的分佈式交換查詢平臺。在PINGO以前，百度的大數據查詢做業主要由基於Hive的百度QueryEngine去完成。QueryEngine很好的支持着百度的離線計算任務，但是它對交互式的在線計算任務支持並很差。爲了更好的支持交互式任務，咱們在大約一年前設計了基於SparkSQL與Tachyon的PINGO的雛形。在過去一年中， 經過跟不一樣業務的結合，PINGO逐漸的演變成一套成熟高效的交互式查詢系統。本文將詳細介紹PINGO的架構迭代過程以及性能評估。

PINGO設計目標

QueryEngine是基於Hive的百度內部的大數據查詢平臺，這套系統在過去幾年中較好的支撐了百度的相關業務。 圖1展現了QueryEngine的架構圖，其服務入口叫作Magi。用戶向Magi提交查詢請求, Magi爲此次請求分配一個執行機, 執行機會調用Hive讀取Meta信息並向Hadoop隊列提交任務. 在這一過程當中, 用戶須要自行提供計算須要的隊列資源。隨着近幾年對大數據平臺的要求愈來愈高， 咱們在使用QueryEngine過程當中也發現了一些問題：首先QueryEngine須要由用戶提供計算資源, 這使得數據倉庫的用戶須要去了解Hadoop以及相關的管理流程, 這增長了用戶使用數據的門檻。 第二, 對於不少小型計算任務而言, MR的任務的起動時間也較長, 每每用戶的計算任務只須要1分鐘, 可是排隊/提交任務就須要超過一分鐘甚至更長的時間。這樣的結果是，QueryEngine雖然很好的支持線下執行時間長的任務，可是對線上的一些交換式查詢任務（要求延時在一到兩分鐘內）確是無能爲力。

圖1: Query Engine 的執行流程

爲了解決這些問題, 在大約一年前，咱們嘗試在離線計算的技術棧上搭建起一套具備在線服務屬性的SQL計算服務 PINGO。如圖2所示: PINGO使用了SparkSQL爲主要的執行引擎， 主要是由於Spark具備下面的特色:

• 內存計算：Spark以RDD形式把許多數據存放在內存中，儘可能減小落盤，從而提高了計算性能。

• 可常駐服務：Spark能夠幫助實現常駐式的計算服務, 而傳統的Hadoop作不到這一點。常駐式的計算服務有助於下降數據查詢服務的響應延遲。

• 機器學習支持：對於數據倉庫的使用, 不該僅僅侷限於SQL類的統計任務。 Spark的機器學習庫能夠幫助咱們未來擴展數據倉庫, 提供的交互式的數據挖掘功能。

• 計算功能多元：雖然PINGO是一個查詢服務, 不過仍然有其餘類型的計算需求, 如數據預處理等。 使用Spark可使咱們用一個計算引擎完成全部的任務, 簡化系統的設計。

圖2: PINGO第一版系統架構設計

PINGO系統迭代

在過去一年中，PINGO從一個雛形開始，經過跟不一樣業務的結合，逐漸的演變成一套成熟高效的系統。中間經歷過幾回的架構演變，在本章中，咱們將詳細介紹PINGO的迭代過程。

PINGO 1.0

PINGO第一版的目標是提高性能，讓其能支持交互式查詢的任務。因爲Spark是基於內存的計算引擎，相對於Hive有必定的性能優點， 因此第一步咱們選擇使用Spark SQL。爲求簡單，最初的服務是搭建在Spark Standalone集羣上的。咱們知道, Spark在Standalone模式下是不支持資源伸縮的, 一個Spark Application在啓動的時候會根據配置獲取計算資源。 這時就算一個查詢請求只有一個Task還在運行, 該Application所佔用的全部資源都不能釋放。好在一個Spark Application能夠同時運行多個Job, 每一個Job都可以’平分’其所佔用的計算資源。

圖3: PINGO 1.0 系統架構

基於上面的考慮, 如圖3所示，咱們利用Spark的Standalone集羣搭建了初版PINGO服務. 咱們的主服務節點叫作Operation Manager, 它自己也是一個Spark Driver。全部用戶的請求都會從Magi Service發送到Magi Worker, 再由Magi Worker分配給Operation Manager, 經過Operation Manager在Spark集羣上執行。

PINGO 1.1

經過PINGO 1.0, 咱們把數據倉庫的計算引擎從Hive/MR轉變成了Spark。 不過, 單純的替換計算引擎, 也許能夠把查詢的性能提高1-2倍, 可是並不會使查詢的速度有數量級上的提升. 想要從本質上提升查詢速度, 咱們還須要改進存儲。對此咱們設計了PINGO 1.1， 加入了以Tachyon爲載體的緩存系統，並在緩存層上搭建了緩存管理系統ViewManager, 進一步提高系統性能。

不少快速的查詢系統都將Parquet之類列存儲格式和分佈式KeyValue存儲引擎結合起來, 經過創建索引/物化視圖等手段加速SQL查詢的速度. 經過將部分查詢條件下推到存儲引擎, 某些SQL查詢語句甚至能夠被提速至100倍以上。然而咱們的數據都存儲在舊版本的Hive數據倉庫中, 文件系統被限定爲HDFS, 數據格式通常爲低版本的ORC File, 也有很多表是ORCFile或者文本。 由於遷移成本/數據上下游依賴等兼容性等緣由, 咱們沒有辦法更改Hive數據倉庫自己的存儲。

不過根據局部性原理, 任何數據訪問都有熱點. 咱們能夠創建緩存系統, 在緩存系統中應用最新的文件系統和存儲格式. 經過把熱點輸入經過預加載的方式導入到緩存系統, 咱們能夠加速整個系統的訪問性能.爲此, 咱們設計瞭如下的系統架構：

圖4: PINGO 1.1 系統架構

在這個架構中, 咱們引入了一個新模塊 ViewManager, 該模塊負責管理緩存系統。 它的主要功能是識別熱點數據, 將數據導入到緩存系統中, 並在查詢時改變SQL的執行過程, 使得Spark從緩存而不是原始位置讀取數據。在這個架構下，當一個Query進來時，會先向OperationManager請求。當接受到請求後，OperationManager會向ViewManager查詢數據是否已經在緩存中。若是已經在緩存，數據將被直接返回給用戶， Query完成。若是數據不在緩存中，OperationManager會向底層Data Warehouse發起請求， 等數據到達時返回給用戶。同時，ViewManager也會向底層Data Warehouse發起請求， 等數據到達時創建一個Cache Entry， 這樣的話下次一樣Query進來時，就會從緩存讀數據。注意這裏咱們OperationManager與ViewManager對會向底層Data Warehouse的數據讀取是兩個獨立的STREAM， 這樣保證二者不會互相干擾。

那PINGO又是如何動態去讀取緩存中或者底層Data Warehouse的數據的呢？畢竟在Query Plan中，數據地址是寫死的。 爲了可以讓緩存對用戶透明, 咱們在SparkSQL上作了一些改進, 以下圖所述

圖5: PINGO1.1 對SparkSQL的改進

Spark中利用Catalyst框架來執行SQL語句。 對於Catalyst框架來講, 其輸入是Unresolved Logical Plan, 這能夠簡單理解爲SQL語句的結構化描述。 Catalyst調用Catalog來向MetaService查詢表的元數據, Catalog返回MetastoreRelation來表明Hive表, 其中含有讀取該表的全部必要信息,以便後續的解析和優化。 而在執行時, Catalyst會把MetastoreRelation轉化爲HiveTableScan, 來完成對數據的讀取。

爲了實現對緩存的透明利用, 咱們在其中兩個地方了作了擴展。 首先咱們在Catalog中爲咱們緩存的表返回CachableRelation來代替MetastoreRelation。 而在將LogicalPlan轉變爲真正執行的PhysicalPlan時, 咱們會把CachableRelation翻譯爲兩種TableScan的Union, 一個針對那些被緩存住的數據, 另一個針對那些沒有被緩存住的數據。

經過這種方式, 咱們可以作到在用戶不感知的狀況下, 完成對數據倉庫存儲層的替換和優化。 目前咱們作的僅僅是改變存儲的文件系統和格式, 未來也會將一樣的實現思路應用到索引和物化視圖上。

PINGO 1.2

PINGO 1.1服務很好的提升了系統性能， 可是在運營了一段時間以後, 咱們逐漸感受到Spark的集羣管理問題正在成爲整個系統的瓶頸。這具體表如今兩個方面

1. 咱們的整個服務實際上是一個Spark Application, 服務的主節點同時是Spark的Driver。 而咱們知道, Spark並不以高可靠性見長, 咱們在實踐中也發現把全部的計算壓力放到單個Spark Application容易致使比較嚴重的GC問題和負載問題。 而當Spark出問題須要重啓的時候, 咱們的整個服務也就暫停了。

   
   

2. 單一Spark集羣的架構沒法適應多機房的基礎設施。 百度內部的離線集羣規模仍是很大的, 機房分佈在全國的多個地區。 這樣, 咱們的服務可以獲取的資源每每來自多個機房, 而多個機房的機器是不適合組成一個Spark集羣的。 另外, 當集羣讀取其餘機房的數據時, 帶寬每每成爲整個計算任務的瓶頸。

   
   

發現這些問題是好事，說明系統已經逐漸成熟，開始往怎麼更好的運維的方向發展了。 爲了解決上面的問題, 咱們對系統進行了進一步的迭代. 迭代後的架構以下圖所示：

圖6: PINGO 1.2 系統架構

在這一版架構中, 咱們將PINGO的服務和調度功能獨立出來, 與真正執行計算的部分剝離。 支持了單一查詢入口PinoMaster進行調度, 多個集羣Pingo Applicatoin執行計算的多集羣功能。 PingoMaster同時維護多個Spark Application。 當一個Query到來時, 咱們會根據集羣的歸屬/存儲的位置選擇一個最優的Application執行任務。 另外, 這一版本的PINGO還支持在yarn集羣上起動Application。 百度內部有本身的資源管理系統, 提供了和yarn兼容的接口. 經過遷移PINGO服務到yarn上避免了本來Standalone版本須要的不少運維工做, 而且能夠經過按期重啓Application來解決Spark的可靠性問題。

爲了可以在PINGO中實現更好的調度策略, 咱們也對Spark SQL進行了深度擴展。

圖7: PINGO1.2 對SparkSQL的改進

當Pingo收到一個Query後, 咱們在Master端就完成對這條Query的分析和部分優化, 並將這些信息保存到QueryContext中。 當SQL在Application端真正執行的時候, 咱們從Master端而不是Meta端拿到執行所須要的信息. 這樣作的好處是能夠在根據Query的內容來作出調度. 基於這個執行流程, 目前咱們開發了兩個比較有效的調度策略：

1. 根據數據的存儲位置進行調度。 由於咱們在Master端就可以知道Query所需數據的存儲位置, 因此能夠選擇合適的PingoApplication來執行這條Query。

   
   

2. 根據數據量大小進行調度. 如上文所說, 一個Spark Aplication能夠支持同時運行多個Job, 可是在不少大型的Job同時運行在一個Application又會形成FullGC等穩定性問題. 咱們會根據Query輸入量的大小, 爲大型Query單獨啓動Application, 而讓小Query複用同一個Application。 這樣的好處是對於多數的小Query , 咱們節省了起動Application的開銷, 而又避免了大Query形成的穩定性問題。
   

   
   

PINGO性能評估

在上一章中，咱們詳細介紹了PINGO架構的迭代，在本章中，咱們重點看一下PINGO的性能。如圖8所示，首先咱們對比了使用Hive以及使用Spark做爲計算引擎的性能。 這裏的Benchmark選取的Query是百度內部交互式數據分析的典型負載, 主要是Join/Aggregate等操做, 輸入的數據量從100M到2T左右.咱們能夠看到, Spark相比Hive有較大的性能優點。在比較大比較複雜的Query中， Spark取得了2到3倍的加速比。注意在這個對比中，內存資源有限，用的是64GB內存的機器，不少Spark的數據被迫落盤， 若是有更多內存，加速比會更高。

圖8: Query執行時間: Hive vs. Spark

下一步咱們瞭解一下加了緩存後的性能， 在這個實驗中，咱們使用了192GB的機器，有更多的內存去支持緩存以及內存計算。如圖9所示，在這個環境下，使用Spark相對於Hive有大概5到7倍的提速。 在加入了Tachyon後，相對於Hive無緩存的狀況有30到50倍的提速。咱們相信在將來的幾年內，內存價格會大幅下降，內存計算會變成主流，因此使用內存作緩存是一個正確的方向。

圖9: 緩存性能提升

PINGO服務目前被應用在交互式查詢場景中, 旨在爲PM和RD提供快速的數據分析服務。 圖10展現了在一樣的計算資源下, 在生產環境使用PINGO先後的Query執行時間分佈圖。注意，在這個生產環境中，咱們用的是64GB內存的機器， 爲了提供足夠的緩存空間，咱們使用了Tachyon的Tiered Storage功能，讓緩存分佈在內存以及本地磁盤中。 咱們能夠看到, 在傳統的基於Hive+MR的服務模式下, 只有1%左右的Query可以在兩分鐘內完成. 而採用了基於Spark的PINGO服務, 有50%+的Query能夠在兩分鐘內執行完成。 可以取得這樣的加速效果, 部分緣由是Spark自己的執行效率比Hive要高一些。這個自己還有很大的優化空間，好比若是咱們使用內存緩存的話，執行時間能夠輕易的壓縮到30秒內。

圖10: Query執行時間: Hive vs. PINGO

結論和展望

通過過去一年的迭代與發展，PINGO已經很好的支持了百度內部的交互式查詢業務。經過PINGO，不少查詢的延時從之前的30到40分鐘，下降到了2分鐘之內，很大的提升了查詢者的工做效率。從此PINGO的發展將朝着更好用，已經更快兩個方向發展。爲了讓PINGO更好用，咱們正在PINGO之上開發UI， 讓用戶與圖形界面交互（而不是經過輸入SQL Query）就能輕易的查詢到想要的數據。爲了讓PINGO更快，更有交互性，咱們但願能夠把90%以上的Query 時間下降到30秒如下。第一步就是要擴大咱們的Cache的覆蓋率與性能，設計出更好的緩存預取以及緩存替換機制，而且加速Cache讀取的延時。第二步，咱們也經過硬件加速的方法使用FPGA加速某些經常使用的SQL Operator。就在最近，咱們對SQL的JOIN Operator進行了FPGA硬件加速，相對CPU取得了3倍的加速比。咱們也會很快把這項技術使用到PINGO上。