MaxCompute:Parquet 外部テーブル

サポートされるデータ型

MaxCompute のデータ型に関する詳細については、「データ型バージョン 1.0」および「データ型バージョン 2.0」をご参照ください。

• JNI モード： set odps.ext.parquet.native=false。これは、元の Java ベースのオープンソース実装を使用して Parquet データファイルを解析します。読み取りと書き込みの両方をサポートします。

• ネイティブモード： set odps.ext.parquet.native=true。これは、新しい C++ ベースのネイティブ実装を使用して Parquet データファイルを解析します。読み取りのみをサポートします。

  モード	Java モード (読み取り/書き込み)	ネイティブモード (読み取り専用)
  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 パラメーター」をご参照ください。

• サポートされる Parquet ファイル形式： ZSTD、SNAPPY、GZIP 圧縮。

スキーマエボリューションのサポート

Parquet 外部テーブルは、テーブルスキーマとファイル列を照合し、名前によって列値をマッピングします。

次の表のデータ互換性に関する注意では、スキーマが変更された外部テーブルで、新しいスキーマに一致するデータを読み取れるかどうか、また、新しいスキーマに一致しない既存の履歴データを読み取れるかどうかについて説明します。

構文

Parquet ファイルのスキーマが外部テーブルのスキーマと異なる場合：

• 列数の不一致： Parquet ファイルの列数が外部テーブルの DDL より少ない場合、不足している列は NULL を返します。Parquet ファイルの列数が多い場合、余分な列は無視されます。

• 列の型の不一致： Parquet ファイルの列の型が外部テーブルの DDL の対応する列の型と異なる場合、読み取りは失敗します。たとえば、STRING (または INT) を使用して 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>' 
;

共通パラメーター

共通パラメーターの詳細については、「基本構文パラメーター」をご参照ください。

固有パラメーター

使用方法

• 述語プッシュダウン (Parquet PPD) の有効化

  SQL クエリを実行する前に、SET コマンドを使用して次の 2 つのセッションレベルのパラメーターを設定し、Parquet PDD を有効にします。

  -- Parquet ネイティブリーダーを有効にします。
  set odps.ext.parquet.native = true; 
  -- Parquet PPD を有効にします。
  set odps.sql.parquet.use.predicate.pushdown = true; 

• 例

  1 TB の 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/';
  
  -- 1 TB の 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;

   

   

   多くのマッパーは空であり、データの読み取りをスキップします：

   

   実際の 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 外部テーブルに書き込む際に OutOfMemoryError が発生します。

• 解決策

  外部テーブルを作成する際に、まず mcfed.parquet.block.row.count.limit を減らします。OOM が解消されない場合や出力ファイルが大きすぎる場合は、mcfed.parquet.page.size.row.check.max を減らしてメモリをより頻繁にチェックします。詳細については、「固有パラメーター」をご参照ください。

  Parquet 外部テーブルに書き込む前に、これらの設定を追加します：

  -- UDF の最大 JVM ヒープメモリを設定します。
  SET odps.sql.udf.jvm.memory=12288;
  -- ランタイム側でバッチサイズを制御します。
  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;