MaxCompute複雜數據分佈的查詢優化實踐

摘要： 2017年中國大數據技術大會於12月7-9日在北京新雲南皇冠假日酒店隆重舉行, 大會就大數據時代社會各行業的智能化進程和行業實踐展開深刻討論。 在12月8日的「大數據分析與生態系統」分論壇上，來自阿里巴巴計算平臺事業部的高級技術專家少傑，以「MaxCompute 複雜數據分佈的查詢優化實踐」爲題，爲現場來賓分享了阿里雲MaxCompute最新技術與實踐的洞察與經驗。

2017年中國大數據技術大會於12月7-9日在北京新雲南皇冠假日酒店隆重舉行, 大會就大數據時代社會各行業的智能化進程和行業實踐展開深刻討論。

在12月8日的「大數據分析與生態系統」分論壇上，來自阿里巴巴計算平臺事業部的高級技術專家少傑，以「MaxCompute 複雜數據分佈的查詢優化實踐」爲題，爲現場來賓分享了阿里雲MaxCompute最新技術與實踐的洞察與經驗。

概述
數據分佈的問題在大數據處理領域由來已久。很不幸，現在流行的大數據處理系統仍然沒有很好地解決這個問題。在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服從Hashid, b服從Hashid，對於Sorted Merge Join，它能夠選擇要求Hashid，或Hashid，甚至任意Hash(id)。
這些複雜度加大了最優計劃的搜索空間。事實上，最優計劃是相對於關係代數數量的一個NPC問題。爲了縮小搜索空間，咱們引入了啓發式的分支選擇算法。在編譯一個關係代數時，不只須要知足後繼操做的要求，還要考慮前序操做提供知足的分佈的可能性，後者被實現爲稱做Pulled Up Property的模塊。

Pulled Up Property猜想並篩選可能的前序delivered property，用於在編譯時減小搜索寬度。考慮上圖的查詢，在Join編譯時，由於Sink的需求下推，它被要求提供一個Hashc1。Pulled Up Property則從前序操做猜想可能會提供Hashc1和Hashc1，綜合考慮，Join可能會直接要求Hashc1，從而減小了Hashc1和Hashc1這兩個分支。

規避不合適的分佈特性
數據傾斜（Skew）是指在分佈中少許節點被分配了大部分數據，致使整個算法退化爲單機操做。低併發（Under Partition）是指分佈指定了過少的節點，是的分佈式資源不能被有效利用。咱們但願能避免這兩種狀況。
很顯然，更好的統計信息會幫助咱們規避這兩種狀況。Skew和Under Partition的狀況下，須要對代價估計作相應的懲罰，下降他們被選爲最優計劃的可能性。咱們定義」好」的分佈是每一個節點處理的數據量在一個預設的範圍，低於或高於這個範圍都會被施加懲罰。估計這個數據量的依據包括：
一、輸入數據記錄數（row count）
二、重複度最高的數據（top values）
三、直方圖（histogram）

總結在這篇文章中，咱們介紹了數據分佈優化的問題和意義，並解釋了MaxCompute在數據分佈優化上的實踐。這一優化效果已經體如今MaxCompute最新的發佈中。從咱們的測試來看，這個優化有至關顯著的效果。咱們對TPC-H進行了適當分區後，總體性能提高在20%的量級。即便沒有對錶數據分區，對用戶徹底透明的運行時分區優化也有很好的效果。在咱們線上運行的環境中，14%的查詢由於這個優化減小了至少一次數據重分佈。