MaxCompute:Parquet external tables

Sintaxe

Quando o schema do arquivo Parquet difere do schema da tabela externa:

• Incompatibilidade na quantidade de colunas: se o arquivo Parquet tiver menos colunas do que as definidas na DDL da tabela externa, as colunas ausentes retornarão NULL. Se o arquivo tiver mais colunas, o sistema ignorará as colunas extras.

• Incompatibilidade no tipo de coluna: se o tipo de uma coluna no arquivo Parquet não corresponder ao tipo correspondente na DDL, a operação de leitura falhará. Por exemplo, ocorrerá um erro como ODPS-0123131:User defined function exception - Traceback:xxx se você tentar ler uma coluna INT como um campo STRING.

CREATE EXTERNAL TABLE [IF NOT EXISTS] <mc_oss_extable_name>
(
  <col_name> <data_type>,
  ...
)
[COMMENT <table_comment>]
[PARTITIONED BY (<col_name> <data_type>, ...)]
STORED AS parquet 
LOCATION '<oss_location>' 
[tblproperties ('<tbproperty_name>'='<tbproperty_value>',...)];

CREATE EXTERNAL TABLE [IF NOT EXISTS] <mc_oss_extable_name>
(
  <col_name> <data_type>,
  ...
)
[COMMENT <table_comment>]
[PARTITIONED BY (<col_name> <data_type>, ...)]
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
WITH serdeproperties(
    'odps.properties.rolearn'='acs:ram::<uid>:role/<role_name>',
    'mcfed.parquet.compression'='ZSTD/SNAPPY/GZIP'
)
STORED AS parquet 
LOCATION '<oss_location>' 
;

Parâmetros comuns

Para mais informações sobre os parâmetros comuns, consulte Parâmetros de sintaxe básica.

Parâmetros específicos

Lista de permissões e lista de bloqueios

As tabelas externas do OSS no MaxCompute suportam filtragem por lista de permissões e lista de bloqueios. Defina os parâmetros de lista de permissões e lista de bloqueios em tblproperties para filtrar os arquivos a serem lidos de um diretório. Para mais detalhes, consulte Lista de permissões e lista de bloqueios.

Uso

• Ativar o Predicate Push Down (PPD)

  Antes de executar uma consulta SQL, use o comando set para definir os dois parâmetros a seguir no nível da sessão e ativar o PPD para Parquet.

  -- Enable the Parquet native reader. 
  set odps.ext.parquet.native = true; 
  -- Enable Parquet PPD.
  set odps.sql.parquet.use.predicate.pushdown = true; 

• Exemplo

  Este exemplo usa um conjunto de dados de teste TPC-DS de 1 TB e a tabela externa Parquet tpcds_1t_store_sales. Neste exemplo, com o PPD ativado, executamos uma consulta de filtro. O volume total de dados é de 2,879,987,999 linhas.

  -- Create the external table tpcds_1t_store_sales.
  CREATE EXTERNAL TABLE IF NOT EXISTS tpcds_1t_store_sales (
      ss_sold_date_sk         BIGINT,
      ss_sold_time_sk         BIGINT,
      ss_item_sk              BIGINT,
      ss_customer_sk          BIGINT,
      ss_cdemo_sk             BIGINT,
      ss_hdemo_sk             BIGINT,
      ss_addr_sk              BIGINT,
      ss_store_sk             BIGINT,
      ss_promo_sk             BIGINT,
      ss_ticket_number        BIGINT,
      ss_quantity             BIGINT,
      ss_wholesale_cost       DECIMAL(7,2),
      ss_list_price           DECIMAL(7,2),
      ss_sales_price          DECIMAL(7,2),
      ss_ext_discount_amt     DECIMAL(7,2),
      ss_ext_sales_price      DECIMAL(7,2),
      ss_ext_wholesale_cost   DECIMAL(7,2),
      ss_ext_list_price       DECIMAL(7,2),
      ss_ext_tax              DECIMAL(7,2),
      ss_coupon_amt           DECIMAL(7,2),
      ss_net_paid             DECIMAL(7,2),
      ss_net_paid_inc_tax     DECIMAL(7,2),
      ss_net_profit           DECIMAL(7,2)
  )
  ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
  WITH serdeproperties(
    'odps.properties.rolearn'='acs:ram::<uid>:role/<role_name>',
    'mcfed.parquet.compression'='zstd'
  )
  STORED AS parquet
  LOCATION 'oss://oss-cn-hangzhou-internal.aliyuncs.com/oss_bucket_path/';
  -- Use the 1 TB TPC-DS test dataset.
  INSERT OVERWRITE TABLE tpcds_1t_store_sales
  SELECT 
      ss_sold_date_sk,
      ss_sold_time_sk,
      ss_item_sk,
      ss_customer_sk,
      ss_cdemo_sk,
      ss_hdemo_sk,
      ss_addr_sk,
      ss_store_sk,
      ss_promo_sk,
      ss_ticket_number,
      ss_quantity,
      ss_wholesale_cost,
      ss_list_price,
      ss_sales_price,
      ss_ext_discount_amt,
      ss_ext_sales_price,
      ss_ext_wholesale_cost,
      ss_ext_list_price,
      ss_ext_tax,
      ss_coupon_amt,
      ss_net_paid,
      ss_net_paid_inc_tax,
      ss_net_profit
  FROM 
      bigdata_public_dataset.tpcds_1t.store_sales;
  -- Run the query.
  SELECT SUM(ss_sold_date_sk) FROM tpcds_1t_store_sales
    WHERE ss_sold_date_sk >= 2451871 AND ss_sold_date_sk <= 2451880;

