Data Lake Formation:Build a streaming data lakehouse with DLF and Flink

Create a Paimon catalog

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

2. In the left navigation pane, choose Metadata Management and click Create Catalog.

3. On the Built-in Catalog tab, click Apache Paimon and then click Next.

4. Configure the following parameters, select DLF as the storage class, and click OK.

   Configuration item	Description	Required	Remarks
   metastore	The metastore type.	Yes	Use dlf.
   catalog name	The name of the DLF data catalog.	Yes	Select an existing catalog. To create one, see Data catalog. This example uses a catalog named paimoncatalog.

5. Create the order_dw database in the catalog. In the left navigation pane, choose Data Query > Query Script and click New to open a temporary query. Run:

   -- Switch to the paimoncatalog data source
   USE CATALOG paimoncatalog;
   -- Create the target database
   CREATE DATABASE order_dw;

   The following statement has been executed successfully! confirms the database is created.

For more information, see Manage a Paimon catalog.

Create a MySQL catalog

1. On the Metadata Management page, click Create Catalog.

2. On the Built-in Catalog tab, click MySQL and then click Next.

3. Enter the following parameters and click OK. This creates a catalog named mysqlcatalog.

   Configuration item	Description	Required	Remarks
   catalog name	The catalog name.	Yes	This example uses mysqlcatalog.
   hostname	The IP address or hostname of the MySQL instance.	Yes	Use the internal network address if the RDS instance and Flink workspace share the same VPC. See View and manage endpoints and port numbers.
   port	The MySQL service port. Default: 3306.	No	See View and manage endpoints and port numbers.
   default-database	The default MySQL database.	Yes	Use order_dw.
   username	The MySQL account username.	Yes	The account created in Step 1.
   password	The MySQL account password.	Yes	The password for the account created in Step 1.

Create the wide table

In the Flink Development Console, choose Development > Scripts. Click + > New Script. Run the following DDL:

CREATE TABLE paimoncatalog.order_dw.dwd_orders (
    order_id BIGINT,
    order_user_id STRING,
    order_shop_id BIGINT,
    order_product_id BIGINT,
    order_product_catalog_name STRING,
    order_fee BIGINT,
    order_create_time TIMESTAMP,
    order_update_time TIMESTAMP,
    order_state INT,
    pay_id BIGINT,
    pay_platform INT COMMENT 'platform 0: phone, 1: pc',
    pay_create_time TIMESTAMP,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'merge-engine' = 'partial-update', -- Merge partial records from multiple source streams into one wide row
    'changelog-producer' = 'lookup'    -- Generate low-latency changelogs for downstream consumption
);

Query has been executed confirms the table is created.

Start the DWD streaming job

In the left navigation pane, choose Development > ETL. Click + > New Blank Stream Draft, name the draft dwd, and paste the following SQL. Deploy the draft and start the job deployment with initial states.

The job uses UNION ALL to run two streams concurrently: it joins orders with product_catalog as a dimension lookup, and independently streams orders_pay. Both streams write to dwd_orders, and Paimon's partial-update merge engine assembles the complete wide row when records with the same order_id arrive.

SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
SET 'table.exec.sink.upsert-materialize' = 'NONE';
SET 'execution.checkpointing.interval' = '10s';
SET 'execution.checkpointing.min-pause' = '10s';

-- Paimon does not allow multiple INSERT statements targeting the same table
-- in the same deployment. Use UNION ALL instead.
INSERT INTO paimoncatalog.order_dw.dwd_orders
SELECT
    o.order_id,
    o.user_id,
    o.shop_id,
    o.product_id,
    dim.catalog_name,
    o.buy_fee,
    o.create_time,
    o.update_time,
    o.state,
    NULL,  -- pay_id (populated by orders_pay stream)
    NULL,  -- pay_platform
    NULL   -- pay_create_time
FROM
    paimoncatalog.order_dw.orders o
    LEFT JOIN paimoncatalog.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dim
    ON o.product_id = dim.product_id
UNION ALL
SELECT
    order_id,
    NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL,  -- order columns (populated by orders stream)
    pay_id,
    pay_platform,
    create_time
FROM
    paimoncatalog.order_dw.orders_pay;

Verify the DWD layer: In the left navigation pane, choose Development > Scripts. Click + > New Script. Run:

SELECT * FROM paimoncatalog.order_dw.dwd_orders ORDER BY order_id;

Create the DWS and DWM tables

In the left navigation pane, choose Development > Scripts. Click + > New Script. Run the following DDL:

-- User-dimension aggregate metric table
CREATE TABLE paimoncatalog.order_dw.dws_users (
    user_id STRING,
    ds STRING,
    payed_buy_fee_sum BIGINT COMMENT 'Total payment amount completed on the current day',
    PRIMARY KEY (user_id, ds) NOT ENFORCED
) WITH (
    'merge-engine' = 'aggregation',
    'fields.payed_buy_fee_sum.aggregate-function' = 'sum'
    -- No changelog-producer needed: dws_users is queried by StarRocks, not consumed as a stream
);

