什么RangeClustering,有哪些功能,如何使用_云原生大数据计算服务 MaxCompute(MaxCompute)-阿里云帮助中心

Range Clustering作为一种新的数据切分方式，提供了一个全局有序的数据分布，一是可以避免Hash Clustering可能造成的数据倾斜问题；二是在数据有序分布的前提下，创建两级索引（Index），支持对Clustering Key的区域查询以及多键的组合查询等场景。本文为您介绍如何在MaxCompute中使用Range Clustering。

背景信息

哈希聚簇（Hash Clustering）表有以下优点：

对于等值的列条件查询，可以利用Hash算法直接定位到对应的哈希桶（Bucket Pruning），如果桶内数据排序存储，还可以进一步利用索引定位，从而减少数据扫描量，提高查询效率。
如果对不同表某一列上做Join，其中一张表因为已经对其中一列做Hash分布，可以省掉Shuffle的步骤（称之为Shuffle Remove），进而节省计算资源。

关于Hash Clustering功能的详细介绍请参见 Hash Clustering 。

但是Hash Clustering也有一些局限性：

使用Hash算法分桶，有可能产生Data Skew的问题。和Join Skew一样，这是Hash算法本身固有的局限性，输入数据存在某些特定的数据分布时，可能造成倾斜，进而导致各个哈希桶之间数据量差异较大。因为Hash Clustering之后，并发处理单位往往是一个桶，如果哈希桶数据量不一致，往往容易造成长尾现象。
Bucket Pruning只支持等值查询。因为使用哈希分桶方法，对于区间查询比如使用某列值大于0这样的条件，无法在哈希桶级别定位，只能把查询下发到所有桶内进行。
对于多个Cluster Key的组合查询，只有所有Cluster Key都出现并且都为等值条件，才能达到优化效果。
比如，对于下表只有查询条件包含 C1=x AND C2=y 才可以优化。单独的 C1=x 或者 C2=y 查询条件皆无法利用Hash Clustering特性加速。这是因为对于组合键的情况做了Combine Hash，要求查询的时候一定要成对出现，否则无法定位哈希桶，也无法做Bucket Pruning。
```
CREATE TABLE T2 (C1 int, C2 int, C3 string)
   CLUSTERED BY (C1, C2)
      SORTED by (C1, C2)
       INTO 1024 BUCKETS;
```

针对这些局限性，MaxCompute推出了新的Clustering方法，即Range Clustering。

功能简介

Range Clustering在对Cluster Key全排序的基础上，将其值域空间切分成若干个不连续值域（Disjointed Range），每个Range作为一个Bucket，并且满足如下条件：

相同数值存在于同一个Bucket里面。
每个Bucket包含数值个数尽可能接近。

举一个简单的例子：假设有表T，定义如下。

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

同时，C1列取值为 { 1, 8, -3, 2, 4, 1, 1, 3, 8, 20, -8, 9 } 。

在Range Clustering之后，得到如下3个Bucket：

Bucket 0 : { -8, -3, 1, 1, 1 }
Bucket 1 : { 2, 3, 4 }
Bucket 2 : { 8, 8, 9, 20 }

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]

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

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

--更改表为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;

DESC EXTENDED <table_name>;

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

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;

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

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

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

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

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;

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

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

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

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

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;

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

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

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

```
create table t1(a bigint, b bigint, c bigint, d bigint)
       range clustered by(a,b,c)
       sorted by(a,b,c) into 3 buckets;
create table t2(a bigint, b bigint, c bigint, d bigint)
       range clustered by(a,b,c)
       sorted by(a,b,c) into 3 buckets;
```
```
create table t1(a bigint, b bigint, c bigint, d bigint)
       range clustered by(a,b,c)
       sorted by(a,b,c) into 5 buckets;
create table t2(a bigint, b bigint, c bigint, d bigint)
       range clustered by(a,b,c)
       sorted by(a,b,c) into 3 buckets;
```
```
SELECT * FROM t1 JOIN t2 ON t1.a=t2.a AND t1.b=t2.b AND t1.c=t2.c;
```
```
SELECT * FROM t1 JOIN t2 ON t1.a=t2.a AND t1.b=t2.b;
```

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;
CREATE TABLE PART ( P_PARTKEY BIGINT NOT NULL,
                    P_NAME VARCHAR(55) NOT NULL,
                    P_MFGR CHAR(25) NOT NULL,
                    P_BRAND CHAR(10) NOT NULL,
                    P_TYPE  VARCHAR(25) NOT NULL,
                    P_SIZE   BIGINT NOT NULL,
                    P_CONTAINER   CHAR(10) NOT NULL,
                    P_RETAILPRICE DECIMAL(15,2) NOT NULL,
                    P_COMMENT VARCHAR(23) NOT NULL)
             RANGE CLUSTERED BY(P_PARTKEY)
             SORTED BY(P_PARTKEY)
             INTO 64 BUCKETS;

select s_acctbal,
       s_name,
       n_name,
       p_partkey,
       p_mfgr,
       s_address,
       s_phone,
       s_comment
  from part,
       supplier,
       partsupp,
       nation,
       region
 where p_partkey = ps_partkey
   and s_suppkey = ps_suppkey
   and p_size = 15
   and p_type like '%BRASS'
   and s_nationkey = n_nationkey
   and n_regionkey = r_regionkey
   and r_name = 'EUROPE'
   and ps_supplycost = (select min(ps_supplycost)
                          from partsupp, supplier, nation, region
                         where p_partkey = ps_partkey
                           and s_suppkey = ps_suppkey
                           and s_nationkey = n_nationkey
                           and n_regionkey = r_regionkey
                           and r_name = 'EUROPE')
  order by s_acctbal desc, n_name, s_name, p_partkey limit 100;

- ```
 set odps.optimizer.enable.range.partial.repartitioning=false;
```
- ```
set odps.optimizer.enable.range.partial.repartitioning=true;
```

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

Range Clustering

背景信息

功能简介

使用说明

创建Range Clustering表

更改表的Hash Clustering属性

表属性显式验证

使用场景

过滤查询优化

多键组合查询

Group By优化

Aggregate 优化

Range表的Join优化

全局排序加速

局限性和注意事项