Realtime Compute for Apache Flink:Stream storage Fluss

Overview

Fluss primary key tables support full and incremental consumption. The system first reads a full historical snapshot of the table. After the snapshot read is complete, it automatically and seamlessly switches to incremental stream consumption. This process ensures data completeness and consistency.

Consumption modes

You can control the consumption starting point using the scan.startup.mode parameter.

SQL examples

-- Full and incremental consumption (default mode)
SELECT * FROM `fluss_catalog`.`fluss_db`.orders;

-- Consume from a specific timestamp by using a hint
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
  /*+ OPTIONS('scan.startup.mode' = 'timestamp', 'scan.startup.timestamp' = '2025-01-01 00:00:00') */;

-- Consume from the latest offset
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
  /*+ OPTIONS('scan.startup.mode' = 'latest') */;

Log subscription

Fluss primary key tables natively support CDC (Change Data Capture) log subscription. This allows downstream Flink jobs to consume a complete changelog stream (INSERT, UPDATE_BEFORE, UPDATE_AFTER, and DELETE events) in real time without requiring a separate binlog configuration.

This means:

• All changes to a primary key table automatically generate CDC logs.

• Downstream applications can directly subscribe to the primary key table to receive the complete changelog stream for real-time data synchronization or streaming ETL.

SQL example

-- Read the primary key table directly to get the CDC changelog stream
SELECT * FROM `fluss_catalog`.`fluss_db`.orders;
-- The output includes complete INSERT, UPDATE, and DELETE events.

Virtual tables

In addition to subscribing to the CDC stream directly, Fluss provides a virtual table mechanism. You can access changelogs with metadata by appending a specific suffix to the base table name, which eliminates the need for extra data storage. The system generates these virtual tables automatically, eliminating the need for manual creation.

-- Read the $changelog virtual table for the orders table
SELECT * FROM `fluss_catalog`.`fluss_db`.orders$changelog;

-- Consume from the latest offset by using a hint
SELECT * FROM `fluss_catalog`.`fluss_db`.orders$changelog
  /*+ OPTIONS('scan.startup.mode' = 'latest') */;

-- Filter for delete events
SELECT * FROM `fluss_catalog`.`fluss_db`.orders$changelog
WHERE _change_type = 'delete';

-- Read the $binlog virtual table for the users table
SELECT * FROM `fluss_catalog`.`fluss_db`.users$binlog;

-- Get the before and after values of a field from an update event
SELECT
  `before`.user_name AS old_name,
  `after`.user_name  AS new_name,
  _commit_timestamp  AS change_time
FROM `fluss_catalog`.`fluss_db`.users$binlog
WHERE _change_type = 'update';

Overview

Fluss supports partial updates, which allow multiple data streams to independently write to their respective columns in the same wide primary key table. The Fluss engine automatically merges fields at the record level based on the primary key. This approach avoids the need to maintain complex multi-stream join state in Flink and prevents issues related to large state sizes.

Use case

A typical use case is building real-time user profile wide tables. Different data sources, such as recommendation systems, impression logs, click logs, and shopping cart data, can update their respective fields in the user profile table. This process creates a complete 360-degree view of the user.

Limitations

• The target table must be a primary key table.

• Each data stream's write operation must include all primary key fields.

• The computation topology must not contain multi-stream join nodes. Fluss handles the merge operation directly at the storage layer.

SQL examples

-- 1. Create a wide table
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.user_rec_wide (
    user_id   STRING,
    item_id   STRING,
    rec_score DOUBLE,
    imp_cnt   INT,
    click_cnt INT,
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');

-- 2. Use multiple streams to write to their respective fields
-- Stream 1: Writes recommendation scores
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, rec_score)
SELECT user_id, item_id, rec_score 
FROM `fluss_catalog`.fluss_db.recommendations;

-- Stream 2: Writes impression counts
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, imp_cnt)
SELECT user_id, item_id, imp_cnt 
FROM `fluss_catalog`.fluss_db.impressions;

