Transform Real‑Time Data Warehousing with Paimon: From Flink ROW_NUMBER to Streaming Lakehouse

Background

Real‑time business needs include providing performance metrics for sprinting and delivering the number of new/renewed members for the current month.

The original architecture used Flink, Kafka, HBase and MySQL. Data flowed from ODS to APP, undergoing custom transformations, being written to Kafka, and then processed with Flink SQL before landing in MySQL for downstream consumption.

Problems of the Original Architecture

Frequent requirement changes : each change required component maintenance, hurting flexibility.

High development cost : many components (Kafka, HBase, Redis, MySQL) and state loss on logic changes.

Resource waste : duplicate storage in MySQL and Hive, intermediate results stored in MySQL for debugging.

Inconsistent data definitions : missing updates to data definitions caused errors.

High O&M cost : troubleshooting data issues consumed much time.

Building a Streaming Lakehouse with Paimon

What Is Paimon?

Real‑time updates via LSM‑tree, supporting primary‑key tables with deduplication and changelog generation.

Unified batch and stream processing, compatible with Flink, Spark, Hive.

OLAP optimizations: columnar ORC storage, Z‑order/Hilbert sorting, min‑max indexes.

Lakehouse features: ACID transactions, time travel, scalable metadata, schema evolution.

Branch management: lock‑free data branches, fast‑forward sync.

Low cost and ecosystem compatibility: object storage (OSS/S3) and seamless integration with major compute engines.

Use Cases and Implementation

Deduplication

Original solution used Flink SQL ROW_NUMBER() window, which caused heavy shuffle, high memory consumption, large checkpoint files, and complex SQL.

New solution defines a Paimon primary‑key table with 'merge-engine'='deduplicate'. Flink writes directly to the table; the merge engine keeps the latest record, achieving the same effect without extra SQL.

CREATE TABLE paimon_table (
    v_date STRING,
    id STRING,
    ...,
    event_time TIMESTAMP,
    PRIMARY KEY (v_date, id) NOT ENFORCED
) PARTITIONED BY (v_date)
WITH (
    'merge-engine'='deduplicate',
    'bucket'='4',
    'deletion-vectors.enabled'='true',
    'async-file-write'='true',
    'sort-spill-threshold'='10',
    'lookup-wait'='false',
    'snapshot.num-retained.max'='10000'
);

Flink insert:

INSERT INTO paimon_table SELECT ... FROM source_table;

Out‑of‑Order Handling

Previous approach cached the latest timestamp in Redis per key, leading to network I/O and checkpoint issues.

Paimon’s sequence.field='event_time' automatically keeps the newest record per primary key, discarding older out‑of‑order events.

CREATE TABLE paimon_table (
    v_date STRING,
    id STRING,
    ...,
    event_time TIMESTAMP,
    PRIMARY KEY (v_date, id) NOT ENFORCED
) PARTITIONED BY (v_date)
WITH (
    'merge-engine'='deduplicate',
    'bucket'='4',
    'sequence.field'='event_time',
    'deletion-vectors.enabled'='true',
    'async-file-write'='true',
    'sort-spill-threshold'='10',
    'lookup-wait'='false',
    'snapshot.num-retained.max'='10000'
);

Streaming Read

Instead of Kafka, Flink can consume the changelog stream of a Paimon table directly.

CREATE VIEW view_paimon_changelog AS
SELECT * FROM paimon_ods_table;
-- downstream jobs consume view_paimon_changelog

Aggregation

Original cumulative windows caused performance bottlenecks and large state.

Paimon aggregation tables define primary key plus aggregate functions, e.g., sum for score and last_non_null for update_time. The engine maintains pre‑aggregated results and emits changelog streams.

CREATE TABLE agg_table (
    v_date STRING,
    agent_id STRING,
    score DECIMAL(20,4),
    update_time TIMESTAMP(3),
    PRIMARY KEY (v_date, agent_id) NOT ENFORCED
) PARTITIONED BY (v_date)
WITH (
    'merge-engine'='aggregation',
    'fields.score.aggregate-function'='sum',
    'fields.update_time.aggregate-function'='last_non_null'
);

Lookup Join

Paimon provides three lookup strategies:

FullCacheLookupTable : loads the whole table into local RocksDB; high lookup speed but large disk usage.

LocalPrimaryKeyPartialLookupTable : loads only needed partitions based on incoming keys; lower disk usage but may suffer cache fragmentation.

BucketShuffleLookupTable : adds a custom shuffle so each task loads a single bucket; balances speed and resource consumption.

Example lookup join:

SELECT
    t1.id,
    t1.product_id,
    t2.name AS product_name,
    t2.category,
    t2.category_name
FROM (
    SELECT id, product_id, PROCTIME() AS proc_time FROM source_table
) AS t1
LEFT JOIN dim_order_info FOR SYSTEM_TIME AS OF t1.proc_time AS prd_info
ON t1.product_id = prd_info.id;

Batch Repair

Instead of replaying ODS data through Kafka, Paimon’s primary‑key tables allow direct batch updates via Spark or Flink, achieving exactly‑once semantics and drastically reducing repair time.

-- Spark SQL example
INSERT INTO paimon_table
SELECT ... FROM raw_table WHERE v_date BETWEEN '2024-01-01' AND '2024-01-01';

Benefits Summary

Significant resource efficiency: CPU and memory usage drop, checkpoint size shrinks, task recovery speeds up.

Simplified development: deduplication, out‑of‑order handling, and aggregation are expressed by table metadata instead of complex SQL.

Improved data consistency: primary‑key guarantees atomic updates and exactly‑once semantics.

Unified data platform: replaces multiple systems (Kafka, MySQL, HBase, Redis) with a single lakehouse storage.

Scalable for various scenarios such as real‑time dashboards, metric monitoring, and batch repair.