阿里的程序員們如何解決複雜數據的查詢優化問題？

數據分布的問題在大數據處理領域由來已久。很不幸，如今流行的大數據處理系統仍然沒有很好地解決這個問題。在MaxCompute 2.0全新的優化器中，我們引入了複雜數據分布，添加了分區剪枝、分布上拉、下推以及分布對齊等優化措施。

本文將從數據分布的歷史和原理開始，介紹我們的思路和解決辦法。

理解數據分布

提到數據分布，很多人會想到MPP DBMS。的確，我們通常說只有MPP DBMS才需要考慮數據分布優化。先考慮一個流行的分散式資料庫分類學：

Shared Everything：區別於後兩類，這一類基本不是分散式的。

Shared Disk：資料庫伺服器可以橫向擴展，他們本身沒有存儲器，通過SAN或NAS技術連接到後端同樣可以橫向擴展的統一存儲。受限於這層網路連接，資料庫伺服器的擴展能力非常有限。Oracle RAC等商業分散式資料庫屬於這類。

Shared Nothing：區別於Shared Disk，這種架構讓資料庫伺服器和存儲落在相同的物理節點上（co-located），使得物理節點之間不share任何信息，這大幅減少了網路IO。MPP DBMS和Hadoop屬於這類。

顯然，只有Shared Nothing的資料庫才需要考慮數據分布，你需要預知怎樣把數據分布到不同的物理節點（而不是像Shared Disk那樣放在統一存儲），會使後續的操作代價更小。例如，在Greenplum中，必須在建表時指定partition key，系統會按照指定的key（哈希）分布數據。如果Join的兩張表都按照join key來partition，這個Join就不需要網路IO。如果其中一張表使用了另一組partition key，那麼可能要做一次re-partition。

這就是為什麼要理解數據分布的原因：它對應用優化和系統優化都是非常重要的。MPP DBMS在數據分布上都有比較深的積累。但是為什麼Hadoop這種大數據處理系統沒有這類優化？是因為他們需要更強的擴展能力（以及非結構化數據支持，我們不展開這個話題）。

區別於MPP，Hadoop並不是在物理上強制數據和計算在相同節點，如果這麼做，系統的橫向擴展能力仍然受限。特別是動態擴展能力，考慮正在運行的一個50個節點的Greenplum集群，我們基本無法做到快速地加入例如2個節點還能高效工作。Hadoop在這方面是很在行的，它的解決辦法主要是：

存儲計算分離；

去中心化的設計支持高效的peer to peer讀寫（HDFS）。

這就是為什麼你在Hive中創建一張表時，無須像Greenplum中那樣指定partition key，同時Hive在Join的效率低於Greenplum的原因。

數據分布優化的目的

如上文所述，大數據分散式系統在存儲系統上通常傾向隨機分布，這提升了擴展性，犧牲了性能。但是重新審視這個權衡，在存儲系統上隨機分布並不意味著我們不能利用數據分布優化查詢。分布優化的目的是希望儘可能的利用已經存在的分布，並儘可能滿足未來要求的分布。這種優化包括：

分區剪枝：利用數據分布特性，我們可以做分區剪枝來減少數據讀取。例如，哈希分布對於點查詢，範圍分布對於區間查詢可以應用分區剪枝。

消除重分布：如果當前的分布滿足後續演算法的要求，我們可以消除額外的重分布操作。眾所周知，重分布（在Hadoop中叫做shuffle）是分散式演算法最主要的消耗。

避免數據傾斜：可以使用更好的數據分布演算法避免數據傾斜。例如，某些單值重複率很高（end-biased）的數據集，使用範圍分布而不是哈希分布可能會有效避免數據傾斜帶來的性能影響。

數據分布類型

數據分布類型和對應的意義和範例如下所示：

實現

在不破壞Volcano優化器語義的前提下，我們把分布特性實現為一種physical property，稱作distribution。和其他property一樣，它有required property和delivered property成對的屬性。例如，對於sorted merge join，它對所有輸入會施加一個Partial Ordered的required property，同時自身會deliver一個Partial Ordered property，這使得它的後繼操作有機會利用這個property，避免一次重新分布。考慮以下查詢：

此時Join如果被實現為Sorted Merge Join，它可能會deliver一個Hash[uid]的property，這正好被Aggregate要求，那麼這裡我們就可以省去一次不必要的重分布操作。

