Creating and using Parquet external tables in MaxCompute-MaxCompute(MaxCompute)-阿里云帮助中心

Scope

OSS external tables do not support the cluster property.
The size of a single file cannot exceed 2 GB. You must split files that are larger than 2 GB.
MaxCompute and OSS must be in the same region.
Supported data types.
Supported compression formats.
Schema evolution support.

Create an external table

Syntax

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

Column count mismatch: If the Parquet file has fewer columns than defined in the external table DDL, the missing columns return NULL. If the file has more columns, the extra columns are ignored.
Column type mismatch: If a column type in the Parquet file does not match the corresponding type in the DDL, the read operation fails. For example, an error such as ODPS-0123131:User defined function exception - Traceback:xxx is reported if you try to read an INT column as a STRING field.

Simplified syntax

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>',...)];

Detailed syntax

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

Common parameters

For more information about common parameters, see Basic syntax parameters.

Unique parameters

with serdeproperties parameters

property_name	When to use	Description	property_value	Default
mcfed.parquet.compression	Add this property to write Parquet data to OSS in a compressed format.	Parquet compression property. Parquet data is uncompressed by default.	ZSTD SNAPPY GZIP	None
mcfed.parquet.compression.codec.zstd.level	Add this property when `'mcfed.parquet.compression'='zstd'`. If this property is not specified, the default compression level of 3 is used.	A higher level increases the compression ratio. However, tests show that high levels provide minimal gains in data size reduction while significantly increasing time and resource consumption. For big data scenarios, a low ZSTD level (3 to 5) provides the best balance of performance and compression. For example: `'mcfed.parquet.compression.codec.zstd.level'= '5'`.	The value can be from 1 to 22.	3
parquet.file.cache.size	Add this property to improve the performance of reading OSS data files when processing Parquet data.	Specifies the amount of data that can be cached when reading OSS data files. Unit: KB.	1024	None
parquet.io.buffer.size	Add this property to improve the performance of reading OSS data files when processing Parquet data.	Specifies the amount of data that can be cached when the OSS data file size exceeds 1024 KB. Unit: KB.	4096	None

tblproperties parameters

property_name	When to use	Description	property_value	Default
io.compression.codecs	Add this property if your OSS data files are in Raw-Snappy format.	Enables the built-in open source parser for the SNAPPY format. If you set this parameter to True, MaxCompute can read the compressed data. Otherwise, the read operation fails.	com.aliyun.odps.io.compress.SnappyRawCodec.	None
odps.external.data.output.prefix (Compatible with odps.external.data.prefix)	Add this property to specify a custom prefix for output files.	Must contain only letters, digits, and underscores (a-z, A-Z, 0-9, _). The length must be 1 to 10 characters.	A valid combination of characters, such as 'mc_'.	None
odps.external.data.enable.extension	Add this property to show the extension of output files.	Set to True to show the file extension. Otherwise, the extension is hidden.	True False	False
odps.external.data.output.suffix	Add this property to specify a custom suffix for output files.	Must contain only letters, digits, and underscores (a-z, A-Z, 0-9, _).	A valid combination of characters, such as '_hangzhou'.	None
odps.external.data.output.explicit.extension	Add this property to specify a custom extension for output files.	Must contain only letters, digits, and underscores (a-z, A-Z, 0-9, _). The length must be 1 to 10 characters. This parameter has a higher priority than `odps.external.data.enable.extension`.	A valid combination of characters, such as "jsonl".	None
mcfed.parquet.compression	Add this property to write Parquet data to OSS in a compressed format. No extra parameters are needed to read compressed files.	Parquet compression property. Parquet data is uncompressed by default.	SNAPPY GZIP ZSTD	None
mcfed.parquet.block.size	Controls the block size of Parquet files, which affects storage efficiency and read performance.	Parquet tuning property. Defines the Parquet block size in bytes.	Non-negative integer	134217728 (128 MB)
mcfed.parquet.block.row.count.limit	When writing data to a Parquet external table, limits the number of records in each row group to prevent out-of-memory (OOM) errors.	Parquet tuning property. Controls the maximum number of records per row group. If an OOM error occurs, reduce the value of this parameter. Suggestion: If the JVM memory is only 1 GB and the average record size is 1 MB, set this parameter to about 100. The default row group size is 128 MB. Do not set this parameter to a very small value.	Non-negative integer	2147483647 (Integer.MAX_VALUE)
mcfed.parquet.page.size.row.check.min	When writing data to a Parquet external table, controls the frequency of memory checks to prevent OOM errors.	Parquet tuning property. Limits the minimum number of records between memory checks. If an OOM error occurs, reduce the value of this parameter.	Non-negative integer	100
mcfed.parquet.page.size.row.check.max	When writing data to a Parquet external table, controls the frequency of memory checks to prevent OOM errors.	Parquet tuning property. Limits the minimum number of records between memory checks. If an OOM error occurs, reduce the value of this parameter. Because frequent memory checks add overhead, adjusting this parameter may affect performance. Parameter recommendations: By default, a memory check is performed every 10,000 records. If the record size is small, you can set this parameter to a smaller value, such as 1000, to perform memory checks more frequently and prevent OOM errors. First, try reducing the value of `mcfed.parquet.block.row.count.limit`. If OOM errors persist or the output files are too large, reduce the value of `mcfed.parquet.page.size.row.check.max` to check memory more frequently.	Non-negative integer	1000
mcfed.parquet.compression.codec.zstd.level	Add this property to specify the compression level of the ZSTD algorithm when writing Parquet data to OSS with ZSTD compression.	Parquet compression property. Specifies the compression level of the ZSTD algorithm. The value can be from 1 to 22.	Non-negative integer	3

