?

背景

在MaxCompute查詢中，Join是很常見的場景。例如以下Query，就是一個簡單的Inner Join把t1表和t2表通過id連接起來：

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雙方做一個歸并，具體流程如下圖所示：

實際上對于MaxCompute今天的數據量和規模，我們絕大多數情況下都是使用的Sort Merge Join，但這其實是非常昂貴的操作。從上圖可以看到，Shuffle的時候需要一次計算，并且中間結果需要落盤，后續Reducer讀取的時候，又需要讀取和排序的過程。對于M個Mapper和R個Reducer的場景，我們將產生M x R次的IO讀取。對應的Fuxi物理執行計劃如下所示，需要兩個Mapper Stage，一個Join Stage，其中紅色部分為Shuffle和Sort操作：

與此同時，我們觀察到，有些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根據數據已有的存儲特性，優化執行計劃，提高效率，節省資源消耗。

功能描述

目前Hash Clustering功能已經上線，缺省條件下即打開支持。

• 創建Hash Clustering Table

用戶可以使用以下語句創建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);

例如：

Hash Clustering的其他優點

• Bucket Pruning優化

考慮以下查詢：

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”。

以下是安全部基于User ID查詢場景的一個例子。下面這個logview是普通的表的查詢操作，可以看到，由于數據量很大，一共起了1111個Mapper，讀取了427億條記錄，最后找符合條件記錄26條，總共耗時1分48秒：

同樣的數據，同樣的查詢，用Hash Clustering表來做，我們可以直接定位到單個Bucket，并利用Index只讀取包含查詢數據的Page，可以看到這里只用了4個Mapper，讀取了10000條記錄，總共耗時只需要6秒，如果用service mode這個時間還會更短：

• Aggregation優化

例如，對于以下查詢：
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的第一階段開發工作完成，但還存在以下限制和不足：

不支持insert into，只能通過insert overwrite來添加數據。

不支持tunnel直接upload到range cluster表，因為tunnel上傳數據是無序的。

?

原文鏈接
本文為云棲社區原創內容，未經允許不得轉載。

?

總結

以上是生活随笔為你收集整理的MaxCompute Hash Clustering介绍的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。