Comparação de desempenho

Ativar o PPD reduz a quantidade de dados verificados, o que diminui a latência da consulta e o consumo de recursos.

Detalhes do teste

1. Tabela externa Parquet sem PPD

   SET odps.ext.parquet.native = true;
   SET odps.sql.parquet.use.predicate.pushdown = false;
   SELECT SUM(ss_sold_date_sk) FROM tpcds_1t_store_sales
     WHERE ss_store_sk = 2 AND ss_sold_date_sk >= 2451871 AND ss_sold_date_sk <= 2451880;

   

   Os resultados mostram que, para a tarefa M1 em Fuxi Jobs, IO Records Input é 2,9 G, IO Bytes Input é 18,06 GB e Latency é 00:00:18.000.

   A aba Summary dos resultados de execução mostra que o consumo de recursos é cpu 19.25 Core × Min e memória 24.07 GB × Min. O tempo de execução do job é de 23.000 segundos, e o modo de execução é fuxi job 2.0. A tarefa M1 tem 1.404 instâncias, tempo de execução de 18.000 segundos, 2.879.987.999 registros de entrada e 355 registros de saída. A tarefa R2_1 tem 1 instância, tempo de execução de 4.000 segundos e 1 registro de saída.

2. Tabela externa Parquet com PPD

   SET odps.ext.parquet.native = true;
   SET odps.sql.parquet.use.predicate.pushdown = true;
   SELECT SUM(ss_sold_date_sk) FROM tpcds_1t_store_sales
     WHERE ss_store_sk = 2 AND ss_sold_date_sk >= 2451871 AND ss_sold_date_sk <= 2451880;

   

   Após executar esta consulta, o Summary do job mostra que o consumo de recursos é cpu 11.47 Core × Min, memory 14.33 GB × Min, com um tempo total de execução de 15 segundos. O estágio M1 tem 1.404 instâncias, tempo de execução de 12 segundos e 762.366.649 registros de entrada (aproximadamente 3.339.386.880 bytes). O estágio R2_1 tem 1 instância e tempo de execução de 3 segundos.

   Muitos mappers estão vazios e não precisam ler dados:

   

   Log do recorte real de grupos de linhas:

   [2024-05-10 22:29:22.692182]    [INFO]   [239551]    [/home/admin/odps_build/workspace/IRDS_CMK_7u/jenkins-IRDS_CMK_7u-70
   16/common/table/file_formats/parquet/parquet_row_group_pruner.cpp:100]    The expression to prune row groups:(((ss_store_s
   k == 2:int64) and (ss_sold_date_sk >= 2451871:int64)) and (ss_sold_date_sk <= 2451880:int64))
   [2024-05-10 22:29:22.705508]    [INFO]   [239551]    [/home/admin/odps_build/workspace/IRDS_CMK_7u/jenkins-IRDS_CMK_7u-70
   16/common/table/file_formats/parquet/parquet_reader_factory.cpp:136]    Parquet row group pruning is enabled, millisecon
   ds elapsed:13    Total row group count:1 Pruned row group count:1    The first several row group indexes:
   [2024-05-10 22:29:22.705532]    [INFO]   [239551]    [/home/admin/odps_build/workspace/IRDS_CMK_7u/jenkins-IRDS_CMK_7u-70
   16/common/table/file_formats/parquet/parquet_reader_factory.cpp:60]  total feasible parquet row group count:0]