Whitelist and blacklist

MaxCompute OSS external tables support whitelist and blacklist filtering. By setting whitelist and blacklist parameters in tblproperties, you can filter which files to read from a directory. For details, see Whitelist and blacklist.

Write data

For details about the write syntax in MaxCompute, see Write syntax.

Query and analysis

See Query syntax for details on SELECT syntax.
See Query optimization for details on optimizing query plans.
For more information about reading LOCATION files directly, see Feature: Schemaless Query.
Query optimization: Parquet external tables support query optimization by enabling Predicate Push Down (PPD). For performance results, see Predicate Push Down (Parquet PPD) support.

Add the following parameters before your SQL statement to enable PPD:
```
-- PPD parameters must be used in Native mode, which means the Native switch must be set to true.
-- Enable the Parquet native reader.
SET odps.ext.parquet.native = true; 
-- Enable Parquet PPD.
SET odps.sql.parquet.use.predicate.pushdown = true; 
```

Predicate Push Down (Parquet PPD) support

By default, Parquet external tables do not support Predicate Push Down (PPD). When you run a query with a WHERE filter condition, MaxCompute scans all data. This causes unnecessary I/O, resource consumption, and query latency. To resolve this issue, you can enable PPD using a parameter. This feature uses the metadata in Parquet files to filter data at the row group level during the scan phase, which improves query performance and reduces resource consumption and costs.

Usage

Enable Predicate Push Down (PPD)

Before you run an SQL query, use the set command to set the following two session-level parameters to enable Parquet PPD.

-- Enable the Parquet native reader. 
set odps.ext.parquet.native = true; 
-- Enable Parquet PPD.
set odps.sql.parquet.use.predicate.pushdown = true;

Example

This example uses a 1 TB TPC-DS test dataset and the tpcds_1t_store_sales Parquet external table. In this example, PPD is enabled and a filter query is run. The total data volume is 2,879,987,999 rows.

-- Create the external table 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/';
-- Use the 1 TB TPC-DS test dataset.
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;
-- Run the query.
SELECT SUM(ss_sold_date_sk) FROM tpcds_1t_store_sales
  WHERE ss_sold_date_sk >= 2451871 AND ss_sold_date_sk <= 2451880;

Performance comparison

Enabling PPD reduces the amount of data scanned, which lowers query latency and resource consumption.

