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。

支持Schema Evolution

Parquet外部表的Schema和文件列之间按名称映射列值。

下表中的数据兼容问题说明是指：已经进行Schema Evolution操作后的外表，针对“符合修改后的Schema结构的数据”是否可以正常读取；针对于“存量旧Schema数据”（即和修改后schema不匹配的存量数据）是否可以正常读取。

语法结构

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

   

   执行结果显示，Fuxi Jobs 中 M1 任务的 IO Records Input 为 2.9 G，IO Bytes Input 为 18.06 GB，Latency 为 00:00:18.000。

   执行结果的 Summary 页签显示：资源消耗为 cpu 19.25 Core * Min，memory 24.07 GB * Min；Job 运行时间为 23.000 秒，运行模式为 fuxi job 2.0。其中 M1 任务实例数 1404，运行时间 18.000 秒，输入记录数 2879987999，输出记录数 355；R2_1 任务实例数 1，运行时间 4.000 秒，输出记录数 1。

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;

   

   执行该查询后，作业 Summary 显示资源消耗为 cpu 11.47 Core * Min, memory 14.33 GB * Min，总运行时间 15 秒。其中 M1 阶段 instance count 为 1404，运行耗时 12 秒，输入记录数 762366649 条（约 3339386880 bytes）；R2_1 阶段 instance count 为 1，运行耗时 3 秒。

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

   

   实际裁减RowGroup的日志：

   [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. 内表+开启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;

   执行该查询后，作业 DAG 显示数据源总行数为 2,879,987,999，实际扫描行数仅为 32,830,000。M1 阶段（703 个实例）读取 32,830,000 行后输出 323 行，R2_1 阶段接收 323 行最终输出 1 行，表明内表开启 PPD 且数据有排序属性时裁剪效果显著。

   Fuxi Jobs 作业监控显示作业 SQL_0_1_0_job_0 已完成，包含 M1（703 个实例，Input 32.8M 条记录 / 1.52 GB，Output 323 条，Latency 00:00:09.375）和 R2_1（1 个实例，Input 323 条，Output 1 条，Latency 00:00:03.873）两个 Fuxi Task，状态均为 Terminated。M1 实例明细中存在 4 个 Data-Skew 实例，其中 M1#101_0、M1#103_0、M1#105_0 等实例的 Input 和 Output 均为 0，属于空跑实例，表明该查询存在数据倾斜问题。

   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)

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;