2017年中國大資料技術大會于12月7-9日在北京新雲南皇冠假日酒店隆重舉行, 大會就大資料時代社會各行業的智能化程序和行業實踐展開深入讨論。
概述
資料分布的問題在大資料處理領域由來已久。很不幸，如今流行的大資料處理系統仍然沒有很好地解決這個問題。在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在這方面是很在行的，它的解決辦法主要是：
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，避免一次重新分布。考慮以下查詢：
此時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)等不同的組合。
這些複雜度加大了最優計劃的搜尋空間。事實上，最優計劃是相對于關系代數數量的一個NPC問題。為了縮小搜尋空間，我們引入了啟發式的分支選擇算法。在編譯一個關系代數時，不僅需要滿足後繼操作的要求，還要考慮前序操作提供滿足的分布的可能性，後者被實作為稱作Pulled Up Property的子產品。
規避不合适的分布特性
資料傾斜（Skew）是指在分布中少量節點被配置設定了大部分資料，導緻整個算法退化為單機操作。低并發（Under Partition）是指分布指定了過少的節點，是的分布式資源不能被有效利用。我們希望能避免這兩種情況。
很顯然，更好的統計資訊會幫助我們規避這兩種情況。Skew和Under Partition的情況下，需要對代價估計做相應的懲罰，降低他們被選為最優計劃的可能性。我們定義”好”的分布是每個節點處理的資料量在一個預設的範圍，低于或高于這個範圍都會被施加懲罰。估計這個資料量的依據包括：
1、輸入資料記錄數（row count）
2、重複度最高的資料（top values）
3、直方圖（histogram）
總結
在這篇文章中，我們介紹了資料分布優化的問題和意義，并解釋了MaxCompute在資料分布優化上的實踐。這一優化效果已經展現在MaxCompute最新的釋出中。
從我們的測試來看，這個優化有相當顯著的效果。我們對TPC-H進行了适當分區後，整體性能提升在20%的量級。即使沒有對表資料分區，對使用者完全透明的運作時分區優化也有很好的效果。在我們線上運作的環境中，14%的查詢因為這個優化減少了至少一次資料重分布。