Parquet外部表

支持数据类型

MaxCompute数据类型详情请参见1.0数据类型版本、2.0数据类型版本。

• JNI模式：set odps.ext.parquet.native=false，表示读外部表解析Parquet数据文件时，使用原有基于Java的开源社区实现，支持读和写。

• Native模式：set odps.ext.parquet.native=true，表示读外部表解析Parquet数据文件时，使用新的基于C++的Native实现，仅支持读。

  模式	Java模式（读写）	Native模式（只读）
  TINYINT
  SMALLINT
  INT
  BIGINT
  BINARY
  FLOAT
  DOUBLE
  DECIMAL(precision,scale)
  VARCHAR(n)
  CHAR(n)
  STRING
  DATE
  DATETIME
  TIMESTAMP
  TIMESTAMP_NTZ
  BOOLEAN
  ARRAY
  MAP
  STRUCT
  JSON

支持压缩格式

• 当读写压缩属性的OSS文件时，需要在建表语句中添加with serdeproperties属性配置，详情请参见with serdeproperties属性参数。

• 支持读写的数据文件格式：以ZSTD、SNAPPY、GZIP方式压缩的Parquet。

语法结构

当Parquet文件中的Schema与外表Schema不一致时：

• 列数不一致：如果Parquet文件中的列数小于外表DDL的列数，则读取Parquet数据时，系统会将缺少的列值补充为NULL。反之（大于时），会丢弃超出的列数据。

• 列类型不一致：如果Parquet文件中的列类型与外表DDL中对应的列类型不一致，则读取Parquet数据时会报错。例如：使用STRING（或INT）类型接收Parquet文件中INT（或STRING）类型的数据，报错ODPS-0123131:User defined function exception - Traceback:xxx。

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>' 
;

公共参数

公共参数说明请参见基础语法参数说明。

独有参数

使用方式

• 开启Predicate Push Down（Parquet PPD）

  执行SQL查询前，通过set命令设置以下两个Session级别的参数以开启Parquet PDD。

  -- 开启parquet native reader。 
  set odps.ext.parquet.native = true; 
  -- 开启parquet ppd。
  set odps.sql.parquet.use.predicate.pushdown = true; 

• 使用示例

  基于1TB的TPCDS测试数据集，以tpcds_1t_store_sales Parquet外部表为例，开启PPD并执行过滤查询，总数据量为2879987999。

  -- 创建外部表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/aliyunodpsdefaultrole',
    'mcfed.parquet.compression'='zstd'
  )
  STORED AS parquet
  LOCATION 'oss://oss-cn-hangzhou-internal.aliyuncs.com/oss_bucket_path/';
  
  -- 使用1TB的TPCDS测试数据集。
  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;
  
  -- 执行查询。
  SELECT SUM(ss_sold_date_sk) FROM tpcds_1t_store_sales
    WHERE ss_sold_date_sk >= 2451871 AND ss_sold_date_sk <= 2451880;

性能对比结果

开启PPD功能可减少数据扫描量，从而降低查询延迟和资源消耗。

测试详情

1. Parquet外表+未开启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;

   

   

   

2. Parquet外表+开启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;

   

   

   存在很多空Mapper，无需读数据：

   

   实际裁减RowGroup的日志：

   

3. 内表+开启PPD

   内表数据有排序属性，因此裁减效果更明显。

   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;

   

   

   

Parquet文件列类型与外表DDL类型不一致

• 报错信息

  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)

• 错误描述

  Parquet文件的LongWritable字段类型与外表DDL的INT类型不一致。

• 解决方案

  外部表DDL中的INT类型需要改为BIGINT类型。

写外部表时报错java.lang.OutOfMemoryError

• 报错信息

  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)

• 错误描述

  大量数据写入Parquet外表时出现OOM报错。

• 解决方案

  建议创建外部表时，先调小mcfed.parquet.block.row.count.limit参数，如果还会发生OOM，或者输出文件太大，可以调小mcfed.parquet.page.size.row.check.max参数，更加频繁地检查内存。详情请参见独有参数。

  在向Parquet外表写入数据前，加上如下参数。

  -- 设置UDF JVM Heap使用的最大内存。
  SET odps.sql.udf.jvm.memory=12288;
  -- 控制Runtime侧batch size。
  SET odps.sql.executionengine.batch.rowcount =64;
  -- 设置每个Map Worker的内存大小。
  SET odps.stage.mapper.mem=12288;
  -- 修改每个Map Worker的输入数据量，即输入文件的分片大小，从而间接控制每个Map阶段下Worker的数量。
  SET odps.stage.mapper.split.size=64;