E-MapReduce:Build a unified stream and batch data lake for analytics using Flink, EMR Serverless Spark, and Paimon

Create a Paimon catalog

A Paimon catalog manages all Paimon tables under a single warehouse directory and exposes them to multiple compute engines through DLF. For background on catalogs, see Manage Paimon Catalogs.

1. Log on to the Realtime Compute for Apache Flink console.

2. Click Console in the Actions column of your workspace.

3. In the navigation pane, choose Development > Scripts.

4. Create a query script and paste the following SQL, then replace each placeholder with your actual values:

   Parameter	Description	Notes
   type	Catalog type	Must be paimon
   metastore	Metadata backend	Set to dlf to share metadata across engines
   warehouse	OSS path for table data	For example, oss://my-bucket/paimon-warehouse
   dlf.catalog.id	DLF catalog ID	Find this in the Data Lake Formation console
   dlf.catalog.accessKeyId	AccessKey ID for DLF access
   dlf.catalog.accessKeySecret	AccessKey secret for DLF access
   dlf.catalog.endpoint	DLF service endpoint	Use the VPC endpoint when Flink and DLF are in the same region; otherwise use the public endpoint. See Regions and endpoints
   dlf.catalog.region	Region where DLF is deployed	Must match the region in dlf.catalog.endpoint

   CREATE CATALOG `paimon` WITH (
     'type' = 'paimon',
     'metastore' = 'dlf',
     'warehouse' = '<warehouse>',
     'dlf.catalog.id' = '<dlf.catalog.id>',
     'dlf.catalog.accessKeyId' = '<dlf.catalog.accessKeyId>',
     'dlf.catalog.accessKeySecret' = '<dlf.catalog.accessKeySecret>',
     'dlf.catalog.endpoint' = '<dlf.catalog.endpoint>',
     'dlf.catalog.region' = '<dlf.catalog.region>'
   );

5. In the lower-right corner, click Execution Environment and select a session cluster running VVR 8.0.4 or later. If you don't have one, create one first. See Create a session cluster.

6. Select the SQL statement and click Run.

Create a Paimon table

In the Scripts editor, run the following SQL to create the target table. The write-only = true option optimizes the table for high-throughput streaming writes by disabling read-time merge overhead.

CREATE TABLE IF NOT EXISTS `paimon`.`test_paimon_db`.`test_append_tbl`
(
    id       STRING,
    data     STRING,
    category INT,
    ts       STRING,
    dt       STRING,
    hh       STRING
) PARTITIONED BY (dt, hh)
WITH (
    'write-only' = 'true'
);

Create and start a stream job

1. In the navigation pane, choose Development > ETL.

2. Click New, configure the job in the New Blank Stream Draft dialog, and click Create:

   Parameter	Description
   Name	A unique name within the current project
   Engine version	The Flink engine version. See Engine versions

3. In the job editor, paste the following code. It uses the DataGen connector to generate 100 rows per second with randomized category values (1-10) and inserts them into the Paimon table:

   CREATE TEMPORARY TABLE datagen
   (
       id        STRING,
       data      STRING,
       category  INT
   )
   WITH (
       'connector' = 'datagen',
       'rows-per-second' = '100',
       'fields.category.kind' = 'random',
       'fields.category.min' = '1',
       'fields.category.max' = '10'
   );
   
   INSERT INTO `paimon`.`test_paimon_db`.`test_append_tbl`
   SELECT
       id,
       data,
       category,
       CAST(LOCALTIMESTAMP AS STRING) AS ts,
       CAST(CURRENT_DATE AS STRING) AS dt,
       CAST(HOUR(LOCALTIMESTAMP) AS STRING) AS hh
   FROM datagen;

4. Click Deploy to publish the job to the production environment.

5. On the Deployments page, start the job. See Start a job.

The job is now running and writing data to the Paimon table continuously.