MaxCompute:Range clustering

As tabelas com hash clustering oferecem as seguintes vantagens:

• Para consultar dados com base em um valor específico de coluna, utilize o algoritmo de hash para localizar diretamente o hash bucket. Esse processo chama-se bucket pruning. Se os dados no bucket estiverem armazenados de forma ordenada, use índices para localizá-los com maior precisão. Isso reduz o volume de dados verificados e melhora a eficiência da consulta.

• Ao unir duas tabelas por uma coluna específica e essa coluna estiver com hash em uma das tabelas, elimine a etapa de shuffle. Essa otimização economiza recursos computacionais.

Para obter mais informações sobre o recurso de hash clustering, consulte Hash clustering.

O hash clustering apresenta as seguintes limitações:

• O uso do algoritmo de hash para criar buckets pode causar problemas de skew de dados. Assim como ocorre no join skew, o data skew é inerente ao algoritmo de hash. Caso os dados de entrada estejam distribuídos de maneira desigual entre os buckets, haverá variação significativa no volume de dados em cada um deles. Como cada bucket geralmente funciona como uma unidade de processamento concorrente no hash clustering, essa disparidade tende a gerar long tails.

• O bucket pruning suporta apenas consultas de igualdade. Para consultas baseadas em condições de não igualdade, como valores de uma coluna maiores que 0, não é possível localizar os buckets específicos onde os dados residem. Nesse caso, a consulta precisa percorrer todos os buckets.

• Em consultas com múltiplas chaves de cluster, a melhoria de desempenho ocorre somente se todas as chaves estiverem presentes e todas as condições forem de igualdade.

  Por exemplo, ao consultar dados na tabela criada pela instrução abaixo, o desempenho melhora apenas se a condição for C1=x AND C2=y. O uso isolado de C1=x ou C2=y não permite a aceleração via hash clustering. Isso acontece porque os valores de hash das chaves são combinados em pares durante a consulta. Sem essa combinação, torna-se impossível localizar o bucket correto ou aplicar o bucket pruning.

  CREATE TABLE T2 (C1 int, C2 int, C3 string)
     CLUSTERED BY (C1, C2)
        SORTED by (C1, C2)
         INTO 1024 BUCKETS;

O range clustering divide os dados em vários intervalos disjuntos com base na ordenação completa das chaves de cluster. Cada intervalo é tratado como um bucket e deve atender simultaneamente às seguintes condições:

• Valores duplicados permanecem no mesmo bucket.

• A quantidade de valores em cada bucket é aproximadamente igual.

A instrução de exemplo a seguir cria uma tabela chamada T.

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

Os valores na coluna C1 são { 1, 8, -3, 2, 4, 1, 1, 3, 8, 20, -8, 9 }.

Após ativar o range clustering, obtêm-se os seguintes buckets:

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

• Bucket 1 : { 2, 3, 4 }

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

O MaxCompute executa o processo de range clustering automaticamente. Não é necessário definir manualmente cada intervalo. Em cenários de big data, a configuração manual seria ineficiente ou inviável. O sistema classifica e amostra os dados, gera um histograma baseado na distribuição de cada intervalo e, em seguida, combina e calcula esses histogramas. Dessa forma, o MaxCompute garante o desempenho ideal do range clustering.

Ao criar uma tabela, especifique tanto RANGE CLUSTERED BY quanto SORTED BY para assegurar a ordenação global dos dados. O MaxCompute cria então dois níveis de índices automaticamente: índice global e índice de arquivo. Esses índices permitem localizar e buscar valores-chave rapidamente, conforme ilustra a figura a seguir.

O range clustering oferece as seguintes vantagens em relação ao hash clustering:

• Suporte a consultas por intervalo.

  Por exemplo, se a condição da consulta for c < 3, o sistema exclui o Bucket 2 e o Bucket 3 com base no índice global e consulta os dados apenas nos Buckets 0 e 1. No hash clustering, o bucket pruning funciona exclusivamente para consultas de igualdade.

