Paimon basic features - Realtime Compute for Apache Flink

Prerequisites

If you use a RAM user or a RAM role, ensure that it has the required permissions to access the Realtime Compute for Apache Flink console. For more information, see Permission management.
You have created a Flink workspace. For more information, see Activate Realtime Compute for Apache Flink.

You have activated Object Storage Service (OSS) and created a bucket of the Standard storage class. For more information, see Get started by using the OSS console. OSS is used to store files for Paimon tables, including data files and metadata files.
Paimon tables are supported only in Ververica Runtime (VVR) 8.0.5 or later.

Step 1: Create a Paimon catalog

Go to the script editor.
1. Log on to the Realtime Compute for Apache Flink console.
2. In the Actions column of the target workspace, click Console.
3. In the left navigation bar, click Data Development > Data Query. On the Query Script tab, create a new Query Script.

In the editor, enter the following code to create a Paimon catalog.

-- my-catalog is a custom catalog name.
CREATE Catalog `my-catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'filesystem',
  'warehouse' = '<warehouse>',
  'fs.oss.endpoint' = '<fs.oss.endpoint>',
  'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
  'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
);

The following table describes the parameters.

Parameter	Description	Required	Remarks
type	The catalog type.	Yes	The value must be 'paimon'.
metastore	The metastore type.	Yes	This example uses filesystem. For more information about other types, see Manage Paimon catalogs.
warehouse	The warehouse directory in your OSS bucket.	Yes	The format is oss://<bucket>/<object>, where: bucket: the name of the OSS bucket that you created. object: the path where you want to store data. You can find your bucket and object names on the OSS console.
fs.oss.endpoint	The endpoint of the OSS service.	No	Required if the OSS bucket specified by warehouse is in a different region from your Flink workspace or belongs to another Alibaba Cloud account. For more information, see Regions and Endpoints. Note If you want to store a Paimon table in OSS-HDFS, you must specify the fs.oss.endpoint, fs.oss.accessKeyId, and fs.oss.accessKeySecret parameters. The value of fs.oss.endpoint is in the cn-<region>.oss-dls.aliyuncs.com format, for example, cn-hangzhou.oss-dls.aliyuncs.com.
fs.oss.accessKeyId	The AccessKey ID of an Alibaba Cloud account or a RAM user that has permissions to read from and write to OSS.	No	Required if the OSS bucket specified by warehouse is in a different region from your Flink workspace or belongs to another Alibaba Cloud account. For information about how to obtain an AccessKey pair, see Create an AccessKey pair.
fs.oss.accessKeySecret	The AccessKey secret of the Alibaba Cloud account or RAM user that has permissions to read from and write to OSS.	No

Select the code that creates the Paimon catalog and click Run on the left.

A The following statement has been executed successfully! message indicates that the catalog is created.

Step 2: Create a Paimon database and table

On the Scripts page, enter the following code to create a Paimon database named my_db and a Paimon table named my_tbl.
```
CREATE DATABASE `my-catalog`.`my_db`;
CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  dt STRING,
  id BIGINT,
  content STRING,
  PRIMARY KEY (dt, id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'changelog-producer' = 'lookup'  
);
```
Note
To enable streaming consumption of the Paimon table later, this example sets changelog-producer to lookup in the WITH clause, which uses the lookup strategy to generate the changelog. For more information, see Changelog production.
Select the code that creates the Paimon database and Paimon table, and then click Run on the left.

If the message The following statement has been executed successfully! is returned, the database my_db and the table my_tbl are created.

Step 3: Write data to the Paimon table

On the Drafts tab of the Development > ETL page, create a blank stream draft. For more information, see Develop an SQL draft. Copy the following INSERT statement to the SQL editor.

-- Paimon commits data only after each checkpoint is complete.
-- The checkpoint interval is shortened to 10s to commit data faster.
-- In a production environment, typically set the checkpoint interval and the minimum interval between checkpoints from 1 to 10 minutes, based on your latency requirements.
SET 'execution.checkpointing.interval'='10s';
INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108',1,'apple'), ('20240108',2,'banana'), ('20240109',1,'cat'), ('20240109',2,'dog');

In the upper-right corner of the SQL editor, click Deploy. In the Deploy draft dialog box, configure the parameters as prompted and click Confirm.
On the O&M > Deployments page, find the target deployment and click Start in the Actions column. Select Initial Mode and click Start.

When the job status changes to FINISHED, the data has been written.

Step 4: Stream data from the Paimon table

Create a blank stream draft. Copy the following SQL code to the SQL editor to output all data from the my_tbl table to the logs by using the print connector.

CREATE TEMPORARY TABLE Print (
  dt STRING,
  id BIGINT,
  content STRING
) WITH (
  'connector' = 'print'
);
INSERT INTO Print SELECT * FROM `my-catalog`.`my_db`.`my_tbl`;

In the upper-right corner of the SQL editor, click Deploy. In the Deploy draft dialog box, configure the parameters as prompted and click Confirm.
On the O&M > Deployments page, click Start in the Actions column of the target deployment. Select Initial Mode and click Start.
On the deployment details page, view the Flink computation results.
1. On the O&M > Deployments page, click the name of the target deployment.
2. On the Logs tab, under Running Logs, go to the Running Task Managers tab and click a task link in the Path, ID column.
3. Click Stdout to view the consumed Paimon data.
The Stdout output shows the INSERT records for the Paimon table: +I[20240108, 1, apple], +I[20240108, 2, banana], +I[20240109, 1, cat], and +I[20240109, 2, dog], which confirms successful streaming consumption.

Step 5: Update data in the Paimon table

Create a blank stream draft and copy the following SQL code to the SQL editor.

SET 'execution.checkpointing.interval' = '10s';
INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108', 1, 'hello'), ('20240109', 2, 'world');

In the upper-right corner of the SQL editor, click Deploy. In the Deploy draft dialog box, configure the parameters as prompted and click Confirm.
On the O&M > Deployments page, click Start in the Actions column of the target deployment. Select Initial Mode and click Start.

When the job status changes to FINISHED, the data has been updated.
On the Stdout tab of the Deployments page for the job from Step 4, view the data that is updated to the Paimon table.

On the Running Task Managers tab, click the target TaskManager and open the Stdout subtab. The CDC records for the Paimon table appear: -U[20240108, 1, apple], +U[20240108, 1, hello], -U[20240109, 2, dog], and +U[20240109, 2, world]. -U indicates the record before the update, and +U indicates the record after the update.

Step 6 (Optional): Cancel the job and clean up resources

After testing, cancel the streaming consumption job and clean up resources.

On the O&M > Deployments page, click Cancel in the Actions column of the target deployment to stop the job.

In the editor on the Scripts tab, enter the following code to delete the Paimon data files and the Paimon catalog.

DROP DATABASE `my-catalog`.`my_db` CASCADE; -- Deletes all data files for the Paimon database from OSS.
DROP CATALOG `my-catalog`; -- Removes the Paimon catalog from the Realtime Compute for Apache Flink console metadata but does not affect the data files in OSS.

A The following statement has been executed successfully! message indicates that the Paimon data files and catalog are deleted.

Realtime Compute for Apache Flink:Paimon quick start: Basic features