Range Clustering在對(duì)Cluster Key全排序的基礎(chǔ)上，將其值域空間切分成若干個(gè)不連續(xù)值域（Disjointed Range），每個(gè)Range作為一個(gè)Bucket，并且滿足如下條件：

• 相同數(shù)值存在于同一個(gè)Bucket里面。

• 每個(gè)Bucket包含數(shù)值個(gè)數(shù)盡可能接近。

舉一個(gè)簡(jiǎn)單的例子：假設(shè)有表T，定義如下。

CREATE TABLE T (C1 int) 
    RANGE CLUSTERED BY (C1) 
    SORTED BY (c1) 
    INTO 3 BUCKETS;

同時(shí)，C1列取值為{ 1, 8, -3, 2, 4, 1, 1, 3, 8, 20, -8, 9 }。

在Range Clustering之后，得到如下3個(gè)Bucket：

• Bucket 0 : { -8, -3, 1, 1, 1 }

• Bucket 1 : { 2, 3, 4 }

• Bucket 2 : { 8, 8, 9, 20 }

Range的切分過(guò)程是MaxCompute自動(dòng)完成的，不需要手動(dòng)指定每一個(gè)Range。因?yàn)樵诖髷?shù)據(jù)場(chǎng)景下手動(dòng)指定Range不高效，往往也不現(xiàn)實(shí)。MaxCompute內(nèi)部會(huì)自動(dòng)對(duì)數(shù)據(jù)進(jìn)行排序、采樣，對(duì)各個(gè)區(qū)間的數(shù)據(jù)密度建立直方圖，最后通過(guò)合并計(jì)算各個(gè)區(qū)間直方圖，來(lái)實(shí)現(xiàn)最終理想的Range切分。

此外RANGE CLUSTERED BY可以和SORTED BY結(jié)合，從而保證了數(shù)據(jù)全局有序，在這個(gè)基礎(chǔ)上，MaxCompute將自動(dòng)創(chuàng)建兩級(jí)索引：Global Index和File Index，用于快速定位和查找鍵值，如下圖所示：

和Hash Clustering相比，Range Clustering優(yōu)勢(shì)如下：

• 支持區(qū)間查詢。

  假設(shè)有查詢條件c < 3，則可以根據(jù)Global Index快速排除掉Bucket 2和Bucket 3，從而只在Bucket 0和Bucket 1中查找。Hash Clustering只能對(duì)等值查詢做Bucket Pruning。

• 支持多個(gè)Cluster列的組合查找。

  假設(shè)使用RANGE CLUSTERED BY (c1, c2, c3) SORTED BY (c1, c2, c3)，那么在Range切分和數(shù)據(jù)存儲(chǔ)都是按照c1、c2、c3排序，這就可以允許做更復(fù)雜的組合查詢，比如c1 = 100 AND c2 > 0或者c1 = 100 AND c2 = 50 AND c3 < 5，這個(gè)也是Hash Clustering無(wú)法支持的。

  重要

  在多鍵組合查詢時(shí)，條件需要按順序列出排序鍵，并且只有最后一個(gè)Key允許使用區(qū)間條件。

• Range Clustering提供了一個(gè)高效的全局ORDER BY實(shí)現(xiàn)。

  在沒(méi)有Range Clustering之前為了保證全局有序，MaxCompute只能通過(guò)一個(gè)Instance進(jìn)行ORDER BY排序，效率很低。使用Range Clustering之后做Range切分之后，各個(gè)Range可以并發(fā)分別排序，最后在組合到一起，效率極大提高。

您可以使用以下語(yǔ)句創(chuàng)建Range Clustering表。您需要指定Range Cluster Key、Bucket數(shù)目（可選）。Sort是可選項(xiàng)，但在大多數(shù)情況下，建議和Range Cluster Key一致，以便取得最佳的優(yōu)化效果。

• 命令語(yǔ)法

  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>], ...)]
               [RANGE CLUSTERED BY (<col_name> [, <col_name>, ...])
               [SORTED BY (<col_name> [ASC | DESC]
               [, <col_name> [ASC | DESC] ...])]
               [INTO <number_of_buckets> BUCKETS]]
               [AS select_statement]

• 使用示例。

  • 非分區(qū)表。

    CREATE TABLE T1 (a string, b string, c int)
                 RANGE CLUSTERED BY (c)
                 SORTED by (c)
                 INTO 1024 BUCKETS;

  • 分區(qū)表。

    CREATE TABLE T1 (a string, b string, c int)
                 PARTITIONED BY (dt int)
                 RANGE CLUSTERED BY (c)
                 SORTED by (c)
                 INTO 1024 BUCKETS;