• Compatibilidade com consultas de múltiplas chaves.

  Se você especificar RANGE CLUSTERED BY (c1, c2, c3) SORTED BY (c1, c2, c3) durante a criação da tabela, o range clustering e o armazenamento dos dados seguem a ordem c1, c2 e c3. Isso possibilita consultas com condições complexas, como c1 = 100 AND c2 > 0 ou c1 = 100 AND c2 = 50 AND c3 < 5. O hash clustering não suporta esse tipo de consulta.

  Importante

  Em consultas com múltiplas chaves, as chaves na condição devem seguir a ordem de classificação definida, e apenas a última chave pode definir um intervalo de valores.

• Implementação eficiente de ordenação global.

  Antes do range clustering, o MaxCompute utilizava apenas uma instância para ordenar os dados globalmente, resultando em baixa eficiência. Com o range clustering, os dados em cada intervalo são classificados concurrentemente e depois combinados, o que aumenta significativamente a eficiência.

Utilize a instrução CREATE TABLE para criar uma tabela com range clustering. Nessa instrução, defina obrigatoriamente o parâmetro RANGE CLUSTERED BY. Os parâmetros INTO number_of_buckets BUCKETS e SORTED BY são opcionais. Na maioria dos casos, recomenda-se definir os mesmos valores em SORTED BY e RANGE CLUSTERED BY para obter o melhor efeito de otimização.

• Sintaxe

  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]

• Exemplos

  • Tabela não particionada

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

  • Tabela particionada

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

• Parâmetros

  • RANGE CLUSTERED BY

    Define as chaves para o range clustering. Após especificar este parâmetro, o MaxCompute classifica e amostra uma ou mais colunas de dados em intervalos adequados, conforme o número de buckets definido. Para evitar skew de dados, hot spots e melhorar o desempenho de consultas concorrentes, prefira colunas com ampla faixa de valores e poucas chaves duplicadas em RANGE CLUSTERED BY. Para otimizar o desempenho das consultas, indique as chaves de agregação ou filtros mais utilizadas.

  • SORTED BY

    Determina a ordenação dos campos dentro de um bucket. Para alcançar o melhor desempenho de consulta, alinhe os valores de SORTED BY com os de RANGE CLUSTERED BY. Ao definir SORTED BY, o MaxCompute gera automaticamente um índice global e índices de arquivo, utilizando-os para acelerar as consultas.

  • INTO number_of_buckets BUCKETS

    Diferentemente do hash clustering, o parâmetro INTO number_of_buckets BUCKETS é opcional no range clustering. Se omitido, o MaxCompute determina automaticamente a quantidade de buckets com base no volume de dados. Contudo, recomenda-se definir esse número conforme a necessidade real.

    Assim como no hash clustering, dimensione os buckets entre 512 MB e 1 GB para o range clustering. Tabelas excessivamente grandes exigem muitos buckets, mas evite ultrapassar 4.000 buckets por tabela.

Em tabelas particionadas, execute a instrução ALTER TABLE para adicionar ou remover as propriedades de range clustering.

• Sintaxe

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

• Notas de uso

  • A instrução ALTER TABLE modifica apenas as propriedades de clustering de tabelas particionadas. Em tabelas não particionadas, não é possível alterar as propriedades de clustering após a definição inicial.

  • A instrução ALTER TABLE aplica-se somente às novas partições da tabela, incluindo aquelas geradas por INSERT OVERWRITE. As novas partições seguem as propriedades de clustering definidas, enquanto o formato de armazenamento das partições existentes permanece inalterado.

  • Como a instrução ALTER TABLE afeta apenas novas partições, não é permitido especificar partições nesta instrução.

A instrução ALTER TABLE é adequada para tabelas existentes. Após adicionar as propriedades de range clustering, as novas partições serão armazenadas seguindo essas configurações.

