大数据领域隔一阵就会出现一个新名词：流批一体、湖仓一体、HSAP……本质上都是在抹平界限。但在我经历过的项目里，真正落地的并不多，核心原因是很难在不牺牲原有能力的前提下解决核心痛点。最近体验了结合 Fluss 的湖流一体，感觉它的特性确实有可能重塑实时开发，这篇文章记录一下初步的体验。

传统架构：实时链路为什么这么重

下图是一个典型的大数据架构：数据集成任务把各种数据源分别接入实时消息队列和离线存储，实时和离线两条链路各自做数仓分层，最后统一写进 OLAP 引擎对外服务。

最明显的问题是实时和离线的一致性，那是流批一体要解决的事。这篇文章只看实时链路，它的中间层用的是 Kafka，面临三个问题：

问题	后果
不支持更新	处理更新流时，只能把全部数据物化在 Flink 状态里，消耗大量资源
查询能力弱	开发调试时看不到中间数据，一般要再同步到 OLAP 里，增加成本
回溯数据难	消息只保留几天，需要初始化或回溯时，得另开一条离线链路

Kafka 是为事件流设计的，而实时业务大多是分析场景，这个错配是上面这些痛点的共同根源。结果就是所有复杂逻辑都只能在 Flink 里开发，实时开发的成本居高不下。我们需要一个面向分析场景的流存储。

更多背景可以看 Fluss 官方博客 Why Fluss。下面逐个看 Fluss 的能力怎么解决这些问题：

痛点	Fluss 的能力	本文章节
去重、更新都压在 Flink 状态里	主键表，写入时自动生成变更日志	更新
双流 Join 的状态爆炸	部分列更新、Delta Join	双流关联
去重计数需要整型 ID	Insert If Not Exists 做全局字典	全局字典
聚合状态过大	存储层的聚合合并引擎	聚合
回溯要另开离线链路	分层存储到数据湖，联合读取	湖流一体

更新：存储那缺失的能力

Fluss 是面向实时分析（现在也加上了 AI）设计的流存储。它从底层把流和表融为一体：数据被看作结构化的、支持更新的列式表。

列式存储：基于 Apache Arrow，支持列裁剪和谓词下推，只读需要的列；
主键更新：支持主键表，写入时自动生成变更日志（Changelog），省掉大量 Flink 状态；
数据湖集成：内置分层（Tiering）服务，自动把温冷数据同步到数据湖，并支持合并查询；
交互式查询：内置 KV 索引，可以按主键高效点查，调试方便。

举个例子：用 Kafka 处理更新流时，Flink 任务要先用 ROW_NUMBER 按主键去重，把每个主键的最新值存在状态里，然后才能开发真正的业务逻辑。换成 Fluss 主键表后，存储本身就会生成正确的变更日志，下游任务可以直接写业务逻辑；主键表还能直接作为维表被关联，整体架构更简单，数据的复用度也更高。

Kafka 只追加事件，下游要自己按主键去重；Fluss 主键表写入时就生成正确的变更日志

双流关联：终极 BOSS

在 Flink 的双流 Regular Join 中，为了让两条流里相隔很久才到达、但 key 相同的数据能正确关联，Flink 必须在状态中永久、完整地保存两条流的历史数据。这带来了两个问题：状态爆炸，以及一边更新就要回撤另一边已经关联过的结果，回撤会被放大（回撤的问题详见 Flink SQL 回撤机制与问题）。Fluss 分两种情况解决。

关联键是两边的主键：部分列更新

如果关联键恰好是两条流的主键，可以直接利用 Fluss 的 Partial Update：两条流各自只写自己那几列，Fluss 找到该主键的现有记录，只更新提供的列，其余列保持不变。下游读到的是拼好的完整行的最新变更，关联在存储里完成了，Flink 不需要任何关联状态。

Partial Update示例代码

sql

CREATE CATALOG fluss_catalog 
WITH (
    'type' = 'fluss',
    'bootstrap.servers' = 'localhost:9123'
);

-- read from file for example
CREATE TEMPORARY TABLE IF NOT EXISTS fluss_catalog.fluss.fact_pk (
    dim_a BIGINT,
    metric BIGINT,
    ts BIGINT,
    PRIMARY KEY (dim_a) NOT ENFORCED
) WITH (
    'connector' = 'filesystem',
    'path' = 'file:///path/to/whatever',
    'format' = 'json'
);

-- read from file for example
CREATE TEMPORARY TABLE IF NOT EXISTS fluss_catalog.fluss.dim (
    dim_a BIGINT,
    attr BIGINT,
    ts BIGINT,
    PRIMARY KEY (dim_a) NOT ENFORCED
) WITH (
    'connector' = 'filesystem',
    'path' = 'file:///path/to/whatever',
    'format' = 'json'
);


CREATE TABLE IF NOT EXISTS fluss_catalog.fluss.fact_pk_dim_join (
    dim_a BIGINT,
    metric BIGINT,
    fact_ts BIGINT,
    attr BIGINT,
    dim_ts BIGINT,
    PRIMARY KEY (dim_a) NOT ENFORCED
) WITH (
    'bucket.key' = 'dim_a'
);