3. Tabela interna com PPD

   O efeito de recorte é mais significativo porque os dados na tabela interna estão ordenados.

   SELECT SUM(ss_sold_date_sk) FROM bigdata_public_dataset.tpcds_1t.store_sales
     WHERE ss_store_sk = 2 AND ss_sold_date_sk >= 2451871 AND ss_sold_date_sk <= 2451880;

   Após executar esta consulta, o DAG do job mostra que o número total de linhas na fonte de dados é 2.879.987.999, mas o número real de linhas verificadas é de apenas 32.830.000. O estágio M1 (703 instâncias) lê 32.830.000 linhas e gera 323 linhas. O estágio R2_1 recebe 323 linhas e gera 1 linha. Isso mostra que o efeito de recorte é significativo quando o PPD está ativado para uma tabela interna com dados ordenados.

   O monitoramento do Fuxi Jobs mostra que o job SQL_0_1_0_job_0 foi concluído. Ele inclui duas Fuxi Tasks, M1 e R2_1, ambas com status Terminated. M1 tem 703 instâncias, entrada de 32,8M registros (1,52 GB), saída de 323 registros e latência de 00:00:09.375. R2_1 tem 1 instância, entrada de 323 registros, saída de 1 registro e latência de 00:00:03.873. Os detalhes da instância M1 mostram 4 instâncias Data-Skew. Instâncias como M1#101_0, M1#103_0 e M1#105_0 têm 0 tanto para Input quanto para Output. Isso indica que são instâncias de dry-run e que esta consulta tem um problema de data skew.

   resource cost: cpu 5.62 Core * Min, memory 7.02 GB * Min
   inputs:
       lakehouse47_2.default.tpcds_1t_store_sales2: 32830000 (1633880386 bytes)
   outputs:
   Job run time: 14.000
   Job run mode: fuxi job 2.0
   Job run engine: execution engine
   M1:
       instance count: 703
       run time: 9.000
       instance time:
           min: 0.000, max: 2.000, avg: 0.000
       input records:
           TableScan1: 32830000  (min: 0, max: 210000, avg: 46699)
       output records:
           StreamLineWrite1: 323  (min: 0, max: 1, avg: 0)
       metrics_output_count:
           Calc1: 58025  (min: 0, max: 461, avg: 82)
           HashAgg1: 323  (min: 0, max: 1, avg: 0)
           StreamLineWrite1: 323  (min: 0, max: 1, avg: 0)
           TableScan1: 32830000  (min: 0, max: 210000, avg: 46699)
       metrics_inner_time_ms:
           Calc1: 4  (min: 0, max: 2, avg: 0)  MaxInstance: 21
           GlobalInit: 57752  (min: 60, max: 386, avg: 82)  MaxInstance: 17
           HashAgg1: 0  (min: 0, max: 0, avg: 0)   MaxInstance: 2
           StreamLineWrite1: 20469  (min: 5, max: 899, avg: 29)   MaxInstance: 400
           TableScan1: 131977  (min: 51, max: 1417, avg: 187)  MaxInstance: 301
   R2_1:
       instance count: 1
       run time: 4.000
       instance time:
           min: 0.000, max: 0.000, avg: 0.000
       input records:
           StreamLineRead1: 323  (min: 323, max: 323, avg: 323)
       output records:
           AdhocSink1: 1  (min: 1, max: 1, avg: 1)
       metrics_output_count:
           AdhocSink1: 1  (min: 1, max: 1, avg: 1)

