757
SQL優化器原理－Shuffle優化
這是MaxCompute有關SQL優化器原理的係列文章之一。我們會陸續推出SQL優化器有關優化規則和框架的其他文章。添加釘釘群“關係代數優化技術”（群號11719083）可以獲取最新文章發布動態。
本文主要介紹MaxCompute Optimizer對Shuffle方麵的相關優化。
1 簡介
分布式係統中，Shuffle是重操作之一，直接影響到了SQL運行時的效率。Join、Aggregate等操作符都需要借助Shuffle操作符，確保相同數據分發到同一機器或Instance中，才可以進行Join、Aggregate操作。對於這些需要經常使用到Shuffle場景，如何減少Shuffle，刪除一些不必要的Shuffle是提升性能的一個關鍵。
例如
select count(*), c1 from t group by c1;
假設t表如果是Hash Clustering Table［2］，則GroupBy的計算全部可以本地化處理，不需要再次Shuffle，因為相同c1值已在同一台機器了。
Optimizer Plan
OdpsPhysicalProject(cnt=[$1], c1=[$0]): rowcount = 1.5, cumulative cost = {6.30 rows, 3.00 cpu, 34.40 io, 0.00 memory, 0.00 network}, id = 281
OdpsPhysicalSortedAggregate(group=[{0}], __agg_0=[COUNT()]): rowcount = 1.5, cumulative cost = {4.80 rows, 3.00 cpu, 26.40 io, 0.00 memory, 0.00 network}, id = 280
OdpsPhysicalTableScan(table=[[wlz_p_02.t_test, c1(1) {0}]]): rowcount = 3.0, cumulative cost = {3.30 rows, 0.00 cpu, 26.40 io, 0.00 memory, 0.00 network}, id = 206
隻需要一個Map Task就可以完成整個Query執行。後續會詳細介紹如何優化。
2 Shuffle優化原理
MaxCompute Optimizer是基於開源項目Calcite基礎上搭建的一套Optimizer框架。Calcite提供了Volcano模型的Planner，MaxCompute Optimizer引入了Volcano模型中Enforcer機製[1]來優化Shuffle。
簡單介紹下Enforcer概念，Enforcer是指操作符（算法，如Join的實現SortedMergeJoin、Aggregate的2Pass實現等）要求輸入數據必須具有一些物理數據屬性(Trait)，如order、distribution等。
如SortedMergeJoin要求輸入數據必須基於Join keys進行分布且有序，這些為了確保滿足SortedMergeJoin算法的要求。如果數據不是按照Join Keys分布，則相同Key值的數據不在同一個Instance裏，則無法達到Join的目的；如果數據不是基於Key值有序，則無法滿足Sorted Merge的要求。簡單講就是一個具體算法決定了輸入數據的要求。在分布式環境中，數據要求是通過Shuffle實現。而Shuffle數據特性，我們稱之為Trait。
2.1 Enforce Rule
如何確保對於任何算法滿足其數據特性Trait的要求？
MaxCompute Optimizer實現了一種叫Enforcer Rule，來保證隻要任何操作符對其輸入（Input）要求一個Trait，Enforcer Rule就會保證Input的操作符一定會提供這種特性的數據。
圖1 Enforcer Rule
圖1展現了Enforcer Rule的工作機製。
1）任何Operator（算法）會對Input產生一個Required Trait。如SortedMergeJoin，則要求每一路Input必須是基於Key的分布且有序，類似Trait(Hash(c1) sort(c1 asc))。這一步由Build Rule實現，即對於任何Operator當采用某種算法時，必須將Required Trait的要求下推給Input。如Join當生成SortedMergeJoin時，則SortedMergeJoin的Input必須帶有Required Trait。
2）Enforcer Rule捕獲Required Trait + Input的Pattern。即圖1中Required + Input Operator模式。
3）Shuffle生成。
Required+Input方式處理Shuffle提供了三種可能性：
A）情況1:Input Operator不能確保數據具有Trait特性，則直接在Input輸出後生成基於Trait的Shuffle。這樣Parent給到的要求得到了滿足。
B）情況2：Input Operator操作符已經具有了Required Trait的特性，則Shuffle就不需要添加，即達到了減少Shuffle的目的。Parent要求的Trait也得到了滿足。
C）情況3:Input Operator可以保證數據的特性可以傳遞，則可以將Required Trait繼續下推到自己的Input中。則Required Trait繼續由Input來滿足要求，而當前Input本身可以確保數據特性不會發生改變。這種情況下，Required Trait也得到了滿足。
任何一個Operator都是采用上述三種策略進行處理，從而使得Required Trait可以從Root Operator一直下推。
舉例：Required Trait ＋ Filter
當一個Required Trait與Filter這樣一個Pattern被捕獲時，如何保證Required Trait得到滿足呢？
如果將Filter理解為一個當輸入數據傳入給Filter Operator時，Filter Operator僅僅是將每一行的數據進行判斷是否滿足condition條件，如果不滿足condition條件的行數，則不輸出。所以發現Filter具有一個特性，即不會改變數據的輸入特性。所以當Required Trait要求Filter具有這種特性時，可以有兩種處理方式：
1）直接生成Shuffle。
2）將Required Trait繼續要求Filter的Input保持這種特性。
思考：是否可以直接將Required Trait下推到Filter Input得到滿足？
答：不可以。這裏有兩種選擇，即Shuffle是在Filter之前還是之後生成，這些由Optimizer另外一個特性來決定，即CBO（Cost-based Optimizer），也就是由Cost來決定選擇是1）還是2），因為繼續下推給Input，也有幾種情況，要麼生成Shuffle，要麼繼續下推等，這時Shuffle生成的位置則由CBO控製。
2.2 Operator算法
當根據Optimizer生成的Plan真正運行Operator時，必須嚴格要求如何保證數據特性來實現。如Filter，理論上實現時，可以不保證數據和其輸入數據的特性一致，如果是這樣，則這些優化都無法實現。因為Operator實現的算法不滿足要求。所以Optimizer與Runtime算法必須要求一致。
如SortedAggregate必須保證基於Group By key分布有序等。
3 應用場景
3.1簡單case
假設T1是clustering Table[2]
select count(*), c1 from t group by c1;
圖2 詳細實現步驟
圖2展示了Optimizer如何減少Shuffle的詳細步驟。
1）Pull Trait。根據Input來獲取可能會產生的Trait。
2）convert(input, Trait[hash])。Aggregate build成SortedAggregate，則要求Project基於Key hash且有序，即Trait[hash(c1) sort(c1 asc)]。
3）Enforcer處理Requried Trait[hash(c1) sort(c1 asc)] 與Project Pattern，將Trait下推給TableScan。
4）Requried Trait與TableScan發現Trait一致，則Shuffle不需要添加。從而達到了減少Shuffle的優化目的。
上述邏輯介紹了Optimizer如何一步一步達到減少Shuffle的目的。主要關鍵點是Operator算法要求Input 滿足Trait以及Trait與Input Pattern如何滿足Parent Operator的要求。
3.2 TPC-H
基於TPC-H，給出Q4的Shuffle優化例子。Q4特征: Join的輸入分別是Table和Group。且Join Key是Table的Clustering key和Group by key。
Q4:
create table q4_result_xx as
select
o_orderpriority,
count(*) as order_count
from
tpch_orders o
join
(select
distinct l_orderkey
from
(
select
*
from
tpch_lineitem
where
l_commitdate < l_receiptdate
) tab1
) tab2
on tab2.l_orderkey = o.o_orderkey
where
o.o_orderdate >= '1993-07-01' and o.o_orderdate < '1993-10-01'
group by
o_orderpriority;
圖3 MaxCompute Optimizer Plan
圖3中顯示SortedMergeJoin的兩路輸入都不存在Shuffle，同時Aggregate這路也沒有Shuffle，相當於減少了3次Shuffle（Aggregate一次＋Join兩次）。
圖4 優化Shuffle DAG
圖5 Shuffle不優化DAG
優化Shuffle VS不優化之間差別在於TableScan由於已經基於Key分布，所以Aggregate和Join的Required Trait都可以下推到TableScan，而TableScan都是基於這些key的Clustering Table，Required Trait得到滿足，從而Shuffle都不需要。優化Shuffle後的Plan，M1中執行了Aggregate以及Join整個操作，而不需要類似不優化方式，根據Shuffle將Aggregate和Join切分成不同Task進行處理。從性能上講，Shuffle優化方式耗時54s，而不優化方式耗時121s，性能提升一倍。
4 總結
本文主要從Optimizer實現角度上詳細講解優化Shuffle的實現原理以及一些應用場景。Shuffle優化對性能提升有較大幫助，目前主要應用在Clustering Table上，詳細性能測試對比可以見ATA Hash Clustering文章[2]。
索引
[1] Goetz Graefe. The Volcano Optimizer Generator: Extensibility and Efficient Search
[2] ATA: ODPS Hash Clustering支持 https://www.atatech.org/articles/85967
有任何Optimizer相關問題和反饋，請加入“關係代數優化技術”群獲取相關支持。
同時，有任何Hash Clustering問題和反饋，請聯係白駝，謝寧和龍重，同時也建議用戶加入到“ODPS Hash Clustering用戶群”以獲得最新消息以及技術支持。
最後更新：2017-08-25 12:02:35