MaxCompute:Hash clustering

• Syntax

  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>]

• Examples
  • Non-partitioned table

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

  • Partitioned table

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

• Parameters
  • CLUSTERED BY

    Specifies the hash keys based on which MaxCompute performs hash operations on specified columns. After hash operations are performed, MaxCompute distributes data to buckets based on hash values. To prevent data skew issues and hot spots and improve the concurrent execution efficiency, we recommend that you specify columns that have large value ranges and a small number of duplicate key values in CLUSTERED BY. To optimize join operations, we recommend that you select commonly used join keys or aggregate keys. The join keys and aggregate keys are similar to the primary keys in conventional databases.

  • SORTED BY

    Specifies how to sort fields in a bucket. To improve query performance, we recommend that you keep the configuration of SORTED BY consistent with that of the CLUSTERED BY parameter. If you specify the SORTED BY parameter, MaxCompute automatically generates indexes and accelerates queries based on the indexes.

  • INTO number_of_buckets BUCKETS

    Specifies the number of buckets. This parameter is required and is determined based on the amount of data. More buckets indicate higher concurrency and higher execution efficiency of jobs. However, if an excessive number of buckets exist, excessive small files may be generated. Excessively high concurrency also increases CPU time. We recommend that you keep the data size of each bucket between 500 MB to 1 GB. If a table is extremely large, you can appropriately increase the data size of each bucket. In join optimization scenarios, the shuffle and sort steps must be removed from the join operation on two tables. Therefore, the number of buckets in a table must be a multiple of the number of buckets in the other table. For example, one table has 256 buckets and the other table has 512 buckets. We recommend that you set the number of buckets to 2 raised to the power of N, such as 512, 1,024, 2,048, or 4,096. This way, MaxCompute can automatically split and merge buckets, and the shuffle and sort steps can be removed.

• Syntax

  -- Change a table to a hash-clustered table.
  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];
  -- Change a hash-clustered table to a non-hash-clustered table.
  ALTER TABLE <table_name> NOT CLUSTERED;

• Usage notes
  • The ALTER TABLE statement can only modify the clustering properties of a partitioned table. The clustering properties cannot be modified after they are added to a non-partitioned table.
  • The ALTER TABLE statement takes effect only for the new partitions of a table, which include the new partitions generated by using the INSERT OVERWRITE statement. New partitions are stored based on the hash clustering properties. The storage formats of existing partitions remain unchanged.
  • The ALTER TABLE statement takes effect only for the new partitions of a table. Therefore, you cannot specify a partition in this statement.

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. The query job can find the bucket that corresponds to the value 12345. This way, the job needs to scan data in only a single bucket instead of all the 1,000 buckets. This process is called bucket pruning.
2. Data in a bucket is sorted and stored based on the id column. MaxCompute automatically creates indexes and uses the INDEX LOOKUP function to locate relevant records.

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

• Hash clustering is not used.
• Hash clustering is used.