摘要: Hash Clustering經過容許用戶在建表時設置表的Shuffle和Sort屬性,進而MaxCompute根據數據已有的存儲特性,優化執行計劃,提升效率,節省資源消耗。 對於Hash Clustering總體帶來的性能收益,咱們經過標準的TPC-H測試集進行衡量。
在MaxCompute查詢中,Join是很常見的場景。例如如下Query,就是一個簡單的Inner Join把t1表和t2表經過id鏈接起來:sql
SELECT t1.a, t2.b FROM t1 JOIN t2 ON t1.id = t2.id;
數據庫
Join在MaxCompute內部主要有三種實現方法:安全
Broadcast Hash Join - 當Join存在一個很小的表時,咱們會採用這種方式,即把小表廣播傳遞到全部的Join Task Instance上面,而後直接和大表作Hash Join。服務器
Shuffle Hash Join - 若是Join表比較大,咱們就不能直接廣播了。這時候,我麼能夠把兩個表按照Join Key作Hash Shuffle,因爲相同的鍵值Hash結果也是同樣的,這就保證了相同的Key的記錄會收集到同一個Join Task Instance上面。而後,每一個Instance對數據量小的一路建Hash表,數據量大的順序讀取Join。併發
Sort Merge Join - 若是Join的表更大一些,#2的方法也用不了,由於內存已經不足以容納創建一個Hash Table。這時咱們的實現方法是,先按照Join Key作Hash Shuffle,而後再按照Join Key作排序,最後咱們對Join雙方作一個歸併,具體流程以下圖所示:app
實際上對於MaxCompute今天的數據量和規模,咱們絕大多數狀況下都是使用的Sort Merge Join,但這實際上是很是昂貴的操做。從上圖能夠看到,Shuffle的時候須要一次計算,而且中間結果須要落盤,後續Reducer讀取的時候,又須要讀取和排序的過程。對於M個Mapper和R個Reducer的場景,咱們將產生M x R次的IO讀取。對應的Fuxi物理執行計劃以下所示,須要兩個Mapper Stage,一個Join Stage,其中紅色部分爲Shuffle和Sort操做:
oop
與此同時,咱們觀察到,有些Join是可能反覆發生的,好比上面的Query改爲了:SELECT t1.c, t2.d FROM t1 JOIN t2 ON t1.id = t2.id;
性能
雖然,咱們選擇的列不同了,可是底下的Join是徹底同樣的,整個Shuffle和Sort的過程也是徹底同樣的。
又或者:SELECT t1.c, t3.d FROM t1 JOIN t3 ON t1.id = t3.id;
測試
這個時候是t1和t3來Join,但實際上對於t1而言,整個Shuffle和Sort過程仍是徹底同樣。優化
因而,咱們考慮,若是咱們初始表數據生成時,按照Hash Shuffle和Sort的方式存儲,那麼後續查詢中將避免對數據的再次Shuffle和Sort。這樣作的好處是,雖然建表時付出了一次性的代價,卻節省了未來可能產生的反覆的Shuffle和Join。這時Join的Fuxi物理執行計劃變成了以下所示,不只節省了Shuffle和Sort的操做,而且查詢從3個Stage變成了1個Stage完成:
因此,總結來講,Hash Clustering經過容許用戶在建表時設置表的Shuffle和Sort屬性,進而MaxCompute根據數據已有的存儲特性,優化執行計劃,提升效率,節省資源消耗。
set MaxCompute.sql.cfile2.enable.read.write.index.flag=true;
這個flag打開後,將對排序後的Hash Bucket自動創建Index,提升查詢效率。若是但願使用index功能,則在建表和後續查詢中都加上這個flag。若是但願在project中一直使用index,請與咱們聯繫,咱們能夠把一個project default setting打開。
Clustered Index對於在排序鍵上的查詢(等值或者範圍)有顯著幫助,可是即便沒有enable這個flag,仍然能夠享受到Hash Clustering其餘性能提高的好處。
用戶可使用如下語句建立Hash Clustering表。用戶須要指定Cluster Key(即Hash Key),以及Hash分片(咱們稱之爲Bucket)的數目。Sort是能夠選項,但在大多數狀況下,建議和Cluster Key一致,以便取得最佳的優化效果。
CREATE TABLE [IF NOT EXISTS] table_name
[(col_name data_type [comment col_comment], ...)] [comment table_comment] [PARTITIONED BY (col_name data_type [comment col_comment], ...)]
[CLUSTERED BY (col_name [, col_name, ...]) [SORTED BY (col_name [ASC | DESC] [, col_name [ASC | DESC] ...])] INTO number_of_buckets BUCKETS]
[AS select_statement]
舉個例子以下:
CREATE TABLE T1 (a string, b string, c bigint) CLUSTERED BY (c) SORTED by (c) INTO 1024 BUCKETS;
若是是分區表,則能夠用這樣的語句建立:CREATE TABLE T1 (a string, b string, c bigint) PARTITIONED BY (dt string) CLUSTERED BY (c) SORTED by (c) INTO 1024 BUCKETS;
CLUSTERED BY
CLUSTERED BY指定Hash Key,MaxCompute將對指定列進行Hash運算,按照Hash值分散到各個Bucket裏面。爲避免數據傾斜,避免熱點,取得較好的並行執行效果,CLUSTERED BY列適宜選擇取值範圍大,重複鍵值少的列。此外,爲了達到Join優化的目的,也應該考慮選取經常使用的Join/Aggregation Key,即相似於傳統數據庫中的主鍵。
SORTED BY
SORTED BY子句用於指定在Bucket內字段的排序方式,建議Sorted By和Clustered By一致,以取得較好的性能。此外,當SORTED BY子句指定以後,MaxCompute將自動生成索引,而且在查詢的時候利用索引來加快執行。
INTO number_of_buckets BUCKETS
INTO ... BUCKETS 指定了哈希桶的數目,這個數字必須提供,但用戶應該由數據量大小來決定。Bucket越多併發度越大,Job總體運行時間越短,但同時若是Bucket太多的話,可能致使小文件太多,另外併發度太高也會形成CPU時間的增長。目前推薦設置讓每一個Bucket數據大小在500MB - 1GB之間,若是是特別大的表,這個數值能夠再大點。
目前,MaxCompute只能在Bucket Number徹底一致的狀況下去掉Shuffle步驟,咱們下一個發佈,會支持Bucket的對齊,也就是說存在Bucket倍數關係的表,也能夠作Shuffle Remove。爲了未來能夠較好的利用這個功能,咱們建議Bucket Number選用2的N次方,如512,1024,2048,最大不超過4096,不然影響性能以及資源使用。
對於Join優化的場景,兩個表的Join要去掉Shuffle和Sort步驟,要求哈希桶數目一致。若是按照上述原則計算兩個表的哈希桶數不一致,怎麼辦呢?這時候建議統一使用數字大的Bucket Number,這樣能夠保證合理的併發度和執行效率。若是表的大小實在是相差太遠,那麼Bucket Number設置,能夠採用倍數關係,好比1024和256,這樣未來咱們進一步支持哈希桶的自動分裂和合並時,也能夠利用數據特性進行優化。
對於分區表,咱們支持經過ALTER TABLE語句,來增長或者去除Hash Clustering屬性:
ALTER TABLE table_name
[CLUSTERED BY (col_name [, col_name, ...]) [SORTED BY (col_name [ASC | DESC] [, col_name [ASC | DESC] ...])] INTO number_of_buckets BUCKETS
ALTER TABLE table_name NOT CLUSTERED;
關於ALTER TABLE,有幾點須要注意:
alter table改變彙集屬性,只對於分區表有效,非分區表一旦彙集屬性創建就沒法改變。
alter table只會影響分區表的新建分區(包括insert overwrite生成的),新分區將按新的彙集屬性存儲,老的數據分區保持不變。
因爲alter table隻影響新分區,因此該語句不能夠再指定PARTITION
ALTER TABLE語句適用於存量表,在增長了新的彙集屬性以後,新的分區將作hash cluster存儲。
在建立Hash Clustering Table以後,能夠經過:
DESC EXTENDED table_name;
來查看錶屬性,Clustering屬性將顯示在Extended Info裏面,以下圖所示:
對於分區表,除了可使用以上命令查看Table屬性以後,因而須要經過如下命令查看分區的屬性:
DESC EXTENDED table_name partition(pt_spec);
例如:
考慮如下查詢:
CREATE TABLE t1 (id bigint, a string, b string) CLUSTERED BY (id) SORTED BY (id) into 1000 BUCKETS;
...
SELECT t1.a, t1.b, t1.c FROM t1 WHERE t1.id=12345;
對於普通表,這個一般意味着全表掃描操做,若是表很是大的狀況下,資源消耗量是很是可觀的。可是,由於咱們已經對id作Hash Shuffle,而且對id作排序,咱們的查詢能夠大大簡化:
經過查詢值"12345"找到對應的Hash Bucket,這時候咱們只須要在1個Bucket裏面掃描,而不是所有1000個。咱們稱之爲「Bucket Pruning」。
由於Bucket內數據按ID排序存放,MaxCompute會自動建立Index,利用Index loopup直接定位到相關記錄。
能夠看出來,咱們不只大大減小了Mapper的個數,而且因爲利用了Index,Mapper能夠直接定位到數據所在Page,加載讀入的數據量也大大的減小了。
如下是安所有基於User ID查詢場景的一個例子。下面這個logview是普通的表的查詢操做,能夠看到,因爲數據量很大,一共起了1111個Mapper,讀取了427億條記錄,最後找符合條件記錄26條,總共耗時1分48秒:
一樣的數據,一樣的查詢,用Hash Clustering表來作,咱們能夠直接定位到單個Bucket,並利用Index只讀取包含查詢數據的Page,能夠看到這裏只用了4個Mapper,讀取了10000條記錄,總共耗時只須要6秒,若是用service mode這個時間還會更短:
例如,對於如下查詢:SELECT department, SUM(salary) FROM employee GROUP BY (department);
在一般狀況下,咱們會對department進行Shuffle和Sort,而後作Stream Aggregate,統計每個department group。可是若是表數據已經CLUSTERED BY (department) SORTED BY (department),那麼這個Shuffle和Sort的操做,也就相應節省掉了。
即使咱們不考慮以上所述的各類計算上的優化,單單是把表Shuffle並排序存儲,都會對於存儲空間節省上有很大幫助。由於MaxCompute底層使用列存儲,經過排序,鍵值相同或相近的記錄存放到一塊兒,對於壓縮,編碼都會更加友好,從而使得壓縮效率更高。在實際測試中,某些極端狀況下,排序存儲的表能夠比無序表的存儲空間節省50%。對於生命週期很長的表,使用Hash Clustering存儲,是一個很值得考慮的優化。
如下是一個簡單的實驗,使用100G TPC-H lineitem表,包含了int,double,string等多種數據類型,在數據和壓縮方式等徹底同樣的狀況下,hash clustering的表空間節省了~10%。
對於Hash Clustering總體帶來的性能收益,咱們經過標準的TPC-H測試集進行衡量。測試使用1T數據,統一使用500 Buckets,除了nation和region兩個極小的表之外,其他全部表均按照第一個列做爲Cluster和Sort Key。
總體測試結果代表,在使用了Hash Clustering以後,總CPU時間減小17.3%,總的Job運行時間減小12.8%。
具體各個Query CPU時間對好比下:
Job運行時間對好比下:
須要注意到是TPC-H裏並非全部的Query均可以利用到Clustering屬性,特別是兩個耗時最長的Query沒有辦法利用上,因此從整體上的效率提高並非很是驚人。但若是單看能夠利用上Clustering屬性的Query,收益仍是很是明顯的,好比Q4快了68%,Q12快了62%,Q10快了47%,等等。
如下是TPC-H Q4在普通表的Fuxi執行計劃:
而下面則是使用Hash Clustering以後的執行計劃,能夠看到,這個DAG被大大的簡化,這也是性能獲得大幅提高的關鍵緣由:
目前Hash Clustering的第一階段開發工做完成,但還存在如下限制和不足:
雲服務器99元拼團購!拉新還可贏現金紅包!300萬等你瓜分!
立刻一鍵開團贏紅包: http://click.aliyun.com/m/100...
本文爲雲棲社區原創內容,未經容許不得轉載。