Create an Apache Paimon external table by using Realtime Compute for Apache Flink

Background information

Apache Paimon is a unified streaming and batch lake storage format that provides high-throughput writes and low-latency queries. Common compute engines, such as Spark, Hive, and Trino in Realtime Compute for Apache Flink and E-MapReduce, integrate seamlessly with Paimon. With Apache Paimon, you can quickly build a data lake on Object Storage Service (OSS) and connect it to MaxCompute for data lake analytics. For more information about Apache Paimon, see the Apache Paimon documentation.

Prerequisites

• The Alibaba Cloud account that you use to perform operations has the CreateTable permission to create MaxCompute tables. For more information about table permissions, see MaxCompute permissions.

• A MaxCompute project is created. For more information, see Create a project.

• OSS is activated. A bucket and a file directory are created. For more information, see Create a bucket.

  Note

  MaxCompute is deployed only in specific regions. To prevent a cross-region data connectivity issue, we recommend that you use a bucket in the same region as your MaxCompute project.

• Fully managed Flink is activated. For more information, see Create a workspace.

Step 1: Upload the Apache Paimon plug-in to your MaxCompute project

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

ADD JAR <path_to_paimon_maxcompute_connector.jar>;

Step 2: Create an Apache Paimon external table by using Realtime Compute for Apache Flink

The best practice in this topic is performed based on Realtime Compute for Apache Flink. Realtime Compute for Apache Flink writes data of Apache Paimon files to OSS. An Apache Paimon catalog is created in the Realtime Compute for Apache Flink console, and an Apache Paimon table that can be used by MaxCompute to read the data of Apache Paimon files in OSS is created in the Apache Paimon catalog. Then, MaxCompute uses the Apache Paimon table as an external table to read the Apache Paimon data that is stored in OSS.

1. Log on to the Realtime Compute for Apache Flink console and create a Script. For more information, see Query Scripts.

2. In the script editor, enter the following catalog definition and parameter values. Select the code and click Run.

   CREATE CATALOG `<catalog name>` WITH (
    'type' = 'paimon',
     'metastore' = 'maxcompute',
     'warehouse' = '<warehouse>',
     'maxcompute.endpoint' = '<maxcompute.endpoint>',
     'maxcompute.project' = '<maxcompute.project>',
     'maxcompute.accessid' = '<maxcompute.accessid>',
     'maxcompute.accesskey' = '<maxcompute.accesskey>',
     'maxcompute.oss.endpoint' = '<maxcompute.oss.endpoint>',
     'fs.oss.endpoint' = '<fs.oss.endpoint>',
     'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
     'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
   );

   The following table describes the parameters in the code.

   Parameter	Required	Description
   catalog name	Yes	The name of the Apache Paimon catalog. The name can contain only letters. In this topic, the catalog name is catalogname.
   type	Yes	The type of the catalog. Set the value to paimon.
   metastore	Yes	The type of the metadata storage. Set the value to maxcompute.
   warehouse	Yes	The data warehouse directory in OSS, in the format oss://<bucket>/<object>. bucket: the name of the OSS bucket that you created. object: the path in which your data is stored. You can view your bucket and object names in the OSS console.
   maxcompute.endpoint	Yes	The endpoint of the MaxCompute service. You must configure this parameter based on the region and network connection type that you select when you create the MaxCompute project. For more information about the endpoints that correspond to different regions and network types, see Endpoint.
   maxcompute.project	Yes	The name of the MaxCompute project. MaxCompute projects for which the schema feature is enabled are not supported.
   maxcompute.accessid	Yes	The AccessKey ID of the Alibaba Cloud account or RAM user that has permissions on MaxCompute. You can obtain the AccessKey ID on the AccessKey Management page.
   maxcompute.accesskey	Yes	The AccessKey secret that corresponds to the AccessKey ID.
   maxcompute.oss.endpoint	No	The OSS endpoint that MaxCompute accesses. If you do not configure this parameter, the value of the fs.oss.endpoint parameter is used by default. Important The OSS bucket resides in the same region as the MaxCompute project. We recommend that you set the maxcompute.oss.endpoint parameter to an internal endpoint. For more information about the OSS endpoints of different network types in each region, see Regions and Endpoints.
   fs.oss.endpoint	No	The endpoint of OSS. This parameter is required if the OSS bucket specified by the warehouse parameter is not in the same region as the Realtime Compute for Apache Flink workspace or an OSS bucket within another Alibaba Cloud account is used. Note You must configure the endpoint based on the region and network connection method that you select when you create the OSS bucket. For more information about the endpoints that correspond to different regions and network types, see Regions and Endpoints.
   fs.oss.accessKeyId	No	The AccessKey ID of the Alibaba Cloud account or RAM user that has read and write permissions on OSS. This parameter is required if the OSS bucket specified by the warehouse parameter is not in the same region as the Realtime Compute for Apache Flink workspace or an OSS bucket within another Alibaba Cloud account is used. You can obtain the AccessKey ID on the AccessKey Management page.
   fs.oss.accessKeySecret	No	The AccessKey secret that corresponds to the AccessKey ID. This parameter is required if the OSS bucket specified by the warehouse parameter is not in the same region as the Realtime Compute for Apache Flink workspace or an OSS bucket within another Alibaba Cloud account is used.

3. Create an Apache Paimon table.

   1. Create a table named test_tbl.

      On the Script tab, run the following statement in the script editor. Wait for a completion message to appear on the Result tab. This example uses 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);

   2. Write data to the table test_tbl.

      On the Drafts page, create a SQL job that contains the following statements. Then, deploy and run the job. For more information about how to create and run SQL jobs, see Job development overview.

      -- The checkpoint interval is set to 10s to commit data faster.
      SET 'execution.checkpointing.interval' = '10s';
      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

      • The Apache Paimon result table commits data each time checkpointing is complete.

      • In the production environment, the checkpointing interval and the minimal interval between checkpoints vary based on your business requirements for latency. In most cases, they are set to 1 to 10 minutes.

      • The engine version of the SQL draft must be vvr-8.0.5-flink-1.17 or later.

Step 3: Read data with MaxCompute

1. Run the following commands on the Connect using the local client (odpscmd) or by using another tool that can execute MaxCompute SQL statements:

   SET odps.sql.common.table.planner.ext.hive.bridge = true;
   SET odps.sql.hive.compatible = true;

2. Run the following command to query data from the Apache Paimon external table test_tbl:

   SELECT * FROM test_tbl WHERE dt = '2023-04-21';

   The following result is returned:

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