Paimon external tables

Background

Apache Paimon is a unified streaming and batch data lake storage format that provides high-throughput writes and low-latency queries. Paimon integrates well with common compute engines such as Spark, Hive, and Trino, which are used in services such as Realtime Compute for Apache Flink and E-MapReduce. Using Paimon, you can build a data lake on OSS and connect it to MaxCompute for analytics. For more information, see the official Apache Paimon documentation.

Prerequisites

You have the permissions to create MaxCompute tables (CreateTable). For more information, see MaxCompute permissions.
You have created a MaxCompute project. For instructions, see Create a project.
You have created an OSS bucket and a directory within it. For instructions, see Create a bucket.

Note
To prevent connectivity issues, create the bucket in the same region as your MaxCompute project.
You have activated Realtime Compute for Apache Flink. For instructions, see Activate Realtime Compute for Apache Flink.

Limitations

MaxCompute only supports reading from Paimon external tables. It does not support writing to them or automatically detecting schema changes in the source table.
Paimon does not support MaxCompute projects that have the schema feature enabled.
Paimon external tables do not support the clustering attribute.
Paimon external tables do not support features such as querying historical versions of data.

Paimon external table syntax

Use the following syntax to create a Paimon external table in MaxCompute:

CREATE EXTERNAL TABLE [IF NOT EXISTS] <mc_oss_extable_name>
(
<col_name> <data_type>,
...
)
[PARTITIONED BY (<col_name> <data_type>, ...)]
STORED BY 'org.apache.paimon.hive.PaimonStorageHandler'
WITH SERDEPROPERTIES (
    'odps.properties.rolearn'='acs:ram::xxxxxxxxxxxxx:role/aliyunodpsdefaultrole'
)
LOCATION '<oss_location>'
USING 'paimon_maxcompute_connector.jar';

If you create a partitioned table, you must run an additional command to load the partition data. For more information, see ORC external tables.

Method 1 (Recommended): Automatically parse the OSS directory structure to identify and add partitions to the external table.
```
MSCK REPAIR TABLE <mc_oss_extable_name> ADD PARTITIONS;
```

Method 2: Manually run the following command to add a partition.

ALTER TABLE <mc_oss_extable_name> ADD PARTITION (<col_name>= <col_value>);

Parameters

Parameter	Required	Description
mc_oss_extable_name	Yes	The name of the Paimon external table to create. Table names are case-insensitive. You do not need to consider the case when you query the external table.
col_name	Yes	The name of a column in the Paimon external table. When reading data from Paimon, the schema of the external table must match the schema of the Paimon data files. Otherwise, the read operation fails.
data_type	Yes	The data type of a column in the Paimon external table. When reading data from Paimon, the data type of each column in the external table must match the corresponding column's data type in the Paimon data files. Otherwise, the read operation fails.
odps.properties.rolearn	Yes	The Alibaba Cloud Resource Name (ARN) of a RAM role that has permissions to access OSS. You can obtain the ARN from the role details page in the RAM console.
oss_location	Yes	The path to the OSS data files. The format is `oss://<oss_endpoint>/<BucketName>/<oss_directory_name>/`. MaxCompute reads all data files under this path by default. oss_endpoint: The OSS access domain name. You must use an internal OSS endpoint to avoid data transfer fees. For example: `oss://oss-cn-beijing-internal.aliyuncs.com/xxx`. For a list of internal OSS endpoints, see Regions and endpoints. Note The OSS bucket should be in the same region as your MaxCompute project to prevent connectivity issues, as MaxCompute is deployed in specific regions. BucketName: The name of the OSS bucket. To find your bucket name, see List buckets. oss_directory_name: The name of the OSS directory. You do not need to specify a filename. Example: `oss://oss-cn-shanghai-internal.aliyuncs.com/oss-mc-test/Demo1/`

Procedure

Step 1: Prepare data in Flink

Create a Paimon catalog and a Paimon table, and then insert data into the table. If you already have a Paimon table with data in Realtime Compute for Apache Flink, you can skip this step.

