596      阿裏雲  技術社區[雲棲]

MaxCompute有關優化複雜數據分布的實踐

這是MaxCompute有關SQL優化器原理的係列文章之一。我們會陸續推出SQL優化器有關優化規則和框架的其他文章。添加釘釘群“關係代數優化技術”（群號11719083）可以獲取最新文章發布動態。

概述

數據分布的問題在大數據處理領域由來已久。很不幸，如今流行的大數據處理係統仍然沒有很好地解決這個問題。在MaxCompute 2.0全新的優化器中，我們引入了複雜數據分布，添加了分區剪枝、分布上拉、下推以及分布對齊等優化措施。本文將從數據分布的曆史和原理開始，介紹我們的思路和解決辦法。

理解數據分布

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

1. Shared Everything: 區別於後兩類，這一類基本不是分布式的。
2. Shared Disk: 數據庫服務器可以橫向擴展，他們本身沒有存儲器，通過SAN或NAS技術連接到後端同樣可以橫向擴展的統一存儲。受限於這層網絡連接，數據庫服務器的擴展能力非常有限。Oracle RAC等商業分布式數據庫屬於這類。
3. 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在這方麵是很在行的，它的解決辦法主要是：

1. 存儲計算分離
2. 去中心化的設計支持高效的peer to peer讀寫（HDFS）

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

數據分布優化的目的

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

1. 分區剪枝：利用數據分布特性，我們可以做分區剪枝來減少數據讀取。例如，哈希分布對於點查詢，範圍分布對於區間查詢可以應用分區剪枝。
2. 消除重分布：如果當前的分布滿足後續算法的要求，我們可以消除額外的重分布操作。眾所周知，重分布（在Hadoop中叫做shuffle）是分布式算法最主要的消耗。
3. 避免數據傾斜：可以使用更好的數據分布算法避免數據傾斜。例如，某些單值重複率很高（end-biased）的數據集，使用範圍分布而不是哈希分布可能會有效避免數據傾斜帶來的性能影響。

定義

數據分布類型

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

轉化關係

實現

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

SELECT uid, count(*) FROM (
  SELECT uid FROM user JOIN line ON user.uid = line.uid
) GROUP BY uid

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

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

1. 收集分布特性
2. （局部關係代數編譯）選擇合適的分布特性
3. （全部代價計算上）規避不合適的分布特性

收集分布特性

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

1. 用戶指定：就像MPP那樣，可以在DDL中引入partition key，允許用戶指定數據分布。當然區別於MPP，這種分布僅要求在分布式文件係統上的目錄結構，並不能關聯具體的物理節點。
2. SQL邏輯：SQL邏輯可能產生一次運行時的數據分布。例如distribute by字句聲明了一次運行時的數據分布。
3. 算法的副作用：每個分布式算法可能產生一次運行時數據分布。例如，sorted merge join可以保證它的輸出數據滿足按join key的有序和哈希分布的特征。

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

1. Aggregate：Sorted Aggregate要求grouping key的Hash分布。
2. Join：Sorted Merge Join和Hash Join都要求輸入按照join key的相同Hash分布。
3. Sort：Order by要求sort key上的Range分布，或Singleton分布。

選擇合適的分布特性

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

1. 滿足要求的分布有多種選擇。例如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)等不同的組合。
2. 能利用的實現分布有多種選擇。例如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的情況下，需要對代價估計做相應的懲罰，降低他們被選為最優計劃的可能性。我們定義”好”的分布是每個節點處理的數據量在一個預設的範圍，低於或高於這個範圍都會被施加懲罰。估計這個數據量的依據包括：

1. 輸入數據記錄數（row count）
2. 重複度最高的數據（top values）
3. 直方圖（histogram）

總結

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

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

最後更新：2017-11-27 12:04:18

  上一篇：  ECS 備份數據到NAS（一）：使用Windows Server Backup工具

  下一篇：  必讀！教你一鍵遷移至阿裏雲