-- Shop-dimension aggregate metric table
CREATE TABLE paimoncatalog.order_dw.dws_shops (
    shop_id BIGINT,
    ds STRING,
    payed_buy_fee_sum BIGINT COMMENT 'Total payment amount completed on the current day',
    uv BIGINT COMMENT 'Total number of users that made purchases on the current day',
    pv BIGINT COMMENT 'Total number of purchases made on the current day',
    PRIMARY KEY (shop_id, ds) NOT ENFORCED
) WITH (
    'merge-engine' = 'aggregation',
    'fields.payed_buy_fee_sum.aggregate-function' = 'sum',
    'fields.uv.aggregate-function' = 'sum',
    'fields.pv.aggregate-function' = 'sum'
    -- No changelog-producer needed: dws_shops is queried by StarRocks, not consumed as a stream
);

-- Intermediate DWM table: aggregates by (user_id, shop_id) and emits changelogs to both DWS tables
CREATE TABLE paimoncatalog.order_dw.dwm_users_shops (
    user_id STRING,
    shop_id BIGINT,
    ds STRING,
    payed_buy_fee_sum BIGINT COMMENT 'Total amount paid by this user in this shop on the current day',
    pv BIGINT COMMENT 'Number of purchases made by this user in this shop on the current day',
    PRIMARY KEY (user_id, shop_id, ds) NOT ENFORCED
) WITH (
    'merge-engine' = 'aggregation',
    'fields.payed_buy_fee_sum.aggregate-function' = 'sum',
    'fields.pv.aggregate-function' = 'sum',
    'changelog-producer' = 'lookup',  -- Emit changelogs so dws_users and dws_shops can consume incrementally
    'file.format' = 'avro',           -- Row-oriented format for better write throughput on the intermediate table
    'metadata.stats-mode' = 'none'    -- Skip statistical metadata to reduce write cost (no impact on stream processing)
);

Query has been executed confirms the tables are created.

Populate the DWM table from DWD changelogs

In the left navigation pane, choose Development > ETL. Click + > New Blank Stream Draft, name the draft dwm, and paste the following SQL. Deploy and start with initial states.

This job reads from dwd_orders, filters out records without payment data, and accumulates per-user-per-shop metrics into dwm_users_shops.

SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
SET 'table.exec.sink.upsert-materialize' = 'NONE';
SET 'execution.checkpointing.interval' = '10s';
SET 'execution.checkpointing.min-pause' = '10s';

INSERT INTO paimoncatalog.order_dw.dwm_users_shops
SELECT
    order_user_id,
    order_shop_id,
    DATE_FORMAT(pay_create_time, 'yyyyMMdd') AS ds,
    order_fee,
    1  -- Each input record counts as one purchase
FROM paimoncatalog.order_dw.dwd_orders
WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL;

Populate the DWS tables from DWM changelogs

In the left navigation pane, choose Development > ETL. Click + > New Blank Stream Draft, name the draft dws, and paste the following SQL. Deploy and start with initial states.

Unlike the DWD layer, multiple INSERT statements targeting different Paimon tables are allowed in the same deployment. The dws_shops insert uses a local-merge-buffer-size hint to pre-aggregate records in memory before writing to Paimon, which reduces data skew for high-traffic shops.

SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
SET 'table.exec.sink.upsert-materialize' = 'NONE';
SET 'execution.checkpointing.interval' = '10s';
SET 'execution.checkpointing.min-pause' = '10s';

BEGIN STATEMENT SET;

INSERT INTO paimoncatalog.order_dw.dws_users
SELECT
    user_id,
    ds,
    payed_buy_fee_sum
FROM paimoncatalog.order_dw.dwm_users_shops;

-- Use local merge to pre-aggregate in memory before writing to Paimon.
-- This reduces data skew for popular shops that receive disproportionately high write volumes.
INSERT INTO paimoncatalog.order_dw.dws_shops /*+ OPTIONS('local-merge-buffer-size' = '64mb') */
SELECT
    shop_id,
    ds,
    payed_buy_fee_sum,
    1,   -- Each input record represents one unique user-shop combination
    pv
FROM paimoncatalog.order_dw.dwm_users_shops;

END;

Verify the DWS layer: In the left navigation pane, choose Development > Scripts. Click + > New Script. Run:

-- Check user aggregate metrics
SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;

-- Check shop aggregate metrics
SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;

Ranking query

Find the top three shops by transaction volume on February 15, 2023:

SELECT ROW_NUMBER() OVER (ORDER BY payed_buy_fee_sum DESC) AS rn, shop_id, payed_buy_fee_sum
FROM paimoncatalog.order_dw.dws_shops
WHERE ds = '20230215'
ORDER BY rn LIMIT 3;

Details query

Retrieve all orders placed by user_001 via mobile (pay_platform = 0) in February 2023:

SELECT * FROM paimoncatalog.order_dw.dwd_orders
WHERE order_create_time >= '2023-02-01 00:00:00' AND order_create_time < '2023-03-01 00:00:00'
AND order_user_id = 'user_001'
AND pay_platform = 0
ORDER BY order_create_time;

Report query

Get the total order count and revenue by product category for February 2023:

SELECT
  order_create_time AS order_create_date,
  order_product_catalog_name,
  COUNT(*),
  SUM(order_fee)
FROM
  paimoncatalog.order_dw.dwd_orders
WHERE
  order_create_time >= '2023-02-01 00:00:00' AND order_create_time < '2023-03-01 00:00:00'
GROUP BY
  order_create_date, order_product_catalog_name
ORDER BY
  order_create_date, order_product_catalog_name;