Ingest data to DLF with Flink DataStream APIs - Data Lake Formation

This topic describes how to develop a Realtime Compute for Apache Flink job that uses the DataStream API to write data to a Data Lake Formation (DLF) catalog via Paimon REST.

Prerequisites

You have a fully managed workspace for Realtime Compute for Apache Flink. If you have not created one, see Activate Realtime Compute for Apache Flink.
A DLF catalog is created. For more information, see Get started with DLF.
Your Flink workspace and DLF catalog reside in the same region.
The VPC of your Flink workspace is in DLF's VPC whitelist. For more information, see API usage guide.
Note
DLF enables VPC access by default. To enable public network access, see Enable and configure public network access.

Preparations

Download the Paimon bundled JAR paimon-flink-*.jar of version 1.1 or later from the Apache Paimon website.
Download paimon-oss-*.jar of version 1.1 or later from Apache Paimon Filesystems.

Choose a dependency method

Important

The Flink runtime environment does not include the Paimon connector or the OSS file system. You must use one of the following methods to make sure that paimon-flink-*.jar and paimon-oss-*.jar are available at job runtime.

Method 1: Upload additional files in the console

You do not need to modify the pom.xml file. When you create a JAR job in the Realtime Compute for Apache Flink development console, upload the paimon-flink-*.jar and paimon-oss-*.jar files that you downloaded in the Preparations section as additional dependency files.

Method 2: Package dependencies into a fat JAR with Maven

Add the following dependencies and properties to your project's pom.xml file.

<properties>
    <!-- Paimon version. Specify 1.1 or later. -->
    <paimon.version>1.1.0</paimon.version>
    <!-- Flink major version. Set based on your VVR version. See the following table. -->
    <flink.main.version>1.20</flink.main.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.paimon</groupId>
        <artifactId>paimon-flink-${flink.main.version}</artifactId>
        <version>${paimon.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.paimon</groupId>
        <artifactId>paimon-oss</artifactId>
        <version>${paimon.version}</version>
    </dependency>
</dependencies>

The value of ${flink.main.version} is as follows.

VVR version	`flink.main.version`
VVR 8.x	1.17
VVR 11.x	1.20

With this method, the dependencies are packaged into the fat JAR, so you do not need to upload additional JAR files during deployment.

Step 1: Write the job code

In the main() method of your DataStream job, use the following code to create a DLF catalog instance.

Options options = new Options();
options.set("type", "paimon");
options.set("metastore", "rest");
options.set("uri", "http://<region-id>-vpc.dlf.aliyuncs.com");
options.set("warehouse", "your-catalog-name");
options.set("token.provider", "dlf");
options.set("dlf.access-key-id", "your-access-key-id");
options.set("dlf.access-key-secret", "your-access-key-secret");
Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(options);

Required parameters:

Parameter	Description	Example
`type`	The catalog type, automatically parsed from the custom JAR. Do not change this value.	`paimon`
`metastore`	The metastore type for DLF. Set this to `rest`.	`rest`
`uri`	The VPC endpoint of the DLF REST catalog server. The format is `http://[region-id]-vpc.dlf.aliyuncs.com`. For more information, see Regions and endpoints.	`http://ap-southeast-1-vpc.dlf.aliyuncs.com`
`warehouse`	The Paimon catalog name.	`dlf_test`
`token.provider`	The token provider. Set this to `dlf`.	`dlf`
`dlf.access-key-id`	Your AccessKey ID for authentication. For more information, see View RAM user AccessKey information.
`dlf.access-key-secret`	Your AccessKey secret for authentication.

After the catalog is created, you can register and use it in your DataStream job to read and write Paimon tables.

Step 2: Package and deploy the job

Package your DataStream job into a JAR file.
Upload the job JAR in the Realtime Compute for Apache Flink console and submit the job.
If you chose Method 1 (uploading additional files in the console), add paimon-flink-*.jar and paimon-oss-*.jar to the additional dependencies when you submit the job.

For more information about how to develop and debug Flink JAR jobs, see Develop a JAR job.