• 屬性說(shuō)明

  • RANGE CLUSTERED BY

    指定Range Cluster Key，MaxCompute將對(duì)指定一列或者多列進(jìn)行排序和采樣，并根據(jù)指定Bucket數(shù)目，切分到盡可能理想的若干個(gè)Range里面。為避免數(shù)據(jù)傾斜、避免熱點(diǎn)，取得較好的并行執(zhí)行效果，Clustered By列適宜選擇取值范圍大、重復(fù)鍵值少的列。此外為了達(dá)到優(yōu)化目的，也應(yīng)該考慮選取常用的Aggregation Key或者Filter Key。

  • SORTED BY

    指定在Bucket內(nèi)字段的排序方式，建議Sorted By和Clustered By一致，以取得較好的性能。此外當(dāng)Sorted By子句指定之后，MaxCompute將自動(dòng)生成索引（Global Index和File Index），并且在查詢的時(shí)候利用索引來(lái)加速執(zhí)行。

  • INTO number_of_buckets BUCKETS

    和Hash Clustering不一樣，對(duì)于Range ClusteringINTO ... BUCKETS是可選項(xiàng)。如果不指定Bucket數(shù)目，MaxCompute將根據(jù)數(shù)據(jù)大小自動(dòng)決定分片數(shù)目。在大多數(shù)情況下，建議指定Bucket數(shù)目，以便根據(jù)實(shí)際情況來(lái)做優(yōu)化選擇。

    對(duì)于Bucket數(shù)目的選擇，和Hash Clustering類似，建議保持在每個(gè)Bucket 512MB ~ 1GB這樣的數(shù)據(jù)量，對(duì)于特別大的表，Bucket 數(shù)目可以大些，但建議不要超過(guò)4000。

對(duì)于分區(qū)表支持通過(guò)ALTER TABLE語(yǔ)句，來(lái)增加或者去除Range Clustering屬性。

