企业官网建设流程全解析

先把问题放进一个常见的电商结算场景：订单原价 100 元，客服改价后变成 80 元。Paimon 明细表查询正确，Flink Checkpoint 也连续成功，可下游按门店汇总的销售额却从 100 加到了 180。

如果只盯着“表里是 80”和“Checkpoint 是绿的”，这起事故看起来毫无道理。真正缺失的不是 80 元这个新值，而是“先把旧的 100 元减掉”那条撤回消息。

Exactly-Once 只能保证同一提交不会多算一次，不能凭空补出下游聚合所需的UPDATE_BEFORE。

下文的配置、源码和行为判断固定在Apache Paimon 2.0.0，源码基线为release-2.0.0/604e6d5e...。

先找到丢失的那一步：80 元到了，100 元没有撤回

Changelog Producer 官方文档 明确区分四种模式：

默认none下，Paimon 能表达某个主键的新值，却不一定直接给出旧值。下游若按store_id聚合，而表主键是order_id，没有旧金额就不知道该先减 100 再加 80。

为什么 Checkpoint 成功仍然救不了这笔汇总

源端成功：CDC 事件完整到达 提交成功：Checkpoint 对应的 Commit Message 进入 Snapshot 消费正确：下游收到足以更新自己状态的 RowKind 序列

最小实验不要只数行。写入+I(100)，再更新为-U(100), +U(80)，分别观察 Paimon 流读输出和下游聚合。若上游只给+U(80)却配置input，Paimon 不会发明缺失的-U(100)。

CREATETABLEorders(order_idBIGINT,store_idBIGINT,amountDECIMAL(18,2),PRIMARYKEY(order_id)NOTENFORCED)WITH('changelog-producer'='lookup');

这条 DDL 不是默认建议。lookup把旧值查询放进 Compaction 路径，会使用本地内存与磁盘缓存，并可能拉长提交。官方还特别提醒 Changelog Producer 可能显著降低 Compaction 性能。

别再只数行：同一笔更新必须核对 RowKind

把同一订单写入四张仅修改changelog-producer的表，才能公平验证差异。共同前提是表主键为order_id，下游按store_id汇总；输入先为+I(1001,10,100)，再把金额改成 80。

正确撤回流：+I(1001,10,100) → -U(1001,10,100) → +U(1001,10,80) 危险更新流：+I(1001,10,100) → +U(1001,10,80)

第二条对主键 Upsert Sink 可能足够，对SUM(amount) GROUP BY store_id却缺少减 100 的证据。若结果变成 180，先查审计 Sink 中的 RowKind，再查 Checkpoint；顺序反过来会把语义缺失误诊成重复提交。

固定到release-2.0.0，旧值生成和读取的最短源码链是：

MergeTree Compaction → LookupChangelogMergeFunctionWrapper / FullChangelogMergeFunctionWrapper → ChangelogResult → Changelog File → IncrementalChangelogReadProvider

源码能证明变化在哪里补齐和落盘，不能证明缓存容量足够，也不能证明消费者按 RowKind 正确处理。

-- 观察提交是否携带 Changelog；只读。-- 有记录不能替代逐主键 RowKind 与业务聚合对账。SELECTsnapshot_id,commit_kind,total_record_count,changelog_record_countFROMorders$snapshotsORDERBYsnapshot_id;

本实验能证明 Changelog 契约是否足够，不能证明生产规模下的性能。机器、Bucket、更新率与 SLA 未提供，必须另外压测 Checkpoint P99、Compaction 时长、缓存占用和 Changelog 文件增速。

该选哪种模式，答案藏在下游怎么消费

数据库型 Sink 能按主键覆盖时，none往往已经够用；非主键聚合需要撤回旧值，应先确认输入是否自带完整 Before/After，再在input与lookup之间选择。full-compaction适合能够接受更长生成周期的场景，不能仅为省掉 Normalize State 就默认打开。

用 Java 把 RowKind 打出来，别再猜撤回有没有发生

固定版本源码如何闭环到输出

源码固定为 Apache Paimonrelease-2.0.0/604e6d5e...：

• paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java的getResult()在包含 L0 且策略要求 Changelog 时调用setChangelog(before, after)，决定是否产生-U/+U；
• paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapper.java的getResult()比较合并前后的记录，决定完整 Compaction Changelog；
• 对应源码：Lookup wrapper、Full wrapper。

Java 中的Row.getKind()正是最终观测点：若金额 100→80 只有+U(80)而没有旧值撤回，下游非主键聚合就缺少减 100 的依据。它证明输出契约，不证明消费者已经正确执行撤回。

以下程序按paimon-flink-1.20:2.0.0API 编写，args[0]指向已有测试表的 Warehouse；流式collect()会持续读取，生产应接入可持久化审计 Sink。本环境未运行全文示例。

importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.table.api.EnvironmentSettings;importorg.apache.flink.table.api.TableResult;importorg.apache.flink.table.api.bridge.java.StreamTableEnvironment;importorg.apache.flink.types.Row;importorg.apache.flink.util.CloseableIterator;publicfinalclassChangelogAuditJob{publicstaticvoidmain(String[]args)throwsException{if(args.length!=1)thrownewIllegalArgumentException("warehouse is required");Stringwarehouse=args[0].replace("'","''");StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(30_000L);StreamTableEnvironmentt=StreamTableEnvironment.create(env,EnvironmentSettings.newInstance().inStreamingMode().build());t.executeSql("CREATE CATALOG p WITH ('type'='paimon','warehouse'='"+warehouse+"')");t.executeSql("USE CATALOG p");TableResultresult=t.executeSql("SELECT order_id,amount FROM demo.orders");try(CloseableIterator<Row>rows=result.collect()){while(rows.hasNext()){Rowrow=rows.next();System.out.printf("%s id=%s amount=%s%n",row.getKind(),row.getField(0),row.getField(1));}}}}

生产中应把 Print 换成可持久化审计 Sink。输出缺少旧金额的-U时，Checkpoint 成功不能挽救非主键聚合。lookup/full-compaction分别进入LookupChangelogMergeFunctionWrapper或FullChangelogMergeFunctionWrapper，随后由IncrementalChangelogReadProvider供流读。

数据库型 Sink 通常能按主键 Upsert，none往往足够；Flink 非主键聚合需要完整撤回，应该在上游已完整时用input，否则评估lookup；能容忍较长延迟、又必须从表状态推导完整变化时再考虑full-compaction。

不要为了消除 Flink Normalize 就无条件打开lookup。应同时压测：Checkpoint P95/P99、Compaction 时长、本地缓存占用、Changelog 文件数，以及故障恢复后的业务聚合对账。

Paimon 表里的最终值正确，只证明状态存对了；下游能不能算对，还取决于变化过程有没有被完整表达。

官方资料

• Changelog Producer
• Primary Key Table
• Snapshot Specification
• Flink Consumer ID