-- Stream 3: Writes click counts
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, click_cnt)
SELECT user_id, item_id, click_cnt 
FROM `fluss_catalog`.fluss_db.clicks;

The three INSERT statements above can run as separate Flink jobs. The Fluss storage layer automatically merges the fields for the same primary key into complete rows.

Overview

Delta Join is a next-generation stream-stream join paradigm provided by Fluss. It offloads data update operations to the Fluss storage layer, leveraging primary key tables and secondary indexes for efficient lookups. This design eliminates the need to maintain join state in Flink.

Key advantages:

• No join state: Eliminates redundant state storage and avoids garbage collection (GC) issues caused by large state sizes.

• Low resource consumption: Reduces Flink memory and CPU consumption by over 86% in tests.

• Greater stability and efficiency: Avoids performance bottlenecks caused by state bloat.

Comparison with traditional joins

Limitations

• Both the left and right tables must be Fluss primary key tables (partitioned tables are supported).

• The bucket key must be a prefix of the primary key.

• The join key must include the bucket key. For a partitioned table, it must also include the partition key.

SQL examples

1. Create tables

CREATE TABLE orders (
  user_id BIGINT,
  order_date DATE,
  order_id BIGINT,
  amount DECIMAL(10, 2),
  PRIMARY KEY (user_id, order_id, order_date) NOT ENFORCED
) WITH ('bucket.key' = 'user_id, order_id');

CREATE TABLE order_enhance (
  user_id BIGINT,
  order_id BIGINT,
  risk_level TINYINT,
  is_fraud BOOLEAN,
  update_time TIMESTAMP(3),
  PRIMARY KEY (user_id, order_id) NOT ENFORCED
) WITH ('bucket.key' = 'user_id, order_id');

2. Run a Delta Join

SELECT * FROM orders o
INNER JOIN order_enhance e
ON o.user_id = e.user_id AND o.order_id = e.order_id;

3. Optimize lookup performance with hints

SELECT * FROM orders o /*+ OPTIONS('client.lookup.queue-size'='2560') */
INNER JOIN order_enhance e /*+ OPTIONS('client.lookup.queue-size'='2560') */
ON o.user_id = e.user_id AND o.order_id = e.order_id;

Job parameter configuration

To use a Delta Join, add the following configurations to the runtime parameters of your Flink job:

table.exec.delta-join.cache-enabled: 'true'
table.exec.async-lookup.buffer-capacity: '3000'
table.optimizer.delta-join.strategy: FORCE
table.exec.delta-join.left.cache-size: '100'
table.exec.delta-join.right.cache-size: '1000'

After you enable Delta Join, the Delta Join node appears in the Flink job overview, indicating the optimization is active.

Overview

You can use a Fluss primary key table as a Flink SQL dimension table. A lookup join allows you to enrich streaming data with dimensional data in real time. Fluss supports efficient KV point lookups and secondary index searches, making it suitable for low-latency dimension table join scenarios.

Limitations

• Only a primary key table can be used as a dimension table.

• The join condition must include all primary key fields of the dimension table.

Synchronous and asynchronous modes

SQL examples

-- Configure lookup cache by using a hint to perform a dimension table join
SELECT
  o.order_id,
  o.amount,
  u.user_name,
  p.product_name
FROM `fluss_catalog`.`fluss_db`.orders AS o
LEFT JOIN `fluss_catalog`.`fluss_db`.user_dim
  /*+ OPTIONS('lookup.cache' = 'PARTIAL', 'lookup.partial-cache.expire-after-write' = '60s', 'lookup.partial-cache.max-rows' = '10000') */
  FOR SYSTEM_TIME AS OF o.proctime AS u
  ON o.user_id = u.user_id
LEFT JOIN `fluss_catalog`.`fluss_db`.product_dim
  FOR SYSTEM_TIME AS OF o.proctime AS p
  ON o.product_id = p.product_id;

