MaxCompute：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;