Depois de criar uma tabela com range clustering, execute a instrução abaixo para visualizar suas propriedades. As informações de range clustering aparecem em Extended Info no resultado retornado.

DESC EXTENDED <table_name>;

Saída de exemplo:

Owner: ALIYUNxxx                | Project: xxx
  TableComment:

+----------------------------+
| CreateTime:               2018-01-15 16:05:30  |
| LastDDLTime:              2018-01-15 16:05:30  |
| LastModifiedTime:         2018-01-15 16:05:30  |
+----------------------------+
| InternalTable: YES   | Size: 0               |
+----------------------------+
| Native Columns:                                |
+----------------------------+
| Field            | Type     | Label | Comment  |
+----------------------------+
| l_orderkey       | bigint   |       |          |
| l_partkey        | bigint   |       |          |
| l_suppkey        | bigint   |       |          |
| l_linenumber     | bigint   |       |          |
| l_quantity       | double   |       |          |
| l_extendedprice  | double   |       |          |
| l_discount       | double   |       |          |
| l_tax            | double   |       |          |
| l_returnflag     | string   |       |          |
| l_linestatus     | string   |       |          |
| l_shipdate       | string   |       |          |
| l_commitdate     | string   |       |          |
| l_receiptdate    | string   |       |          |
| l_shipinstruct   | string   |       |          |
| l_shipmode       | string   |       |          |
| l_comment        | string   |       |          |
+----------------------------+
| Extended Info:                                 |
+----------------------------+
| TableID:                   xxx                 |
| IsArchived:                false               |
| PhysicalSize:              0                   |
| FileNum:                   0                   |
| ClusterType:               range               |
| BucketNum:                 1024                |
| ClusterColumns:            [l_orderkey]        |
| SortColumns:               [l_orderkey ASC]    |
+----------------------------+

Na seção Extended Info, os campos ClusterType como range, BucketNum como 1024, ClusterColumns como [l_orderkey] e SortColumns como [l_orderkey ASC] confirmam que a tabela foi configurada corretamente com os atributos de Range Clustering.

Para tabelas particionadas, também é possível executar a seguinte instrução para verificar as propriedades de clustering de uma partição específica.

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

Na saída, os quatro atributos ClusterType, BucketNum, ClusterColumns e SortColumns indicam a configuração de Range Clustering da partição.

odps@ tpch_100g>desc extended xndai_test_range partition(pt="20180115");

| PartitionSize: 0

| CreateTime:               2018-01-15 16:31:10
| LastDDLTime:              2018-01-15 16:31:10
| LastModifiedTime:         2018-01-15 16:31:10

| IsExstore:          false
| IsArchived:         false
| PhysicalSize:       0
| FileNum:            0
| ClusterType:        range
| BucketNum:          1024
| ClusterColumns:     [c1]
| SortColumns:        [c1 ASC]

Por exemplo, dada a condição simples id < 3, o otimizador extrai a condição e a converte no intervalo de valores (-∞, 3). O sistema usa então o índice global para realizar bucket pruning, excluindo o Bucket 2 e o Bucket 3, cujos dados estão fora desse intervalo. Em seguida, utiliza o índice de cada arquivo nos Buckets 0 e 1 para localizar os dados rapidamente. Esse processo chama-se predicate pushdown, conforme mostra a figura a seguir. A instrução de exemplo abaixo refere-se à consulta TPC-H Query 6, executada em um conjunto de dados de 100 GB após a aplicação do range clustering. Nessa consulta, realiza-se uma operação de agregação baseada em filtragem por intervalo. O range clustering aproveita os dois níveis de índices para localizar os dados rapidamente, reduzindo significativamente o tempo de execução e o consumo de recursos de CPU e memória.

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;

Neste exemplo, a instrução a seguir transforma a tabela mf_tab em uma tabela com range clustering, facilitando o entendimento das consultas com múltiplas chaves.

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

