一條SQL在 MaxCompute 分佈式系統中的旅程

MaxCompute–面向企業的超大規模計算

• 全託管，多租戶，超大規模平臺
  
  MaxCompute擁有龐大的用戶羣體，支持阿里集團的各個關鍵業務和複雜場景，支持多個互聯網新興企業核心業務，以及支持關係國計民生、國家安全的關鍵行業。同時擁有超大規模計算存儲，包括單日千萬以上計算任務、多EB級別存儲量、10萬臺以上服務器以及全球超過十個數據中心。
• 企業級高性能計算引擎
  
  TPC-BigBench是更面向於大數據運算的BigBench，覆蓋一些複雜類型，包括機器學習場景，更貼近於大數據場景的業務。在2017年，阿里的TPC-BigBench首個經過100TB 驗證的引擎。在2018年，TPC-BigBench 首個達到18000+引擎。在2019年，進一步提高到 25000+，正式公佈到TPC官網。

MaxCompute不只僅在阿里集團內部被普遍的使用，也支撐着許多著名的互聯網方面的廠商，以及關係到國計民生、國家安全方面的應用。

超大規模企業級SQL引擎–MaxCompute UniSQL

一條SQL在分佈式系統中的旅程



上圖所示爲運行SQL任務中的大概流程。首先使用一條SQL語句，經過Compile，能夠生成一個邏輯的執行計劃，這個邏輯的執行計劃是計算機可以理解的，再通過Optimize過程，不管邏輯計劃有多複雜，都要翻譯成針對目前集羣和運行時刻的Runtime最優的物理執行計劃，對於每個Optimize不必定與原始的SQL相關。而後通過計算調度框架，使得合理快速的安排執行任務。由調度框架作的安排應用到每臺機器以後，每臺機器都會有一個SQL的運行時（Runtime Engine），它是真正可以理解物理執行計劃的，而且一步一步把數據從Storage中讀出來，再通過Shuffle獲得結果，最後返回到Storage中。能夠看出，運行時自己的性能是很是關鍵的，僅僅一條SQL語句有可能消耗幾百T的data，這時，Storage的性能也是相當重要的。

SQL的功能

• Not Only SQL – 腳本模式
  
  

上圖爲SQL的一個腳本，上面是配置語句，下面是建立的表，每句都是SQL的語句，可是這些SQL語句均可以串在一個腳本里，當表述一個很是複雜的邏輯時，不須要把腳本寫成嵌套的形式，這種方式更靈活，可以支撐更復雜的業務場景。阿里內部有很是複雜的業務場景，過去不支持這種方式時，用戶是使用嵌套的方式，使腳本變得複雜和扭曲，而且有大量的重複，以至不能解決時，就會將其拆分，再經過外部調動的方式串連起來。由於人若是承受不了維護的代價，就要引用額外性能的開銷，後面的語句就要引用前面的語句。無論腳本有多複雜，通過編譯器以後，仍是一個單一完整的執行計劃，並不會帶來額外的開銷。優化器看到的上下文越多，優化的機會越多，造成單一完整的執行計劃以後，就能夠以最高效的方式執行整個業務模式。此外，DataWorks也是支持這種模式的。經過腳本模式能夠效仿C++或者Java來寫SQL。

• Not Only SQL– 參數化視圖
  
  

例如，在寫C++或者Java時，常常在公共的邏輯中抽取出來一個函數，把公共的邏輯放到某一個模塊裏，這個過程視爲代碼的重用機制。可是，標準的SQL，尤爲是大數據的SQL是缺少這種機制的。對於阿里這麼複雜的場景，這種需求是很迫切的。底層的數據集提供了各個部門都須要的基礎的數據，不一樣的業務部門可能都要消耗這部分數據，消耗的方式是不一樣的。這時，也想像C++或者Java那樣抽取一個函數，在MaxCompute中是能夠實現的。在MaxCompute裏，上圖中的紅框除了做爲普通的view，也能夠封裝一些SQL複雜的邏輯和對數據的讀取，能夠把表的變量傳入，這時就能夠實現像C++或者Java中函數的功能，能夠把SQL裏公共的業務邏輯封裝在一塊兒，同時結合上文的腳本模式，參數化視圖就能夠組織很是複雜的SQL的業務邏輯用來支撐很是複雜的業務場景。

• Not Only SQL – IF/ELSE
  
  

通常的大數據不支持IF/ELSE，但對於IF/ELSE是有需求的。例如，每週作一次全量的計算，但天天只作增量的計算，若是沒有IF/ELSE的支持，就須要把腳本拆成兩個，經過調度的框架串連起來。可是，在MaxCompute中結合腳本模式，能夠直接的寫入IF語句或SELECT語句，若是返回的是異常的結果，直接能夠放在一個表達式裏，決定執行SQL的分支。全部的SQL的功能都是針對複雜的應用場景的須要。