Log on to the Realtime Compute for Apache Flink console and create a Paimon catalog. For details, see Create a Paimon catalog.
Create a Paimon table. For details, see Manage Paimon tables.
1. On the Metadata Management page, navigate to the Paimon catalog you created, select the default database, and click Create Table.
2. In the Add Table dialog box, select the Apache Paimon connector, enter the following statement, and click OK. This example creates a table named test_tbl.
```
CREATE TABLE `catalogname`.`default`.test_tbl (
    dt STRING,
    id BIGINT,
    data STRING,
    PRIMARY KEY (dt, id) NOT ENFORCED
) PARTITIONED BY (dt);
```
3. On the SQL Editor page, create, deploy, and run an SQL job that contains the following statement. For more information about how to create and run an SQL job, see Job development overview.
```
INSERT INTO `catalogname`.`default`.test_tbl VALUES ('2023-04-21', 1, 'AAA'), ('2023-04-21', 2, 'BBB'), ('2023-04-22', 1, 'CCC'), ('2023-04-22', 2, 'DDD');
```
  Note
  - Make sure that the engine version for the SQL job is vvr-8.0.1-flink-1.17 or later.
  - If your SQL job experiences backpressure (for example, from running INSERT INTO ... VALUES ... statements), go to the Job O&M page, edit the Execution Parameters, and in the Other Configurations section, set execution.checkpointing.checkpoints-after-tasks-finish.enabled: true. For details about how to configure runtime parameters for a job, see Configure Job Deployment Information.

Step 2: Upload the Paimon connector

Use one of the following methods to upload the Paimon connector to your MaxCompute project.

MaxCompute client

Use the client (odpscmd) to access the MaxCompute project and execute the following code to upload paimon_maxcompute_connector.jar to the project.

ADD JAR <path_to_paimon_maxcompute_connector.jar>;

DataWorks

Log on to the DataWorks console. In the left-side navigation pane, click Workspaces. Find your target workspace and click Go to DataStudio in the Actions column.
On the Data Development page, click the Create icon and select Resource > JAR.
In the Create Resource dialog box, configure the parameters for the new resource, upload paimon_maxcompute_connector.jar, and click Create. For more information, see Step 1: Create or upload a resource.

Set Resource Type to JAR and select Upload as ODPS resource.
After the resource is created, click the icon in the toolbar to submit the resource to the server.

Step 3: Create a Paimon external table

Connect by using the local client (odpscmd) or use another tool that can run MaxCompute SQL to create a MaxCompute Paimon external table. This topic uses oss_extable_paimon_1pt as an example.

CREATE EXTERNAL TABLE oss_extable_paimon_1pt
(
    id BIGINT,
    data STRING
)
PARTITIONED BY (dt STRING)
STORED BY 'org.apache.paimon.hive.PaimonStorageHandler'
WITH SERDEPROPERTIES (
    'odps.properties.rolearn'='acs:ram::124*********:role/aliyunodpsdefaultrole'
)
LOCATION 'oss://oss-cn-beijing-internal.aliyuncs.com/paimon_flink/test_db_y.db/test_tbl/'
USING 'paimon_maxcompute_connector.jar'
;

Step 4: Read the Paimon external table

Connect using the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL, and then run the following commands:
```
SET odps.sql.common.table.planner.ext.hive.bridge = true;
SET odps.sql.hive.compatible = true;
```

Run the following command to query the MaxCompute Paimon external table oss_extable_paimon_1pt.

SELECT * FROM oss_extable_paimon_1pt;

The following example result is returned:

+------------+------------+------------+
| id         | data       | dt         |
+------------+------------+------------+
| 1          | AAA        | 2023-04-21 |
| 2          | BBB        | 2023-04-21 |
+------------+------------+------------+

Note

If partitions are not displayed in the result, run the following command to add them:

MSCK REPAIR TABLE oss_extable_paimon_1pt ADD PARTITIONS;