要做到類似的優化效果，我們需要關注下列問題：

收集分布特性；

（局部關係代數編譯）選擇合適的分布特性；

（全部代價計算上）規避不合適的分布特性。

收集分布特性

產生數據分布有3種途徑：

用戶指定：就像MPP那樣，可以在DDL中引入partition key，允許用戶指定數據分布。當然區別於MPP，這種分布僅要求在分散式文件系統上的目錄結構，並不能關聯具體的物理節點。

SQL邏輯：SQL邏輯可能產生一次運行時的數據分布。例如distribute by字句聲明了一次運行時的數據分布。

演算法的副作用：每個分散式演算法可能產生一次運行時數據分布。例如，sorted merge join可以保證它的輸出數據滿足按join key的有序和哈希分布的特徵。

有若干演算法要求一種特殊的數據分布：

Aggregate：Sorted Aggregate要求grouping key的Hash分布；

Join：Sorted Merge Join和Hash Join都要求輸入按照join key的相同Hash分布；

Sort：Order by要求sort key上的Range分布，或Singleton分布。

選擇合適的分布特性

即使給定了一系列required和delivered distribution property， 確定某個操作的分布仍然不是容易的事情。區別於ordering property（僅有排序列和升降序的屬性），distribution property的變化很多，這些變化的原因包括：

滿足要求的分布有多種選擇。例如group by a, b, c這個aggregate，對輸入有按a, b, c的Partial Ordered的要求，它對ordering的要求是a, b, c有序，但是滿足它的分布可以是Hash(a)、Hash(b)、Hash(a,b)、Hash(a,b,c)、RNG(a)等不同的組合。

能利用的實現分布有多種選擇。例如join a and b on a.id = b.id這個join，如果a服從Hash[id](10)，b服從Hash[id](20)，對於Sorted Merge Join，它可以選擇要求Hash[id](10)或Hash[id](20)，甚至任意Hash(id)。

這些複雜度加大了最優計劃的搜索空間。事實上，最優計劃是相對於關係代數數量的一個NPC問題。為了縮小搜索空間，我們引入了啟發式的分支選擇演算法。在編譯一個關係代數時，不僅要滿足後繼操作的要求，還要考慮前序操作提供滿足分布的可能性，後者被稱作Pulled Up Property的模塊。

Pulled Up Property猜測並篩選可能的前序delivered property，用於在編譯時減少搜索寬度。考慮上圖的查詢，在Join編譯時，因為Sink的需求下推，它被要求提供一個Hash[c1](30)。Pulled Up Property則從前序操作猜測可能會提供Hash[c1](10)和Hash[c1](15)，綜合考慮，Join可能會直接要求Hash[c1](30)，從而減少了Hash[c1](10)和Hash[c1](15)這兩個分支。

規避不合適的分布特性

數據傾斜（Skew）是指在分布中少量節點被分配了大部分數據，導致整個演算法退化為單機操作。低並發（Under Partition）是指分布指定了過少的節點，是的分散式資源不能被有效利用。我們希望能避免這兩種情況。

很顯然，更好的統計信息會幫助我們規避這兩種情況。Skew和Under Partition的情況下，需要對代價估計做相應的懲罰，降低他們被選為最優計劃的可能性。我們定義」好」的分布是每個節點處理的數據量在一個預設的範圍，低於或高於這個範圍都會被施加懲罰。估計這個數據量的依據包括：

輸入數據記錄數（row count）；

重複度最高的數據（top values）；

直方圖（histogram）。

總結

在這篇文章中，我們介紹了數據分布優化的問題和意義，並解釋了MaxCompute在數據分布優化上的實踐。這一優化效果已經體現在MaxCompute最新的發布中。

從我們的測試來看，這個優化有相當顯著的效果。我們對TPC-H進行了適當分區後，整體性能提升在20%的量級。即使沒有對錶數據分區，對用戶完全透明的運行時分區優化也有很好的效果。在我們線上運行的環境中，14%的查詢因為這個優化減少了至少一次數據重分布。

作者：徐冬，阿里巴巴計算平台事業部高級技術專家，負責實現MaxCompute的基於代價的優化器。擁有多年分散式大數據處理領域經驗，參與過多個分散式系統的搭建、開發與維護，是Apache Hive、Apache Calcite的貢獻者。

聲明：本文為作者投稿，版權歸其所有。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！