• Not Only SQL – UDT & TRANSFROM
  
  

普通的SQL都會有基本的數據類型，有時也有複雜類型，但都是屬於給定範圍的數據類型。當數據類型特別複雜時，在MaxCompute裏能夠直接使用。右側框架是將Java和SQL無縫的融合在一塊兒，無需UDF封裝。左側爲SELECT TRANSFORM，是直接就能夠在SQL裏調用shell腳本，而且徹底兼容Hive。

SQL的性能

• SQL Engine for Huge Data - Adaptive Join
  
  

Adaptive Join包括Hash Join和Merge Join。Hash Join的性能是比較好的，但有時碰到不合適的場景時，特別是有很是多的Hash衝突時，性能就會變得不好。Merge Join的特色是可以提供一個性能的下限。能夠經過動態的選擇適合哪一種場景，以便作智能的選取。

• SQL Engine for Huge Data – Advanced Shuffle
  
  

Shuffle也有針對特定大規模系統的優化，包括提高Shuffle 70%的性能，提高大規模共享集羣性能，提高穩定性，下降IO壓力。具體包括如下優化方式：

一、Greysort模式(Mapper不排序，Reducer排序），增長與下游流水線機會；下游轉化爲HashJoin時消除排序 
二、Encoding & Adaptive列式壓縮，下降IO與Cache Miss 
三、優化內存結構，下降Working Set Size並消除Pointer Chasing

企業級分佈式智能調度執行框架

打造企業級分佈式調度執行系統



整個系統的發展有兩個維度，一個維度是系統的規模，隨着系統規模的不斷成長，對於分佈式調度執行系統要面對天天千萬級須要解決的問題，在阿里這個大致量的數據下，單個分佈式做業規模已經能達到數十萬個計算節點，已經有上百億鏈接和運行數萬臺的物理機。

另外一個維度是系統的成熟度，一個系統成爲企業級的分佈式執行調度系統就須要達到成熟度，包括三個階段，第一個階段是可用性（正確性），一個做業在單機系統上執行的結果和分佈式系統上執行的結果是不同的，尤爲是在系統的超大規模上，在面對系統各類各樣的節點失敗問題、網絡層的失敗問題和各類容災問題時，怎樣經過正確的方式能保證做業正確的產出是很重要的。第二階段是夠用，是指每個計算的系統都要鍛造本身的性能，能在各類各樣的benchmark上標準結果，經過此方法來提高性能。第三個階段是好用（智能化），是指在動態執行過程當中擁有動態能力和自適應能力，能夠根據做業的不一樣特色來調整做業執行的計劃。

企業級分佈式計算調度框架

企業級分佈式計算調度框架分爲三個階段：

• 動態的智能執行
  
  

上圖所示爲阿里的一個做業在離開優化器之後，在分佈式系統裏執行的過程。能夠理解爲從邏輯圖到物理圖映射的過程。



上圖所示爲三個階段的做業，第一個階段是做業提交開始運行，第二個階段是根據實際產出動態調整併發，第三個階段是產生所需數據提早結束做業。



上圖所示爲智能化DAG執行的動態邏輯圖，包括Sorted Merge Join和Broadcast Join兩種算法。其中Sorted Merge Join的特色包括經典分佈式join算法，可支持大規模做業，可用範圍廣(slow but reliable)，代價較昂貴 (full shuffle + sort)，且shuffle可能帶來數據傾斜。Broadcast Join的特色是隻適用特定類型做業 (一路輸入可載入單計算節點內存)，非適用場景上可能致使OOM，做業失敗。



對動態的選擇執行計劃，在理想狀況下都但願數據的分佈是均勻的，而且能夠理解數據的特性，因此優化器均可以作出「最佳」的計劃，尤爲是在作benchmark時，可是因爲源數據統計不許確 、中間數據特性波動 ，所產生數據的特色是沒有辦法提早預估的，因此容許優化器來給一個非肯定的執行計劃（Conditional Join），這時，優化器會給出兩個執行路徑的計劃，調度執行框架能夠根據上游實際產生的數據量，動態的調整邏輯圖的執行。



上圖所示爲併發度的例子。簡單的併發調整是根據上游總數據量直接取平均做爲併發，僅支持向下調整，但問題是數據多是傾斜的，這種方法已再也不適用。下面給出兩種新的調度方法：

一、依據分區數據統計調整：避免併發調整加劇數據傾斜，可向上向下調整。
二、分區統計基礎上，自動切分大分區：雙重調整，消除分區內的數據傾斜，並支持數據處理歸併，以保留分區特性。

• 高效做業管理
  
  

對於阿里如此大規模的做業，調度的敏捷度是十分重要的，由於集羣規模很大，一個做業怎樣理解各個計算節點和物理機的狀態，作智能的容錯和預判性的容錯是阿里所作的一項工做。隨着做業規模愈來愈大，一個很是優秀的調度框架能帶來的性能提高會愈來愈明顯。

• 多種計算模型融合
  
  

阿里整個計算平臺做爲飛天的底座，不只僅運行SQL，也有可能運行其餘。最經典的SQL是batch執行。離線和一體式的執行是資源利用率和性能優化的兩個極端，做爲一個用戶，會同時關注執行性能和資源利用率，須要思考的問題是，怎樣在兩個點中達到平衡。所以，阿里也支持一種稱爲bubble的調度，所謂bubble調度是容許一個做業的子圖同時調度，下游的子圖分佈調度，在不一樣的SQL上會有不一樣的效果。例如。在TPCH11的狀況下，相對於離線（batch）會有66%的性能提高，相對於一體式（all-in-one）會節省3倍的資源，同時獲取95%的性能。

新一代列式存儲引擎AliOrc

在AliOrc的里程中，起點和終點都是在存儲層，數據的讀和寫是AliOrc執行的開始和結束，存儲引擎做爲AliOrc的底座，承擔着一個很是重要的做用。

基於Apache Orc的深度優化



整個計算引擎是基於列結構的，技術的出發點是Apache Orc。在此基礎上，阿里作了不少深度的優化，包括I/O維度、內存優化、索引和數據編碼壓縮。其中有一部分已經貢獻到了社區。

新一代列式存儲引擎

新一代列式存儲引擎包括如下技術方面：

• 並行化編碼技術
  
  

對於有一系列的大數和小數，直接存放時會產生4個字節，而對於小數，前面會產生不少的零，這些零是沒有意義的。並行化編碼技術的主要思想就是將冗餘的信息刪掉，將真正有意義的batch留下，而且pack到一塊兒。這種編碼方式的好處是能實現並行化。此外，還進行了一些擴展，包括對有序數據的優化，以及對數據的編碼優化。同時從新設計了編碼存儲格式，更利於內存對齊，以及列存儲。



從測試的結果來看，此編碼技術比傳統遊程編碼速度快4到6倍，壓縮率提高大概10%左右，在反應到TPC Benchmark表掃描效率提高24%。之因此有如此快的結果，是由於使用AVX256一條指令能夠處理8個64位數，或者16個32位數，同時充分利用函數模板展開，最大程度避免循環和分支預測失敗。

• 異步並行IO
  
  

阿里是屬於列存儲引擎的，是指在同一個列是放在一塊兒的，好處是在讀數據時選擇幾個列放到存儲引擎中去讀，就不須要讀全部的列。假設在上圖中的場景中，有三個列爲A、B、C。最先的IO模型是串行的，存在許多等待時間。所以，阿里作了一個改進爲Prefetch模型，IO是不須要一個一個發出去的，在一開始時能夠將三個讀取引擎一塊兒發出去，可是須要一個一個的等待它們回來，雖然有了一些提高，可是還仍然存在IO等待的時間。目前爲止，改進的模型爲Prefetch+Async Paraller IO，是將IO所有並行化，將三個一塊兒發出去以後，並不須要按照原來A、B、C的順序等待，能夠按照回來的順序作解壓和解碼。這樣作能夠對IO等待的時間降到最小。



如上圖所示，異步並行IO與同步讀取相比較，IO等待時間減小97%，端到端時間減小45%。
 

• 延遲讀取、延遲解碼
  
  爲了進一步的提升性能，減小數據讀取量，從而減小數據解碼、解壓縮成爲了關鍵。 

如上圖所示爲延遲讀取的一個例子，經過只讀取DEPT列，把ADDRESS以及SALARY列延遲到過濾以後讀取，能夠大幅減小了沒必要要的數據讀取。 



對於字符串類型的列，有一種方法叫字典編碼，是指將字符串裏不同的Key找出而且給予ID，這時，數據在存放時是不須要存放整個字符串的，只須要存放ID就能夠。可是使用此方法是很耗時的。由此，作了如下改進：



使用延遲解碼，跳過解碼步驟，直接在字典上匹配，再以ID到數據列搜索。好處是減小了字符串匹配次數以及減小了字典解碼時間。



如上圖所示，對打開延遲讀寫和沒有打開延遲讀寫作了比較，橫座標爲filter過濾的數據，「1」表示沒有過濾，縱軸是花費的時間，實現延遲讀取以後，讀取數據量隨Selectivity的提高而減小，讀取時間也相應大幅下降。

 

 

 

原文連接

本文爲雲棲社區原創內容，未經容許不得轉載。