-- * just insert, merged by fluss
INSERT INTO fluss_catalog.fluss.fact_pk_dim_join (dim_a, metric, fact_ts)
SELECT
    dim_a,
    metric,
    ts AS fact_ts
FROM
    fluss_catalog.fluss.fact_pk;


-- * just insert, merged by fluss
INSERT INTO fluss_catalog.fluss.fact_pk_dim_join (dim_a, attr, dim_ts)
SELECT
    dim_a,
    attr,
    ts AS dim_ts
FROM
    fluss_catalog.fluss.dim;

关联键是外键：Delta Join

如果关联键是外键（比如订单表用 user_id 关联用户表），可以用 Delta Join：一边的数据到来时，直接拿关联键到另一边的表里按索引查找，代替双流 Join 维护的大状态，也就是直接复用源表本身的数据。

Delta Join示例代码

sql

SET 'table.optimizer.delta-join.strategy' = 'FORCE';


CREATE CATALOG fluss_catalog 
WITH (
    'type' = 'fluss',
    'bootstrap.servers' = 'localhost:9123'
);


CREATE TABLE IF NOT EXISTS fluss_catalog.fluss.fact_cdc (
    dim_a BIGINT,
    dim_b BIGINT,
    metric BIGINT,
    ts BIGINT,
    PRIMARY KEY (dim_a, dim_b) NOT ENFORCED
) WITH (
    'bucket.key' = 'dim_a',
    -- Flink 2.1 only support append-only source
    -- 'table.merge-engine' = 'first-row',
    -- Flink 2.2 support cdc source
    'table.delete.behavior' = 'ignore'
);

CREATE TABLE IF NOT EXISTS fluss_catalog.fluss.dim_cdc (
    dim_a BIGINT,
    attr BIGINT,
    ts BIGINT,
    PRIMARY KEY (dim_a) NOT ENFORCED
) WITH (
    'bucket.key' = 'dim_a',
    'table.delete.behavior' = 'ignore'
);


CREATE TABLE IF NOT EXISTS fluss_catalog.fluss.fact_dim_join (
    dim_a BIGINT,
    dim_b BIGINT,
    metric BIGINT,
    fact_ts BIGINT,
    attr BIGINT,
    dim_ts BIGINT,
    PRIMARY KEY (dim_a, dim_b) NOT ENFORCED
) WITH (
    'bucket.key' = 'dim_a'
);


INSERT INTO fluss_catalog.fluss.fact_dim_join
SELECT
    fact_cdc.dim_a,
    fact_cdc.dim_b,
    fact_cdc.metric,
    fact_cdc.ts AS fact_ts,
    dim_cdc.attr,
    dim_cdc.ts AS dim_ts
FROM
	fluss_catalog.fluss.fact_cdc /*+ OPTIONS('scan.startup.mode' = 'earliest') */ fact_cdc
-- ! only support inner join
INNER JOIN
	fluss_catalog.fluss.dim_cdc /*+ OPTIONS('scan.startup.mode' = 'earliest') */ dim_cdc
ON fact_cdc.dim_a = dim_cdc.dim_a

Delta Join 会输出重复数据，这是它做出的一点妥协。可以用一点关系代数来解释。两张表 A、B 分别新增了 ΔA、ΔB，关联结果的增量是：

\Delta(A\bowtie B) = \Delta A\bowtie B + A\bowtie \Delta B + \Delta A\bowtie \Delta B = \Delta A\bowtie (B+\Delta B) + A\bowtie \Delta B

注意第二项里的 A 是变化之前的 A。要精确计算，就得访问某个历史时刻的快照，只有少数存储支持，而且至少会引入分钟级的延迟。Delta Join 的做法是两边都只查最新的数据：

\Delta A\bowtie (B+\Delta B) + (A+\Delta A)\bowtie \Delta B

这样多出了一项 $\Delta A\bowtie\Delta B$ ，就是重复数据的来源。详见 FLIP-486: Introduce A New DeltaJoin。

全局字典：Insert If Not Exists

用 RoaringBitmap 做高效去重计数时有两个关键要求：ID 必须是整数，而且分布要稠密（原因见大数据去重统计）。这时可以用 Fluss 的 Insert If Not Exists 做一个全局字典：查找时如果映射不存在，就插入一行并用自增列分配一个新的整数 ID。

Insert If Not Exists代码示例

sql

CREATE CATALOG fluss_catalog 
WITH (
    'type' = 'fluss',
    'bootstrap.servers' = 'localhost:9123'
);

CREATE TEMPORARY TABLE IF NOT EXISTS fact (
    dim_a BIGINT,
    dim_b BIGINT,
    metric BIGINT,
    ts BIGINT,
    ptime AS proctime(),
    PRIMARY KEY (dim_a, dim_b) NOT ENFORCED
) WITH (
    'connector' = 'filesystem',
    'path' = 'file:///path/to/whatever',
    'format' = 'json'
);