-- Use synchronous lookup mode (suitable for debugging)
SELECT o.order_id, o.amount, d.user_name
FROM `fluss_catalog`.`fluss_db`.orders AS o
LEFT JOIN `fluss_catalog`.`fluss_db`.user_dim
  /*+ OPTIONS('lookup.async' = 'false') */
  FOR SYSTEM_TIME AS OF o.proctime AS d
  ON o.user_id = d.user_id;

Overview

Fluss primary key tables let you configure a merge engine to control how the system merges records with the same primary key. Use the table.merge-engine parameter to select one of the following strategies:

SQL examples

-- Use first_row for deduplication
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.event_dedup (
    event_id STRING,
    event_type STRING,
    event_time TIMESTAMP(3),
    payload STRING,
    PRIMARY KEY (event_id) NOT ENFORCED
) WITH (
  'table.merge-engine' = 'first_row'
);

-- Use versioned to update based on a version column
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.user_profile (
    user_id BIGINT,
    user_name STRING,
    age INT,
    update_version BIGINT,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'table.merge-engine' = 'versioned',
  'table.merge-engine.versioned.ver-column' = 'update_version'
);

Overview

Fluss supports partitioned tables. You can use PARTITIONED BY to divide data into different physical partitions based on a partition key. This is ideal for managing data lifecycles and improving query performance based on time or business dimensions. Partitioned tables support single-field partitioning and multi-field partitioning, and offer three orthogonal partition management strategies: manual management, automatic management, and dynamic creation.

Basic rules

• The data type of a partition key must be STRING.

• For a primary key table, the partition key must be a subset of the primary key.

• You can only configure automatic partitioning rules when creating the table; you cannot modify them later.

Partition management strategies

These three strategies are orthogonal and can be used in combination. For example, you can enable both automatic management for time-based creation and cleanup, and dynamic creation as a fallback to handle partitions that were not pre-created.

SQL examples

1. Create a single-field partitioned table with automatic partition management

CREATE TABLE `fluss_catalog`.fluss_db.site_access (
  site_id BIGINT,
  user_name STRING,
  pv BIGINT,
  event_day STRING,
  PRIMARY KEY (site_id, event_day) NOT ENFORCED
) PARTITIONED BY (event_day) WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'DAY',
  'table.auto-partition.num-precreate' = '5',
  'table.auto-partition.num-retention' = '30',
  'table.auto-partition.time-zone' = 'Asia/Shanghai'
);

2. Create a multi-field partitioned table

CREATE TABLE `fluss_catalog`.fluss_db.region_access (
  access_id BIGINT,
  user_name STRING,
  event_date STRING,
  region STRING,
  PRIMARY KEY (access_id, event_date, region) NOT ENFORCED
) PARTITIONED BY (event_date, region) WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.key' = 'event_date',
  'table.auto-partition.time-unit' = 'DAY',
  'table.auto-partition.num-retention' = '7'
);

Note: For multi-field partitioning, you must specify the time-based key for automatic partitioning using table.auto-partition.key. Multi-field partitioning supports only automatic expiration and cleanup, not automatic pre-creation. The num-precreate parameter is forced to 0.

3. Manually manage partitions

-- View all partitions
SHOW PARTITIONS `fluss_catalog`.fluss_db.site_access;

-- Manually add a partition
ALTER TABLE `fluss_catalog`.fluss_db.site_access
  ADD PARTITION (event_day = '20250315');

-- Manually drop a partition
ALTER TABLE `fluss_catalog`.fluss_db.site_access
  DROP PARTITION (event_day = '20250101');

4. Dynamically create partitions

When you write data, if the target partition does not exist, the client automatically creates it. This feature is enabled by default.

-- Write data. If the partition '20250701' does not exist, it is automatically created.
INSERT INTO `fluss_catalog`.fluss_db.site_access
VALUES (1, 'hello', 100, '20250701');

The client.writer.dynamic-create-partition.enabled client parameter (default: true) controls dynamic creation. It is subject to the cluster-level limits of max.partition.num (default: 1000) and max.bucket.num (default: 128000).