Mode	Total rows in table	Rows scanned	Bytes scanned	Mapper time	Total resource consumption	Description
Parquet external table without PPD	2,879,987,999	2,879,987,999 (100%)	19386793984 (100%)	18s	CPU 19.25 Core-minutes, Memory 24.07 GB-minutes 100%
Parquet external table with PPD	2,879,987,999	762,366,649 (26.47%)	3,339,386,880 (17.22%)	12s	cpu 11.47 Core × Min, memory 14.33 GB × Min ~59.58%	Reduced data scanning significantly lowers latency and resource consumption.
Internal table with PPD	2,879,987,999	32,830,000 (1.14%)	1,633,880,386 (8.43%)	9s	cpu 5.62 Core × Min, memory 7.02 GB × Min ~29.19%	PPD is more effective on internal tables because the data is sorted.

Test details

Parquet external table without 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;
```
The results show that for the M1 task in Fuxi Jobs, IO Records Input is 2.9 G, IO Bytes Input is 18.06 GB, and Latency is 00:00:18.000.

The Summary tab of the execution results shows that the resource consumption is cpu 19.25 Core × Min and memory 24.07 GB × Min. The job runtime is 23.000 seconds, and the run mode is fuxi job 2.0. The M1 task has 1,404 instances, a runtime of 18.000 seconds, 2,879,987,999 input records, and 355 output records. The R2_1 task has 1 instance, a runtime of 4.000 seconds, and 1 output record.

Parquet external table with 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;

After you run this query, the job Summary shows that the resource consumption is cpu 11.47 Core × Min, memory 14.33 GB × Min, with a total runtime of 15 seconds. The M1 stage has 1,404 instances, a runtime of 12 seconds, and 762,366,649 input records (approximately 3,339,386,880 bytes). The R2_1 stage has 1 instance and a runtime of 3 seconds.

Many mappers are empty and do not need to read data:

Log of actual row group clipping:

[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]

Internal table with PPD

The clipping effect is more significant because the data in the internal table is sorted.

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;

After you run this query, the job DAG shows that the total number of rows in the data source is 2,879,987,999, but the actual number of scanned rows is only 32,830,000. The M1 stage (703 instances) reads 32,830,000 rows and outputs 323 rows. The R2_1 stage receives 323 rows and outputs 1 row. This shows that the clipping effect is significant when PPD is enabled for an internal table with sorted data.

The Fuxi Jobs monitoring shows that job SQL_0_1_0_job_0 is complete. It includes two Fuxi Tasks, M1 and R2_1, both with a Terminated status. M1 has 703 instances, an input of 32.8M records (1.52 GB), an output of 323 records, and a latency of 00:00:09.375. R2_1 has 1 instance, an input of 323 records, an output of 1 record, and a latency of 00:00:03.873. The M1 instance details show 4 Data-Skew instances. Instances such as M1#101_0, M1#103_0, and M1#105_0 have 0 for both Input and Output. This indicates that they are dry-run instances and that this query has a data skew issue.

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 and ZSTD performance comparison

The following section compares the performance of different compression formats for Parquet external tables.

Note: The test results are for reference only. Performance may vary based on the business scenario. We recommend that you perform further testing and evaluation for your specific use case.

Query performance

Dataset: TPC-DS 1 TB

Resources: 900+ CU

Test method: ETL

Metric	Parquet uncompressed	Parquet-Snappy	Parquet-ZSTD
job runtime (s)	4372	4215	3649
CPU cost	14211.89	10131.36	6004.26
Memory cost	26852.91	19323.27	11778.06
Storage (GB)	425.94	335.33	230.87

Latency: ZSTD is 13.4% faster than Snappy and 16.5% faster than uncompressed.
CPU: ZSTD uses 40.7% less CPU than Snappy and 57.75% less than uncompressed.
Memory: ZSTD uses 39.04% less memory than Snappy and 56.13% less than uncompressed.
Storage: ZSTD uses 31.15% less storage than Snappy and 45.8% less than uncompressed.

Storage efficiency

Dataset: TPC-DS 1 TB

Resources: 900+ CU

Test method: ETL

uncompressed: Although this format is uncompressed, it results in larger data volumes and higher I/O overhead, which leads to poor overall performance.
snappy: The compression speed is not faster than low-level ZSTD, but the compression ratio is higher. This results in poorer overall performance than low-level ZSTD.
zstd: The output data size converges quickly. Higher levels provide minimal additional compression (only 13.87% more) but cause a rapid increase in time and resource consumption, which drastically reduces cost-effectiveness. For this scenario, low-level ZSTD (levels 3 to 5) provides the best results. Level 3 is the default.
On the TPC-DS 1 TB dataset, ZSTD used 31.1% less storage space than Snappy and 45.8% less than uncompressed.

Compression format	Output data size (GB / Compression ratio)	Job runtime (s)	TableSink time (s, % of job runtime)	CPU (Core × Min)	Memory (GB × Min)
uncompressed	486.67 (100%)	256.406	~ 134.61 (52.5%)	2353.19	3361.71
snappy	238.33 (48.97%)	239.087	~ 73.88 (30.9%)	2110.31	3014.73
zstd (level 1, min)	164.71 (33.84%)	233.170	~ 65.75 (28.2%)	2110.23	3014.61
zstd (level 2)	165.3 (33.97%)	231.226	~ 64.51 (27.9%)	2100.79	3001.13
zstd (level 3, default)	158.9 (32.65%)	236.985	~ 67.07 (28.3%)	2115.10	3021.57
zstd (level 4)	159.52 (32.77%)	232.477	~ 67.65 (29.1%)	2100.13	3000.19
zstd (level 5)	157.89 (32.44%)	232.248	~ 71.07 (30.6%)	2103.96	3005.66
zstd (level 6)	160.47 (32.97%)	236.669	~ 78.10 (33.0%)	2137.63	3053.75
zstd (level 9)	152.00 (31.23%)	254.073	~ 100.36 (39.5%)	2287.61	3268.01
zstd (level 14)	144.63 (29.72%)	455.019	~ 341.26 (75.5%)	4076.00	5822.86
zstd (level 19)	150.87 (31.00%)	727.841	~ 614.30 (84.4%)	6933.10	9904.43
zstd (level 22, max)	150.81 (30.99%)	5381.359	~ 5,257.59 (97.7%)	42848.13	61211.62

Example scenario

This example shows how to create a partitioned Parquet external table with ZSTD compression, and then read from and write to the table.

Prerequisites
1. You have created a MaxCompute project.
2. You have prepared an OSS bucket and directory. For more information, see Create a bucket and Manage directories.
  
  Ensure your bucket is in the same region as your MaxCompute project.
3. Grant permissions.
  1. You have permission to access OSS. You can access an OSS external table by using an Alibaba Cloud account, a RAM user, or a RAM role. For more information about how to grant permissions, see STS-mode authorization for OSS.
  2. You have the CreateTable permission in the MaxCompute project. For more information about table-related permissions, see MaxCompute permissions.
Prepare a data file in ZSTD format.

In the oss-mc-test bucket for the sample data, create the parquet_zstd_jni/dt=20230418 folder and store the data file in the dt=20230418 partition folder.

Create a Parquet foreign table that uses the ZSTD compression format.

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

Import partition data. If the OSS foreign table is a partitioned table, you must also import the partition data. For more information, see OSS foreign tables.
```
-- Import partition data.
MSCK REPAIR TABLE mc_oss_parquet_data_type_zstd ADD PARTITIONS;
```

Read data from the Parquet external table.

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

The following sample output is returned:

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

Write data to the Parquet external 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');
-- Query the newly written data
SELECT * FROM mc_oss_parquet_data_type_zstd WHERE dt = '20230418' AND recordid=16;

The result is as follows:

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

Supported data types

For more information about MaxCompute data types, see Data types (version 1.0) and Data types (version 2.0).

Java Native Interface (JNI) mode: set odps.ext.parquet.native=false. This mode uses the original open source Java-based implementation to parse Parquet data files when you read from a foreign table. It supports both read and write operations.

Native mode: set odps.ext.parquet.native=true. This mode uses the new C++-based native implementation to parse Parquet data files when you read from a foreign table. It supports only read operations.

Mode	Java mode (read/write)	Native mode (read-only)
TINYINT	Supported	Supported
SMALLINT	Supported	Supported
INT	Supported	Supported
BIGINT	Supported	Supported
BINARY	Supported	Supported
FLOAT	Supported	Supported
DOUBLE	Supported	Supported
DECIMAL(precision,scale)	Not supported	Supported
VARCHAR(n)	Supported	Supported
CHAR(n)	Supported	Supported
STRING	Supported	Supported
DATE	Supported	Supported
DATETIME	Supported	Supported
TIMESTAMP	Supported	Supported
TIMESTAMP_NTZ	Not supported	Not supported
BOOLEAN	Supported	Supported
ARRAY	Supported	Supported
MAP	Supported	Supported
STRUCT	Supported	Supported
JSON	Not supported	Not supported

Supported compression formats

To read or write compressed OSS files, you can add the with serdeproperties property configuration to the table creation statement. For more information, see with serdeproperties property parameters.

Compression property	Read	Write
Gzip	Supported	Supported
ZSTD	Supported	Supported
SNAPPY (SnappyRawCodec)	Supported	Supported
SNAPPY (SnappyCodec)	Supported	Not supported

Support for schema evolution

Parquet foreign tables map column values between the schema and file columns by name.

The Data compatibility issues column in the following table describes whether data can be read correctly after a schema evolution operation. This applies to both new data that conforms to the modified schema and historical data that uses the old schema.

Operation type	Supported	Description	Data compatibility issues
Add column	Supported	New columns are added at the end of the table. You cannot specify their position. If you add a column with a default value, the default value applies only to data written by MaxCompute.	Data that matches the modified schema can be read. If no modifications are made to the columns of existing data that uses the old schema, the table is read using the new schema. For example, if you add a column, historical data for that column is read as NULL.
Delete column	Supported	Parquet foreign tables map column values by name.	Compatible
Reorder columns	Supported	Parquet foreign tables map column values by name.	Compatible
Change column data type	Not supported	This operation is not supported. The Parquet format has strict schema validation. Changing a data type can make the data unreadable.	Not applicable
Rename column	Not supported	This operation is not supported. The Parquet format has strict schema validation, which may cause previously compatible types to become unreadable after modification.	Not applicable
Modify column comment	Supported	The comment must be a valid string of no more than 1024 bytes. Otherwise, an error occurs.	Compatible
Modify the non-null property of a column	Not supported	This operation is not supported. Columns are nullable by default.	Not applicable

FAQ

Mismatched column types between a Parquet file and a foreign table DDL

Error message

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)

Error description

The LongWritable field type of the Parquet file does not match the INT type in the foreign table DDL.
Solution

Change the INT type in the foreign table DDL to BIGINT.

Error when writing to a foreign table: `java.lang.OutOfMemoryError`

Error message

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)

Error description

An out-of-memory (OOM) error occurs when you write a large volume of data to a Parquet foreign table.

Solution

When you create a foreign table, first decrease the mcfed.parquet.block.row.count.limit parameter. If an OOM error still occurs or the output file is too large, decrease the mcfed.parquet.page.size.row.check.max parameter to check memory more frequently. For more information, see Unique parameters.

Before you write data to the Parquet foreign table, add the following parameters.

-- Set the maximum memory size for the UDF JVM heap.
SET odps.sql.udf.jvm.memory=12288;
-- Control the batch size on the runtime side.
SET odps.sql.executionengine.batch.rowcount =64;
-- Set the memory size for each Map worker.
SET odps.stage.mapper.mem=12288;
-- Set the input data volume for each Map worker (input file shard size) to indirectly control the number of workers per Map stage.
SET odps.stage.mapper.split.size=64;