CREATE TABLE IF NOT EXISTS fluss_catalog.fluss.dim_mapping (
    dim_b        BIGINT,
    uid_int32    BIGINT,
    PRIMARY KEY (dim_b) NOT ENFORCED
) WITH (
    'bucket.key' = 'dim_b',
    'auto-increment.fields' = 'uid_int32'
);


SELECT
    fact.dim_a,
    fact.dim_b,
    fact.metric,
    fact.ts,
    dim_mapping.uid_int32
FROM fact
LEFT JOIN fluss_catalog.fluss.dim_mapping /*+ OPTIONS('lookup.insert-if-not-exists' = 'true') */
    FOR SYSTEM_TIME AS OF fact.ptime AS dim_mapping
ON fact.dim_b = dim_mapping.dim_b;

聚合：把聚合下推到存储

目前主流的聚合都在 Flink 计算层执行，聚合结果就是 Flink 的状态，状态管理压力很大。Fluss 的 Aggregation Merge Engine 把聚合逻辑下推到存储层：写入的每一行按主键和已有的值合并（比如求和、取最新值），系统的扩展性和性能都更好。

Aggregation Merge Engine：按主键在存储层合并，metric 求和、ts 取最新的非空值，DELETE 被忽略

注意上图中 DELETE 被忽略了。为什么 Paimon 的聚合引擎支持回撤，而 Fluss 不支持？

根源在可见性模型：

Paimon 是快照事务可见。聚合发生在 compaction 阶段，回撤可以先作为变更日志落盘，之后再用 sum 等函数的逆运算抵消；出错了也能回滚到上一个快照。
Fluss 是写入即可见。聚合在服务端同步应用，没有可以丢弃的中间状态。它靠 Undo Recovery 保证精确一次：checkpoint 时记录每个分桶的位点，故障恢复时用变更日志里的历史值把数据硬覆盖回 checkpoint 时刻，再由 Flink 重放重算。这依赖一个前提：重放的是确定性的 +I 流。而回撤打破了这个前提：上游的回撤序列在重放时不保证一致，一旦同步应用就无法撤销。

所以 Fluss 聚合表默认 'table.delete.behavior' = 'ignore'，这是当前模型下唯一安全的选择。

Paimon 用快照换来了回撤的可逆性，Fluss 用实时可见性换来了写入路径的简单和低延迟——回撤恰好站在 Fluss 这条路径的反面。

湖流一体：Union Read

传统的离线和实时是两条链路，需要长周期计算或实时回溯时，很难把两边的数据对齐。Fluss 自带的分层服务会把实时数据定期同步到数据湖（Paimon），查询时通过 Union Read 先读数据湖里的快照，再无缝切换到 Fluss 中更新的实时数据，对业务暴露的是一张统一的逻辑表。

Union Read：先读 Paimon 里已提交的快照，再接上 Fluss 里还没归档的最新数据

湖流一体代码示例

sql

CREATE CATALOG fluss_catalog
WITH (
    'type' = 'fluss',
    'bootstrap.servers' = 'localhost:9123'
);

-- 需要先启动分层服务的 Flink 作业：
-- https://fluss.apache.org/docs/maintenance/tiered-storage/lakehouse-storage/
CREATE TABLE IF NOT EXISTS fluss_catalog.fluss.fact_lake (
    dim_a BIGINT,
    dim_b BIGINT,
    metric BIGINT,
    ts BIGINT,
    PRIMARY KEY (dim_a, dim_b) NOT ENFORCED
) WITH (
    'bucket.key' = 'dim_a',
    'table.datalake.enabled' = 'true',
    'table.datalake.freshness' = '10s'
);

-- 联合读取：先读 Paimon 中的快照，再接着读 Fluss 中的实时数据
SELECT dim_a, SUM(metric) AS metric
FROM fluss_catalog.fluss.fact_lake
GROUP BY dim_a;

-- 只读数据湖中已提交的部分（会落后约一个新鲜度周期）
-- 注意：实测 Flink 2.2 读不了 $lake，需要用 Flink 1.20
SELECT dim_a, SUM(metric) AS metric
FROM fluss_catalog.fluss.`fact_lake$lake`
GROUP BY dim_a;

总结

通过这些例子可以看出：计算和存储一起解决问题，才是实时计算的未来。Fluss 整体生态还不太成熟，但它的设计理念确实切中了当前实时开发的痛点。也许不久之后，架构就可以换成下面这样，整个大数据体系越来越一体化：

设想中的架构：在原来的分层上加入分层服务（Tiering Service），流和湖连成一体

参考资料

Why Fluss，Fluss 官方博客。
Jack Vanlightly，Understanding Apache Fluss。
Fluss First Impression，Streaming Data Technologies。
FLIP-486: Introduce A New DeltaJoin
FLIP-516: Multi-Way Join Operator
阿里云 Flink 文档：Changelog 事件乱序处理原理
Fluss 聚合表 + RoaringBitmap：实时多维 UV 去重实战
Delta Join：为超大规模流处理实现计算与历史数据解耦

全文评论

正在加载评论…