MaxCompute Hash Clustering介绍

摘要： Hash Clustering经过容许用户在建表时设置表的Shuffle和Sort属性，进而MaxCompute根据数据已有的存储特性，优化执行计划，提升效率，节省资源消耗。 对于Hash Clustering总体带来的性能收益，咱们经过标准的TPC-H测试集进行衡量。

背景

在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功能已经全面上线，缺省条件下即打开支持。可是，若是须要使用clustered index，须要加上一下flag：

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 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和Index优化

考虑如下查询：

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这个时间还会更短：

• 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的第一阶段开发工做完成，但还存在如下限制和不足：

1. 不支持insert into，只能经过insert overwrite来添加数据。
2. 不支持小文件合并。因为区域汇集在切分的时候已经尽可能保证数据在各个bucket里面均匀分布，因此就不存在小文件的问题了。而直接的文件merge将破坏区域汇集属性。可是，咱们仍然支持经过merge和archive命令来改变表文件存储格式，以及RAID文件转换的功能。
3. 不支持tunnel直接upload到range cluster表，由于tunnel上传数据是无序的。

双十一广告：阿里云双十一1折拼团活动:已满6人,都是最低折扣了
【满6人】1核2G云服务器99.5元一年298.5元三年 2核4G云服务器545元一年 1227元三年
【满6人】1核1G MySQL数据库 119.5元一年
【满6人】3000条国内短信包 60元每6月
参团地址：http://click.aliyun.com/m/1000020293/

做者： 晋恒 原文连接 本文为云栖社区原创内容，未经容许不得转载。