为什么流式消费 Paimon 表时，通常需要配置 changelog-producer？

知识点图片

在 Apache Paimon 的架构设计中，Changelog Producer（变更日志产生器） 是一个非常关键的配置项。当您使用 Flink 等流计算引擎流式消费（Streaming Read）Paimon 表时，通常需要显式配置 changelog-producer。

根据 Paimon 官方文档（master 分支）的解释，这主要是为了保证流式消费的正确性、提升计算性能、以及满足特定合并引擎（Merge Engine）的强制要求。

以下是深度解析为什么流式消费通常需要配置 changelog-producer 的原因：

一、默认情况（none）的局限性：缺少“旧值”（UPDATE_BEFORE）

在默认情况下，Paimon 的 changelog-producer 配置为 none。这意味着 Paimon 写入器（Writer）在提交数据时，不会额外生成 changelog 文件。
此时，流式消费的 Reader 只能通过对比 Snapshot 的变化来感知数据的增量更新。这种模式存在致命的局限：

只能感知合并后的新值（Merged Changes）：流式消费只能看到哪些 Key 被删除了，或者哪些 Key 有了新值，但无法直接读取到发生更新之前的“旧值”（即 UPDATE_BEFORE，或 -U 消息）。
无法形成完整的 Changelog 链路：一条完整的更新流水应该包含“旧值撤回（-U）”和“新值累加（+U）”。没有旧值，就无法生成合规的 CDC 变更流。

二、核心原因解析

理由 1：保证下游状态计算（聚合、撤回）的正确性

在流式计算中，下游通常需要对非主键字段进行聚合（例如 SUM、COUNT）或者进行双流 Join。如果缺少 UPDATE_BEFORE 消息，下游的聚合逻辑就会出错。

官方文档中的经典示例：
假设下游流任务正在根据某个非主键字段（如分组键）计算 SUM 累加值。

某条数据发生更新，Paimon 发送了新值 5。
如果没有 changelog-producer，下游只能收到单个新值 5，而不知道“旧值是多少”。
- 如果旧值是 4，下游应该在原来的 SUM 结果上 加 1（即 5 - 4）；
- 如果旧值是 6，下游应该在原来的 SUM 结果上 减 1（即 5 - 6）。
如果拿不到旧值，下游就无法得知该如何对当前的聚合状态进行修正（撤回旧值，累加新值），从而导致统计结果彻底失真。

理由 2：消除 Flink 昂贵的 `Normalize` 状态算子

为了解决上述“拿不到旧值”的问题，Flink 提供了一个内置的 Normalize（标准化）算子。

工作原理：当 Paimon 无法提供完整的 Changelog 时，Flink 会在 RocksDB 状态（State）中把 Paimon 表的所有 Key 及其最新的 Value 完整地“缓存”一份。每当有新数据写入，Flink 会自己在状态中查出旧值，并在内存中拼装出 -U 和 +U 发送给下游。
代价极大：这个 Normalize 算子极其消耗系统资源，会导致 Flink 任务的 State 急剧膨胀，带来严重的内存和 CPU 开销。
如何解决：如果在 Paimon 侧配置了 changelog-producer，Paimon 就会在存储层直接生成好包含 -U 和 +U 的完整变更文件。流消费时可以直接读取，从而允许 Flink 旁路掉（或强行移除）这个昂贵的 Normalize 算子（可通过 scan.remove-normalize 强制移除），极大地优化了流任务的吞吐和资源占用。

理由 3：部分 Merge Engine 的硬性要求

Paimon 提供了多种高效的合并引擎（Merge Engine）来处理主键冲突。对于流式查询，许多 Merge Engine 强制要求或强烈建议配合特定的 changelog-producer 使用，否则无法输出正确的流数据：

Partial Update（部分更新）：在流式查询时，必须配合 lookup 或 full-compaction（或仅返回输入流的 input）使用，否则流读无法获取正确的合并结果。
Aggregation（聚合引擎）：在流式查询时，同样必须配合 lookup 或 full-compaction 使用，以确保下游能收到正确的撤回和更新流。
First Row（首行保留）：仅支持 none 和 lookup 模式。如果需要流式消费，则必须配置为 lookup。

三、 Paimon 提供的几种 Changelog Producer 模式

为了在不同业务场景下适配，Paimon 在配置 changelog-producer 时提供了以下几种选项：

模式	原理	适用场景 / 优缺点
`input`	写入器（Writer）直接将写入 Paimon 的原始输入数据，双写（Double Write）一份到独立的 Changelog 文件中。	适合输入源本身就是完整、规范的 Changelog 的场景（例如直接接收自 MySQL CDC、Kafka 中的 Canal/Debezium 格式，或者上游 Flink 有状态算子产生的流）。此模式开销极低。
`lookup`	在每次向 LSM-Tree 提交（Commit）写入之前，先通过点查（Lookup）内存或本地磁盘中的索引，找出该主键的历史旧值，从而即时生成 `-U` 和 `+U` 并写入 Changelog 文件。	适合输入数据是 Insert-only（如普通日志）或 Partial-update，但需要极低延迟流读的场景。该模式会在写入阶段引入额外的 Lookup 开销。
`full-compaction`	不在写入时即时寻找旧值，而是在后台进行 LSM-Tree 的 Full Compaction（全量合并）时，通过对比合并前后的数据差异，异步产生 Changelog 文件。	这种方式产生的 Changelog 最为准确和干净，且对写入吞吐的直接影响较小。但由于 Compaction 是周期性发生的，导致流式消费的延迟变大（取决于 Full Compaction 的触发间隔）。

总结

在流式 Lakehouse 架构中，配置 changelog-producer 的本质是“将状态计算的压力从下游 Flink 转移（或分摊）到上游 Paimon 存储层”。它通过在 Paimon 侧持久化生成规范的变更日志，既保证了下游流式计算在面对更新、删除、部分更新时能得到数学上绝对正确的撤回结果，又避免了 Flink 侧因状态无限膨胀而崩溃。

播面