Partition pruning

When you query a partitioned table in streaming or batch mode, Fluss supports partition pruning. By filtering on the partition key in a WHERE clause, the query reads only the matching partitions, which reduces unnecessary I/O overhead.

-- Read only the data from the partition for a specific date
SELECT * FROM `fluss_catalog`.`fluss_db`.site_access
WHERE event_day = '20250315';

-- Filter a multi-field partitioned table by region
SELECT * FROM `fluss_catalog`.`fluss_db`.region_access
WHERE region = 'US';

Automatic partitioning parameters

Best practices

• Choose the right partition granularity: Select a time granularity based on your data volume and query patterns. For large data volumes and narrow query time ranges, we recommend partitioning by DAY or HOUR. For smaller data volumes, you can partition by MONTH or QUARTER.

• Set a reasonable retention policy: Use num-retention to control the number of historical partitions. This helps avoid metadata bloat and reduces the cluster management burden from an excessive number of partitions.

• Combine automatic management and dynamic creation: Use automatic management for routine time-based partition creation and cleanup, and use dynamic creation as a fallback to handle abnormal or delayed data.

• Be aware of cluster limits: The max.partition.num parameter (default: 1000) limits the number of partitions per table. The total number of buckets in the cluster is limited by max.bucket.num (default: 128,000). The total number of buckets in a table (number of partitions × bucket.num) must not exceed this limit.

Fluss uses the Apache Arrow columnar storage format for log transmission. This supports pushing down filter and projection operations to the storage layer, which reduces data transfer volume and can improve query performance by up to 10x.

This feature is especially effective in the following scenarios:

• Reading only a subset of columns from a wide table (projection pushdown).

• Filtering a large amount of data based on specific conditions (predicate pushdown).

The log storage format is controlled by the table.log.format parameter, which defaults to ARROW.

Overview

Fluss supports a unified stream-lake architecture that can automatically synchronize real-time data from Fluss to a data lake, such as Apache Paimon. This achieves automatic tiered storage for hot and cold data, balancing the low latency of real-time queries with the low storage costs of historical data.

Key features:

• Automatically writes streaming data from Fluss to Paimon lake storage.

• Automatically tiers hot and cold data. Hot data is retained in Fluss stream storage, while cold data is moved to Paimon.

• Provides a unified Flink SQL query interface to access data in both stream storage and lake storage simultaneously.

Configuration parameters

Overview

Fluss supports quick data exploration on tables in the Flink SQL batch mode, including LIMIT queries, primary key point lookups, and COUNT(*) aggregations. You can quickly view data in a table without starting a continuously running streaming job. This is useful for development, debugging, data validation, and troubleshooting.

LIMIT queries

You can run LIMIT queries on primary key tables and append-only tables to quickly retrieve a small sample of data.

-- View the first 10 records from a primary key table
SELECT * FROM `fluss_catalog`.`fluss_db`.orders LIMIT 10;

-- View the first 5 records from an append-only table
SELECT * FROM `fluss_catalog`.`fluss_db`.events LIMIT 5;

-- Combine with projection to view only specific columns
SELECT order_id, amount, order_time 
FROM `fluss_catalog`.`fluss_db`.orders LIMIT 20;

Primary key point lookups

For primary key tables, you can perform precise point lookups by specifying the primary key in a WHERE clause. This leverages the Fluss KV storage engine to achieve millisecond-level response times.

-- Perform a point lookup for a single record by its primary key
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
WHERE order_id = 1001;

-- Perform a point lookup with a composite primary key
SELECT * FROM `fluss_catalog`.`fluss_db`.user_orders
WHERE user_id = 12345 AND order_id = 6789;

COUNT(*) aggregations

You can run a COUNT(*) query on a primary key table to quickly get the total number of records.

-- Count the total number of rows in a primary key table
SELECT COUNT(*) FROM `fluss_catalog`.`fluss_db`.orders;