• 命令語(yǔ)句

  --更改表為Range Clustering表
  ALTER TABLE <table_name> [RANGE CLUSTERED BY (<col_name> [, <col_name>, ...])
                           [SORTED BY (<col_name> [ASC | DESC] [, <col_name> [ASC | DESC] ...])]
                           [INTO <number_of_buckets> BUCKETS];
  --更改Range Clustering表為非Range Clustering表
  ALTER TABLE <table_name> NOT CLUSTERED;

• 注意事項(xiàng)

  • ALTER TABLE語(yǔ)句改變聚集屬性，只對(duì)于分區(qū)表有效，非分區(qū)表一旦聚集屬性建立就無(wú)法改變。

  • ALTER TABLE語(yǔ)句只會(huì)影響分區(qū)表的新建分區(qū)（包括insert overwrite生成的），新分區(qū)將按新的聚集屬性存儲(chǔ)，老的數(shù)據(jù)分區(qū)保持不變。

  • 由于ALTER TABLE語(yǔ)句只影響新分區(qū)，所以該語(yǔ)句不可以再指定PARTITION。

ALTER TABLE語(yǔ)句適用于存量表，在增加了新的聚簇屬性之后，新的分區(qū)將做Range Clustering存儲(chǔ)。

在創(chuàng)建Range Clustering Table之后，可以通過(guò)如下命令來(lái)查看表屬性，Range Clustering屬性將顯示在Extended Info里面。

DESC EXTENDED <table_name>;

返回結(jié)果示例如下圖所示。對(duì)于分區(qū)表，除了使用以上命令查看表屬性之后，還可以通過(guò)以下命令查看分區(qū)的屬性。

DESC EXTENDED <table_name> partition(<pt_spec>);

返回結(jié)果示例如下圖所示。

例如對(duì)于一個(gè)簡(jiǎn)單的查詢條件id < 3，先在優(yōu)化器里面將查詢條件抽取出來(lái)，并轉(zhuǎn)化成值域空間(-∞, 3)。這個(gè)時(shí)候就可以利用Global Index做Bucket Pruning，把不在這個(gè)區(qū)間的Bucket 2和Bucket 3都去掉。最后再利用每個(gè)Bucket文件自帶的Index，快速在文件內(nèi)部定位，這個(gè)過(guò)程稱之為謂詞下推（Predicate Pushdown），示意圖如下：TPC-H Q6在100GB數(shù)據(jù)集上使用Range Clustering后進(jìn)行查詢，查詢語(yǔ)句如下。Q6是一個(gè)區(qū)間過(guò)濾的基礎(chǔ)上再做的聚合操作，使用Range Clustering可以利用兩級(jí)Index快速定位數(shù)據(jù)，無(wú)論是執(zhí)行時(shí)間、CPU使用率和內(nèi)存使用率的性能都有數(shù)倍的提升。

select sum(l_extendedprice * l_discount) as revenue
  from tpch_lineitem l
 where l_shipdate >= '1994-01-01'
   and l_shipdate < '1995-01-01'
   and l_discount >= 0.05
   and l_discount <= 0.07
   and l_quantity < 24;

為了比較好的理解多鍵組合查詢場(chǎng)景，使用如下命令將mf_tab 表改造成Range Clustering表。

ALTER TABLE mf_project.mf_tab 
            RANGE CLUSTERED BY (project_name, name)
            SORTED BY (project_name, name)
            INTO 1024 BUCKETS;

這樣做的好處是，既可以在Project級(jí)別上做一些聚合查詢，示例命令如下：

SELECT COUNT(*)
  from mf_project.mf_tab
 WHERE project_name="xxxdw"
   AND ds="20180115"
   AND type="TABLE";

又可以用組合鍵來(lái)精確定位某張表，示例命令如下：

SELECT count(*)
  from mf_project.mf_tab
 WHERE project_name="xxxdw"
   AND name="adm_ctu_cle_kba_midun_trade_dd"
   AND type="TABLE";

甚至可以用于區(qū)域查詢，比如統(tǒng)計(jì)以adm開頭的表：

SELECT count(*)
  from mf_project.mf_tab
 WHERE project_name="xxxdw"
   AND name>="adm"
   AND name < "adn"
   AND type="TABLE";

以上查詢都可以充分利用Range Clustering全局排序的特性，下推查詢謂詞，減少表掃描的IO量以及過(guò)濾計(jì)算的CPU和內(nèi)存消耗。

假設(shè)表T的Range Cluster Key為k1, k2 ，則有：

• 查詢條件為k1 < 5 ：可以Index加速。

• k1 = 10 AND k2 = 20：可以Index加速。

• k1 = 10 AND k2 < 0：可以Index加速。

• k2 < 0：不可以Index加速，k1缺失。

• k1 < 0 AND k2 > 0：可以Index加速部分，Index只過(guò)濾k1<0的條件，k2>0條件需要掃描。

例如，對(duì)于如下對(duì)表T數(shù)據(jù)的Group By操作可以直接在一個(gè)Mapper Stage里面完成。

CREATE TABLE T (department int, team string, employee string)
       RANGE CLUSTERED BY (department, team)
       SORTED BY (c1, c2)
       INTO 1024 BUCKETS;

SELECT COUNT(*) from T GROUP BY department, team;

表foo結(jié)構(gòu)如下。

create table foo(a bigint, b bigint, c bigint)
       range clustered by (a,b)
       sorted by(a,b) into 3 buckets;

假設(shè)表foo的Range為：

Bucket 0: [1,1 : 3,3]
Bucket 1: [5,5 : 7,7]
Bucket 2: [8,8 : 9,9]

以上Bucket Range可以理解為：Bucket N: [lower bound values : upper bound values]。如果按照a來(lái)聚合的話，每個(gè)Bucket的范圍就變成：

Bucket 0: [1 : 3]
Bucket 1: [5 : 7]
Bucket 2: [8 : 9]

可以直接產(chǎn)生類似按照a、b聚合的執(zhí)行計(jì)劃（Plan），啟動(dòng)3個(gè)Instance對(duì)每個(gè)Bucket聚合，然后輸出結(jié)果后結(jié)束。

但是對(duì)于a的值橫跨多個(gè)Bucket的情況下，就會(huì)得到錯(cuò)誤的結(jié)果，示例如下。

Bucket 0: [1,1 : 3,3]
Bucket 1: [3,5 : 7,7]
Bucket 2: [7,8 : 9,9]

列a有兩個(gè)值3和7，分別橫跨兩個(gè)Bucket，只有把擁有相同的a值的tuple放到同一個(gè)Instance里面去進(jìn)行聚合才能得到正確的結(jié)果。也就是像下面按照紅色虛線重新切分Bucket，指定每個(gè)Instance讀取的數(shù)據(jù)的范圍才行：但是該從哪個(gè)地方切分， 這時(shí)候就需要直方圖的幫助。對(duì)于Range Clustering表，在Clustering Key和Sort Key相同的情況下，當(dāng)往Range Clustering表插入數(shù)據(jù)的時(shí)候，每個(gè)Bucket對(duì)應(yīng)的Worker會(huì)每10000行采樣一個(gè)tuple，取其Clustering Keys的值，并保存到直方圖中，最后每個(gè)Bucket的直方圖會(huì)存到Clustermeta文件中。 這種直方圖被稱為等高直方圖（equi-depth histogram）。

有了每個(gè)Bucket的直方圖就可以為每個(gè)Worker重新切分Bucket。切分原則如下：

• 擁有相同Grouping Key的tuple必須被劃分到同一個(gè)Bucket中。

• 每個(gè)Bucket中的數(shù)據(jù)盡量均勻的切分。

有了每個(gè)新的Bucket的起始值，每個(gè)Worker就可以讀取正確的數(shù)據(jù)范圍，并返回正確的結(jié)果。

使用TPC-H數(shù)據(jù)集中1TB的partsupp表來(lái)測(cè)試性能提升如何。首先使用如下命令將partsupp表改造為Range Clustering表：

CREATE TABLE partsupp ( PS_PARTKEY BIGINT NOT NULL,
                        PS_SUPPKEY BIGINT NOT NULL,
                        PS_AVAILQTY BIGINT NOT NULL,
                        PS_SUPPLYCOST  DECIMAL(15,2)  NOT NULL,
                        PS_COMMENT     VARCHAR(199) NOT NULL)
             RANGE CLUSTERED BY(PS_PARTKEY, PS_SUPPKEY)
             SORTED BY(PS_PARTKEY, PS_SUPPKEY) INTO 128 BUCKETS;

使用下面的Query進(jìn)行測(cè)試：

SELECT ps_partkey, count(*) c FROM partsupp GROUP BY ps_partkey;

• 使用如下命令關(guān)閉優(yōu)化：

   set odps.optimizer.enable.range.partial.repartitioning=false;

  結(jié)果示例：

• 使用如下命令打開優(yōu)化：

  set odps.optimizer.enable.range.partial.repartitioning=true;

  結(jié)果示例：

由上圖可以看出，針對(duì)此查詢優(yōu)化后速度提升了57%、CPU的使用率降低了52%、內(nèi)存的使用降低了71%。當(dāng)然針對(duì)不同數(shù)據(jù)量、不同的查詢，可能得到性能提升的程度也不同。

Range Clustering的全排序加速缺省關(guān)閉，如果需要打開，請(qǐng)使用以下Flag。

set odps.optimizer.distribute.ordering.enable=true;

Range Clustering和Hash Clustering比較，有優(yōu)勢(shì)的同時(shí)在某些方面也有一些局限：

• Range Clustering在數(shù)據(jù)生成的代價(jià)比Hash Clustering要高。Hash Clustering只是做一個(gè)簡(jiǎn)單的Hash操作和排序，但是對(duì)于Range而言，需要做數(shù)據(jù)采樣，排序，以及Histogram的合并等等，總體的代價(jià)（運(yùn)行時(shí)間、CPU使用率、Memory Cost）比Hash Clustering高。所以在Hash Clustering可以解決的問(wèn)題，就不需要用Range Clustering。

• 不支持DYNAMIC PARTITION，以及INSERT INTO場(chǎng)景。

• 對(duì)Join的支持方面只支持Inner Join、Left/Right Outer Join、Semi Join，不支持Anti Join和Full Outer Join。

• Range Clustering表的Range Cluster Key必須和Sort Key相同，比如對(duì)于表foo中range clustered by (a,b) sorted by (a,b)，本文中的優(yōu)化可以生效；但是對(duì)于表bar中range clustered by(a,b) sorted by (b,a)，本文中的優(yōu)化就不會(huì)生效。

• Join Key和Group Key必須為Range Clustering key的全部或者前綴。比如range clustered by(a,b,c) sorted by(a,b,c)，對(duì)于Join Key和Group Key為a、a,b 或a,b,c的查詢才能應(yīng)用本文中的優(yōu)化；對(duì)于Join Key或Group Key為b或a,c的查詢就無(wú)法應(yīng)用本文中的優(yōu)化。

• 對(duì)于分區(qū)表，如果讀取的Range Clustering表涉及兩個(gè)或兩個(gè)以上的分區(qū)無(wú)法應(yīng)用本文中的優(yōu)化。目前本文中的優(yōu)化只針對(duì)單個(gè)分區(qū)的分區(qū)表和非分區(qū)表有效。