Após a conversão para range clustering, realize consultas agregadas no nível do projeto. Instrução de exemplo:

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

Use também múltiplas chaves para localizar uma tabela com precisão. Instrução de exemplo:

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

Realize consultas por intervalo usando múltiplas chaves. A instrução abaixo busca tabelas cujos nomes começam com adm.

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

Todas as consultas acima aproveitam integralmente a ordenação global do range clustering e aplicam predicate pushdown. Isso reduz o número de operações de I/O na varredura da tabela e economiza recursos de CPU e memória usados na filtragem e computação de dados.

Por exemplo, considere k1, k2 como chaves de cluster em uma tabela chamada T.

• Se a condição for k1 < 5, a aceleração por índice funciona corretamente.

• Para a condição k1 = 10 AND k2 = 20, a aceleração por índice também é aplicada.

• A condição k1 = 10 AND k2 < 0 igualmente permite a aceleração por índice.

• Caso a condição seja k2 < 0, a aceleração por índice falha, pois k1 não foi especificado.

• Se a condição for k1 < 0 AND k2 > 0, a aceleração por índice recupera os dados que satisfazem k1 < 0. Já para os dados que atendem a k2 > 0, torna-se necessária uma varredura completa da tabela.

Considere uma tabela T criada pela instrução CREATE TABLE abaixo. Ao consultar essa tabela, execute a operação GROUP BY diretamente no estágio map.

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;

A instrução a seguir exibe a estrutura de dados da tabela foo.

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

Os dados armazenados nos buckets da tabela foo abrangem os seguintes intervalos:

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

Os intervalos acima seguem o formato Bucket N: [lower bound values : upper bound values]. Se a agregação ocorrer pela Coluna a, os intervalos considerados serão:

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

Gere diretamente um plano de execução para operações de agregação pelas Colunas a e b, inicie três instâncias para agregar os dados de cada bucket e retorne o resultado final.

Entretanto, se os valores da Coluna a estiverem dispersos em vários buckets, o resultado será inválido. Exemplo:

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

A Coluna a possui dois valores, 3 e 7, armazenados em buckets distintos. Para garantir um resultado válido, agrupe as tuplas com o mesmo valor na Coluna a na mesma instância e agregue-as. Isso recria os buckets, conforme ilustra a figura a seguir. O espaço entre as duas linhas tracejadas vermelhas indica o intervalo de dados que cada instância pode ler. Histogramas são essenciais para o range clustering. Em uma tabela com range clustering onde as chaves de cluster e de ordenação coincidem, o worker correspondente a cada bucket amostra uma tupla a cada 10.000 linhas para obter os valores das chaves de cluster durante a inserção de dados. Os valores coletados compõem um histograma, armazenado nos arquivos de metadados do cluster. Esse tipo de histograma é conhecido como equi-depth histogram.

Após obter o histograma de cada bucket, recrie os buckets para cada worker seguindo estas regras:

• Tuplas com a mesma chave de agrupamento permanecem no mesmo bucket.

• Os dados distribuem-se uniformemente entre os buckets.

Com base nos limites inferiores de cada novo bucket, cada worker lê os dados dentro de um intervalo válido e retorna um resultado correto.

O teste abaixo utiliza a tabela partsupp, com 1 TB de dados em um conjunto TPC-H, para mensurar ganhos de desempenho. Execute a instrução a seguir para converter a tabela partsupp em uma tabela com 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;

Execute a seguinte instrução de consulta para realizar o teste:

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

