云原生大数据计算服务 MaxCompute：Hash Clustering

• 命令语法

  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

    指定Hash Key，MaxCompute将对指定列进行Hash运算，按照Hash值分散到各个Bucket里面。为避免数据倾斜，避免热点，取得较好的并行执行效果，CLUSTERED BY列适宜选择取值范围大、重复键值少的列。此外为了达到Join优化的目的，也应该考虑选取常用的Join或Aggregation Key，即类似于传统数据库中的主键。

  • SORTED BY

    指定在Bucket内字段的排序方式，建议Sorted By和Clustered By一致，以取得较好的性能。此外当SORTED BY子句指定之后，MaxCompute将自动生成索引，并且在查询的时候利用索引来加速执行。

  • INTO number_of_buckets BUCKETS

    指定哈希桶的数目，这个数字必须提供，由数据量大小来决定。Bucket越多并发度越大，Job整体运行时间越短，但同时如果Bucket太多的话，可能导致小文件太多，另外并发度过高也会造成CPU时间的增加。目前推荐设置让每个Bucket数据大小在500MB ~ 1GB之间，如果是特别大的表，这个数值可以适当增加。对于Join优化的场景，两个表的Join要去掉Shuffle和Sort步骤，要求哈希桶数目成倍数关系，比如256和512。目前建议桶的数目统一使用2的N次幂，比如512、1024、2048、4096，这样系统可以自动进行哈希桶的分裂和合并时，也可以去除Shuffle和Sort的步骤。

• 命令语句

  --更改表为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];
  --更改Hash Clustering表为非Hash Clustering表
  ALTER TABLE <table_name> NOT CLUSTERED;

• 注意事项
  • ALTER TABLE语句改变聚集属性，只对于分区表有效，非分区表一旦聚集属性建立就无法改变。
  • ALTER TABLE语句只会影响分区表的新建分区（包括insert overwrite生成的），新分区将按新的聚集属性存储，老的数据分区保持不变。
  • 由于ALTER TABLE语句只影响新分区，所以该语句不可以再指定PARTITION。

DESC EXTENDED <table_name>;

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 FROM t1 WHERE t1.id=12345;

1. 通过查询值12345找到对应的Hash Bucket，这时候我们只需要在1个Bucket里面扫描，而不是全部1000个Bucket里面扫描。称之为Bucket Pruning。
2. 因为Bucket内数据按id排序存放，MaxCompute会自动创建Index，利用Index lookup直接定位到相关记录。

SELECT department, SUM(salary) FROM employee GROUP BY (department);

• 未使用Hash Clustering。
• 使用Hash Clustering。