MaxCompute:Tabelas externas Parquet

Comportamento em caso de divergência de schema

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

• Fewer columns in the file than in the DDL: As colunas ausentes retornam NULL.

• Mais colunas no arquivo do que no DDL: As colunas excedentes são ignoradas.

• Incompatibilidade de tipo de coluna: A leitura falha com o erro ODPS-0123131:User defined function exception - Traceback:xxx. Alinhe os tipos de coluna no DDL com o schema real do arquivo Parquet para resolver o problema.

Sintaxe

Use a sintaxe simplificada quando não houver necessidade de compactação ou ajustes de desempenho. Use a sintaxe detalhada para configurar compactação e outras opções específicas do Parquet.

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/aliyunodpsdefaultrole',
    'mcfed.parquet.compression'='ZSTD/SNAPPY/GZIP'
)
STORED AS parquet
LOCATION '<oss_location>';

Para parâmetros comuns compartilhados entre tipos de tabelas externas do OSS, consulte Parâmetros básicos de sintaxe.

Parâmetros específicos

Predicate push down

Por padrão, consultas em tabelas externas Parquet com filtros WHERE varrem todos os dados, independentemente das condições de filtro. O Parquet Predicate Push Down (PPD) usa metadados de grupos de linhas incorporados aos arquivos Parquet para pular grupos que não correspondem ao filtro, reduzindo o volume de dados verificados, a latência da consulta e o uso de recursos.

O PPD exige o modo nativo (odps.ext.parquet.native=true). Ative-o executando estes comandos SET no nível de sessão antes da consulta:

-- Enable Native mode (required for PPD).
SET odps.ext.parquet.native = true;
-- Enable Parquet PPD.
SET odps.sql.parquet.use.predicate.pushdown = true;

Pré-requisitos

Antes de começar, certifique-se de ter:

• Um projeto MaxCompute

• Um bucket e uma pasta no OSS na mesma região do seu projeto MaxCompute. Consulte Criar um bucket e Gerencie pastas

• Permissões de acesso ao OSS por meio de uma conta Alibaba Cloud, um usuário RAM ou uma função RAM. Consulte Autorização STS para OSS

• A permissão CreateTable no seu projeto MaxCompute. Consulte Permissões do MaxCompute

Etapas

1. Prepare arquivos de dados formatados em ZSTD. No bucket oss-mc-test dos dados de amostra, crie o diretório parquet_zstd_jni/dt=20230418 e coloque os dados particionados nesse diretório.

2. Crie uma tabela externa Parquet compactada com ZSTD.

   CREATE EXTERNAL TABLE IF NOT EXISTS mc_oss_parquet_data_type_zstd (
       vehicleId INT,
       recordId INT,
       patientId INT,
       calls INT,
       locationLatitute DOUBLE,
       locationLongtitue DOUBLE,
       recordTime STRING,
       direction STRING
   )
   PARTITIONED BY (dt STRING)
   ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
   WITH serdeproperties(
     'odps.properties.rolearn'='acs:ram::<uid>:role/aliyunodpsdefaultrole',
     'mcfed.parquet.compression'='zstd'
   )
   STORED AS parquet
   LOCATION 'oss://oss-cn-hangzhou-internal.aliyuncs.com/oss-mc-test/parquet_zstd_jni/';

3. Adicione partições à tabela externa.

   MSCK REPAIR TABLE mc_oss_parquet_data_type_zstd ADD PARTITIONS;

   Para mais opções, consulte Adicionar partições a tabelas externas do OSS.

4. Leia dados da tabela.

   SELECT * FROM mc_oss_parquet_data_type_zstd WHERE dt='20230418' LIMIT 10;

   Saída esperada:

   +------------+------------+------------+------------+------------------+-------------------+----------------+------------+------------+
   | vehicleid  | recordid   | patientid  | calls      | locationlatitute | locationlongtitue | recordtime     | direction  | dt         |
   +------------+------------+------------+------------+------------------+-------------------+----------------+------------+------------+
   | 1          | 12         | 76         | 1          | 46.81006         | -92.08174         | 9/14/2014 0:10 | SW         | 20230418   |
   | 1          | 1          | 51         | 1          | 46.81006         | -92.08174         | 9/14/2014 0:00 | S          | 20230418   |
   | 1          | 2          | 13         | 1          | 46.81006         | -92.08174         | 9/14/2014 0:01 | NE         | 20230418   |
   | 1          | 3          | 48         | 1          | 46.81006         | -92.08174         | 9/14/2014 0:02 | NE         | 20230418   |
   | 1          | 4          | 30         | 1          | 46.81006         | -92.08174         | 9/14/2014 0:03 | W          | 20230418   |
   | 1          | 5          | 47         | 1          | 46.81006         | -92.08174         | 9/14/2014 0:04 | S          | 20230418   |
   | 1          | 6          | 9          | 1          | 46.81006         | -92.08174         | 9/14/2014 0:05 | S          | 20230418   |
   | 1          | 7          | 53         | 1          | 46.81006         | -92.08174         | 9/14/2014 0:06 | N          | 20230418   |
   | 1          | 8          | 63         | 1          | 46.81006         | -92.08174         | 9/14/2014 0:07 | SW         | 20230418   |
   | 1          | 9          | 4          | 1          | 46.81006         | -92.08174         | 9/14/2014 0:08 | NE         | 20230418   |
   | 1          | 10         | 31         | 1          | 46.81006         | -92.08174         | 9/14/2014 0:09 | N          | 20230418   |
   +------------+------------+------------+------------+------------------+-------------------+----------------+------------+------------+

5. Grave dados na tabela.

   INSERT INTO mc_oss_parquet_data_type_zstd PARTITION (dt = '20230418')
     VALUES (1,16,76,1,46.81006,-92.08174,'9/14/2014 0:10','SW');
   
   -- Verify the inserted row.
   SELECT * FROM mc_oss_parquet_data_type_zstd WHERE dt = '20230418' AND recordid=16;

   Saída esperada:

   +------------+------------+------------+------------+------------------+-------------------+----------------+------------+------------+
   | vehicleid  | recordid   | patientid  | calls      | locationlatitute | locationlongtitue | recordtime     | direction  | dt         |
   +------------+------------+------------+------------+------------------+-------------------+----------------+------------+------------+
   | 1          | 16         | 76         | 1          | 46.81006         | -92.08174         | 9/14/2014 0:10 | SW         | 20230418   |
   +------------+------------+------------+------------+------------------+-------------------+----------------+------------+------------+

Incompatibilidade de tipo de coluna

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)

O arquivo Parquet armazena a coluna como LongWritable (BIGINT), mas o DDL da tabela externa a define como INT. Altere o tipo da coluna no DDL de INT para BIGINT.

Erro de falta de memória durante gravação

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)

Esse erro ocorre ao gravar grandes volumes de dados em uma tabela externa Parquet. Resolva-o seguindo esta ordem:

1. Reduza mcfed.parquet.block.row.count.limit em tblproperties.

2. Se o OOM persistir ou se os arquivos de saída forem muito grandes, reduza também mcfed.parquet.page.size.row.check.max.

Adicione estas configurações no nível de sessão antes de gravar:

-- Set maximum JVM heap memory for UDFs.
SET odps.sql.udf.jvm.memory=12288;
-- Control batch size on the runtime side.
SET odps.sql.executionengine.batch.rowcount=64;
-- Set memory size per mapper.
SET odps.stage.mapper.mem=12288;
-- Adjust input data size per mapper (file split size).
SET odps.stage.mapper.split.size=64;