Desempenho de consulta

Conjunto de dados: TPC-DS 1 TB

Recursos: mais de 900 CU

Método de teste: ETL

• Latência: o ZSTD é 13,4% mais rápido que o Snappy e 16,5% mais rápido que o formato sem compactação.

• CPU: o ZSTD usa 40,7% menos CPU que o Snappy e 57,75% menos que o formato sem compactação.

• Memória: o ZSTD usa 39,04% menos memória que o Snappy e 56,13% menos que o formato sem compactação.

• Armazenamento: o ZSTD usa 31,15% menos armazenamento que o Snappy e 45,8% menos que o formato sem compactação.

Eficiência de armazenamento

Conjunto de dados: TPC-DS 1 TB

Recursos: mais de 900 CU

Método de teste: ETL

• sem compactação: embora este formato não seja compactado, ele resulta em volumes de dados maiores e maior sobrecarga de E/S, o que leva a um desempenho geral ruim.

• snappy: a velocidade de compactação não é mais rápida que a do ZSTD de nível baixo, mas a taxa de compactação é maior. Isso resulta em um desempenho geral inferior ao do ZSTD de nível baixo.

• zstd: o tamanho dos dados de saída converge rapidamente. Níveis mais altos proporcionam uma compactação adicional mínima (apenas 13,87% a mais), mas causam um aumento rápido no tempo e no consumo de recursos, o que reduz drasticamente a relação custo-benefício. Para este cenário, o ZSTD de nível baixo (níveis 3 a 5) oferece os melhores resultados. O nível 3 é o padrão.

• No conjunto de dados TPC-DS de 1 TB, o ZSTD usou 31,1% menos espaço de armazenamento que o Snappy e 45,8% menos que o formato sem compactação.

Tipos de coluna incompatíveis entre um arquivo Parquet e uma DDL de tabela externa

• Mensagem de erro

  ODPS-0123131:User defined function exception - Traceback:
  java.lang.ClassCastException: org.apache.hadoop.io.LongWritable cannot be cast to org.apache.hadoop.io.IntWritable 
     at org.apache.hadoop.hive.serde2.objectinspector.primitive.WritableIntObjectInspector.getPrimitiveJavaObject(WritableIntObjectInspector.java:46)

• Descrição do erro

  O tipo de campo LongWritable do arquivo Parquet não corresponde ao tipo INT na DDL da tabela externa.

• Solução

  Altere o tipo INT na DDL da tabela externa para BIGINT.

Erro ao gravar em uma tabela externa: java.lang.OutOfMemoryError

• Mensagem de erro

  ODPS-0123131:User defined function exception - Traceback:
  java.lang.OutOfMemoryError: Java heap space
  	at java.io.ByteArrayOutputStream.<init>(ByteArrayOutputStream.java:77)
  	at org.apache.parquet.bytes.BytesInput$BAOS.<init>(BytesInput.java:175)
  	at org.apache.parquet.bytes.BytesInput$BAOS.<init>(BytesInput.java:173)
  	at org.apache.parquet.bytes.BytesInput.toByteArray(BytesInput.java:161)

• Descrição do erro

  Ocorre um erro de falta de memória (OOM) quando você grava um grande volume de dados em uma tabela externa Parquet.

• Solução

  Ao criar uma tabela externa, primeiro diminua o parâmetro mcfed.parquet.block.row.count.limit. Se um erro OOM ainda ocorrer ou o arquivo de saída for muito grande, diminua o parâmetro mcfed.parquet.page.size.row.check.max para verificar a memória com mais frequência. Para mais informações, consulte Parâmetros específicos.

  Antes de gravar dados na tabela externa Parquet, adicione os seguintes parâmetros.

  -- Set the maximum memory size for the UDF JVM heap.
  SET odps.sql.udf.jvm.memory=12288;
  -- Control the batch size on the runtime side.
  SET odps.sql.executionengine.batch.rowcount =64;
  -- Set the memory size for each Map worker.
  SET odps.stage.mapper.mem=12288;
  -- Set the input data volume for each Map worker (input file shard size) to indirectly control the number of workers per Map stage.
  SET odps.stage.mapper.split.size=64;