• Execute o comando abaixo para desativar a otimização:

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

  O log de execução do job com a otimização desativada apresenta o seguinte conteúdo:

  
  resource cost: cpu 9.17 Core * Min, memory 15.15 GB * Min
  inputs:
      yuan_tpch_range_1t.partsupp: 800000000 (3376541880 bytes)
  outputs:
  Job run time: 14.000
  Job run mode: fuxi job
  Job run engine: execution engine
  M1:
      instance count: 128
      run time: 7.000
      instance time:
          min: 2.000, max: 4.000, avg: 2.000
      input records:
          TableScan1: 800000000 (min: 5496040, max: 6780663, avg: 6251772)
      output records:
          StreamLineWrite1: 200001977 (min: 1374023, max: 1695182, avg: 1562958)
      writer dumps:
          StreamLineWrite1: (min: 0, max: 0, avg: 0)
  R2_1:
      instance count: 43
      run time: 14.000
      instance time:
          min: 3.000, max: 4.000, avg: 3.000
      input records:
          StreamLineRead1: 200001977 (min: 4647288, max: 4654888, avg: 4651214)
      output records:
          AdhocSink1: 200000000 (min: 4647242, max: 4654834, avg: 4651168)
      reader dumps:
          StreamLineRead1: (min: 0, max: 0, avg: 0)

• Execute o comando abaixo para ativar a otimização:

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

  Saída de exemplo:

  
  resource cost: cpu 4.38 Core * Min, memory 4.38 GB * Min
  inputs:
      yuan_tpch_range_1t.partsupp: 800000000 (18493328320 bytes)
  outputs:
  Job run time: 6.000
  Job run mode: fuxi job
  Job run engine: execution engine
  M1:
      instance count: 128
      run time: 6.000
      instance time:
          min: 1.000, max: 3.000, avg: 2.000
      input records:
          TableScan1: 800000000 (min: 5625876, max: 6259956, avg: 6254874)
      output records:
          AdhocSink1: 200000000 (min: 1406469, max: 1564989, avg: 1563718)

Os resultados indicam que, com a otimização ativada, a velocidade da consulta aumentou 57%, o uso de CPU caiu 52% e o consumo de memória reduziu 71%. Os ganhos de desempenho variam conforme o volume de dados e o tipo de consulta.

Por padrão, a aceleração de ordenação global fica desativada em tabelas com range clustering. Para ativá-la, execute o seguinte comando:

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

Comparado ao hash clustering, o range clustering apresenta as seguintes limitações:

• O custo de geração de dados no range clustering supera o do hash clustering. Enquanto o hash clustering envolve apenas operações simples de hashing e ordenação, o range clustering exige amostragem, ordenação e combinação de histogramas. O consumo geral, incluindo tempo de execução, custos de CPU e memória, é maior. Portanto, se o hash clustering resolver o problema, prefira-o ao range clustering.

• Não há suporte para range clustering em DYNAMIC PARTITION ou INSERT INTO.

• O range clustering funciona apenas nas seguintes operações de join: inner join, left outer join, right outer join e semi join. Não há suporte para anti-join ou full outer join.

• Em tabelas com range clustering, as chaves definidas em RANGE CLUSTERED BY devem coincidir exatamente com as de SORTED BY. Por exemplo, se uma tabela foo for criada com range clustered by (a,b) sorted by (a,b), as otimizações descritas neste tópico funcionarão normalmente. Porém, se uma tabela bar usar range clustered by(a,b) sorted by (b,a), as otimizações não serão aplicáveis.

• As chaves usadas em JOIN ou GROUP BY devem corresponder aos prefixos ou ao conjunto completo de chaves definidas em RANGE CLUSTERED BY. Por exemplo, se a instrução de criação da tabela contiver range clustered by(a,b,c) sorted by(a,b,c), as otimizações deste tópico só funcionarão se a, a,b ou a,b,c forem as chaves no JOIN ou GROUP BY. O uso de b ou a,c como chave impedirá a aplicação das otimizações.

• Em tabelas particionadas com range clustering, a leitura de dados de duas ou mais partições simultaneamente inviabiliza as otimizações descritas neste tópico. Tais otimizações aplicam-se apenas a tabelas particionadas com uma única partição ou a tabelas não particionadas.