MaxCompute:Parquet external tables

Schema mismatch behavior

When the Parquet file schema differs from the external table DDL:

• Fewer columns in the file than in the DDL: Missing columns return NULL.

• More columns in the file than in the DDL: Extra columns are ignored.

• Column type mismatch: Reading fails with ODPS-0123131:User defined function exception - Traceback:xxx. Align the column types in the DDL with the actual Parquet file schema to resolve this.

Syntax

Use the simplified syntax when you do not need compression or performance tuning. Use the detailed syntax to configure compression and other Parquet-specific options.

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

For common parameters shared across OSS external table types, see Basic syntax parameters.

Unique parameters

Predicate push down

By default, queries on Parquet external tables with WHERE filters scan all data, regardless of filter conditions. Parquet Predicate Push Down (PPD) uses row group metadata embedded in Parquet files to skip row groups that cannot match the filter, reducing scanned data, query latency, and resource usage.

PPD requires Native mode (odps.ext.parquet.native=true). Enable it by running these session-level SET commands before your query:

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

Prerequisites

Before you begin, ensure that you have:

• A MaxCompute project

• An OSS bucket and folder in the same region as your MaxCompute project. See Create a bucket and Manage folders

• OSS access permissions via an Alibaba Cloud account, a RAM user, or a RAM role. See STS authorization for OSS

• The CreateTable permission in your MaxCompute project. See MaxCompute permissions

Steps

1. Prepare ZSTD-formatted data files. In the oss-mc-test bucket from the sample data, create the directory parquet_zstd_jni/dt=20230418 and place the partitioned data in that directory.

2. Create a ZSTD-compressed Parquet external table.

   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. Add partitions to the external table.

   MSCK REPAIR TABLE mc_oss_parquet_data_type_zstd ADD PARTITIONS;

   For more options, see Add partitions to OSS external tables.

4. Read data from the table.

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

   Expected output:

   +------------+------------+------------+------------+------------------+-------------------+----------------+------------+------------+
   | 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. Write data to the table.

   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;

   Expected output:

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

Column type mismatch

Error:

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)

The Parquet file stores the column as LongWritable (BIGINT), but the external table DDL defines it as INT. Change the column type in the DDL from INT to BIGINT.

Out-of-memory error when writing

Error:

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)

This error occurs when writing large amounts of data to a Parquet external table. Resolve it in this order:

1. Reduce mcfed.parquet.block.row.count.limit in tblproperties.

2. If OOM persists or output files are too large, also reduce mcfed.parquet.page.size.row.check.max.

Add these session-level settings before writing:

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