去年有个做实时数仓的朋友拿着一版 SQL 来找我，说 Flink 在 sink 端往 Kafka 写数据，怎么调都报Table sink doesn't support consuming update changes，他明明在 DDL 里声明了主键，也反复换了几个 format，就是过不去。我扫了一眼他的 WITH 参数就明白了——用的是普通的kafkaconnector，而不是upsert-kafka。这道坎在实时链路里出现频率极高：只要上游是聚合、去重、双流 join 出来的变更流，普通 Kafka sink 立刻罢工，而Flink sink 端使用 Upsert Kafka正是为这种"带主键、需要覆盖更新"的场景准备的。

这篇内容就是把我在几个模拟项目里落这套方案的完整过程摊开讲：从它到底补了哪个洞、DDL 每一行为什么这么写、exactly-once 的参数怎么配才不会互相打架，到我实际踩过的四个报错和对应的排查链路。适合已经能跑通 Flink SQL 基础作业、正准备把变更流落到 Kafka 的同学，也适合那些"作业能跑但数据对不上"的老手复查配置。下面所有表名、字段、场景都是虚构的模拟项目，参数组合你可以直接抄，但版本差异部分务必对照自己手里的环境核一遍。

1. Upsert Kafka 在 sink 端到底补了哪个洞

1.1 普通 kafka sink 见到 -U 就报错的真实原因

要理解 upsert-kafka 的价值，得先搞清楚普通kafkaconnector 在 sink 端的能力边界。它本质上是一个append-only 的写入器：拿到一条数据，序列化，追加到 topic 末尾，结束。它不关心这条数据是新增还是更新，也没有任何"覆盖"的概念，因为 Kafka 本身就不支持按 key 覆盖写入——分区内的消息是严格追加的日志，写完就钉在那儿了。

而 Flink 的 Table API 在 planner 阶段会做一次changelog 能力校验：上游算子输出的变更类型（RowKind）必须能被 sink 消费。如果上游是GROUP BY聚合、ROW_NUMBER去重、或者带时间区间的双流 join，输出流里就会混进UPDATE_BEFORE（-U）和UPDATE_AFTER（+U）。普通 kafka sink 只声明自己支持INSERT，planner 一看对不上，直接抛错，作业连提交都提交不了。

这里有个容易被忽略的点：报错发生在编译期，不是运行期。所以你在日志里看不到任何数据，只有一句干巴巴的异常。很多人第一反应是去改 format、改序列化，其实方向完全错了——问题不在格式，在于 sink 根本不具备处理变更语义的能力。我见过有同学为了绕过这个报错，在 SQL 里加WHERE过滤或者用INSERT INTO ... SELECT硬转成 append 流，结果更新记录全被丢掉，下游看到的永远是第一次写入的旧值。这种"改到不报错"的做法，比报错本身危险得多。

1.2 写入协议拆解：key 负责定位，value 负责覆盖，null 负责删除

upsert-kafka 的做法很朴素：既然 Kafka 物理上不能覆盖，那就把"覆盖"这件事交给下游按键取最新来完成，sink 端负责保证同一主键的消息落在同一分区、且顺序正确。具体落成三条规则：

• INSERT和UPDATE_AFTER：把主键序列化后作为消息 key，其余字段序列化后作为 value，正常写入；
• UPDATE_BEFORE：不写入，直接被 sink 吞掉，因为它不携带任何新的状态；
• DELETE：写入一条 key 正常、value 为null的消息，也就是常说的tombstone（墓碑消息）。

关键在于第一行的"同一主键落同一分区"。Kafka 的生产者默认分区器会对 key 的字节做哈希取模，只要 key 的字节序列稳定，同一个订单号永远进同一个分区，分区内又是有序的，所以"后写的覆盖先写的"在逻辑上成立。这也解释了一个必须讲清楚的认知：upsert 是逻辑语义，不是物理覆盖。topic 里依然躺着这个主键的所有历史版本，只是消费方按 key 折叠之后看到的是最新一条。如果你用命令行去consume这个 topic，会看到同一个订单号出现七八次，别慌，这是正常的。

这也是为什么 upsert-kafka 强制要求表上声明PRIMARY KEY ... NOT ENFORCED，并且key.format是必填项。没有 key 就没有分区定位的依据，整套语义直接崩塌。相比之下，普通 kafka sink 的key是可选的元数据，可以完全不用，两者的设计出发点根本不同。

1.3 和 JDBC sink、普通 kafka sink 的选型对照

实际项目里，往哪里落变更流通常有三条路，我按自己的使用频次做个对照，方便你判断什么时候该选谁。

选型上我的判断逻辑很直接：如果下游还要用 Flink 继续消费、并且是"按主键取最新状态"的用法，那 upsert-kafka 是顺理成章的；如果下游要的是能直接 SQL 查询的明细，那 JDBC sink 或某种 OLAP 连接器更合适，硬塞进 Kafka 再抽出来只是多绕一圈。而普通 kafka sink 的定位就是纯日志型追加，别指望它承载状态更新。

还有一个现实因素：Kafka 做状态载体时，容量成本比数据库低得多，写入吞吐也高一个量级，所以很多团队愿意用 upsert-kafka 做"实时数仓的中间层"，把 ODS 到 DWD 的变更流先用 Kafka 承接，再分发给多个下游。这种"一份变更流多处消费"的架构，是它相对 JDBC sink 最大的优势。

2. 模拟案例：订单明细变更流的落地方案

2.1 场景设定与上游变更流的产生方式

假设有个模拟项目，上游是一张订单明细表，业务上同一条明细会被反复修改：改数量、改金额、改状态、甚至整条作废。源端通过 CDC 抽到一个 Kafka topic，我们在 Flink 里做清洗和轻度聚合，然后落到下游的dwd_order_itemtopic 供实时看板和风控消费。

上游的变更流是这样产生的，也是实际项目里最常见的两种形态：

• 形态一：CDC 直接产出。源库的 binlog 被采集工具转成 changelog，+I、-U、+U、-D四种类型齐全，字段里通常还带一个op标识列。
• 形态二：Flink SQL 聚合产出。比如按订单号做GROUP BY算总金额，输出的是 retract 流，包含-U和+U；或者用ROW_NUMBER()做去重取最新一条，输出的是 upsert 流。

这两种形态对 sink 的要求不完全一样，但都超出了普通 kafka sink 的能力。我在模拟项目里用的是形态二，因为它更容易暴露问题——聚合流的状态更大、乱序更明显，正好用来验证 sink 端配置是否扎实。

下游消费侧的要求也很明确：按order_id+item_id取最新一条状态，同一个主键的历史版本允许保留，但读取时必须能折叠。这个要求直接决定了 topic 的清理策略和 key 的设计，后面第 6 节会详细说。

2.2 DDL 逐行拆解：为什么 key.format 和 value.format 必须分开写

先看完整 DDL，这是我在项目里实际用的版本，字段名改过但结构一致：

CREATE TABLE dwd_order_item_sink ( order_id STRING, item_id STRING, user_id STRING, sku_id STRING, quantity INT, amount DECIMAL(18, 2), order_status STRING, update_time TIMESTAMP(3), PRIMARY KEY (order_id, item_id) NOT ENFORCED ) WITH ( 'connector' = 'upsert-kafka', 'topic' = 'dwd_order_item', 'properties.bootstrap.servers' = 'broker-a:9092,broker-b:9092,broker-c:9092', 'key.format' = 'json', 'value.format' = 'json', 'sink.delivery-guarantee' = 'exactly-once', 'sink.parallelism' = '6', 'properties.transaction.timeout.ms' = '900000', 'properties.linger.ms' = '50', 'properties.batch.size' = '262144' );

第一处最容易绊人的地方：必须写key.format和value.format，不能只写一个format。普通 kafka connector 里format是统一配置，key 和 value 用同一套序列化；upsert-kafka 在 DDL 校验阶段就要求两者分开，缺了key.format会直接报The Kafka key format must be specified之类的异常。原因也简单：key 只需要主键那几列，value 是其余全部列，两者字段集不同，无法共用一份 schema 推导。

第二处，PRIMARY KEY后面必须跟NOT ENFORCED。Flink SQL 的主键声明是纯元数据，运行时并不做强约束，所以这个后缀是必需的。主键里列的顺序会影响 key 的序列化结果——一旦上线就别随便调整顺序，否则下游按 key 折叠的逻辑会错乱。我在模拟项目里就因为把主键从(item_id, order_id)改成(order_id, item_id)导致下游解析失败过一次，虽然字段集合没变，但 JSON key 的字节序列变了，哈希分区结果跟着变，历史数据和新数据落到了不同分区。

第三处，key.format = 'json'意味着 key 在 Kafka 里是一个 JSON 对象，比如{"order_id":"A1001","item_id":"I02"}，而不是拼接好的裸字符串。这个细节影响下游实现：如果消费方想用order_id做业务分流，得先解析 JSON 再取值，不能直接把消息 key 当字符串用。想要裸值可以换成csv或raw，但raw只支持单字段且需要额外指定分隔符，复合主键基本用不了。

2.3 key.fields-prefix 与 value.fields-include 的真实用途

大部分案例走不到这一步，但只要你的表里出现"同一个字段既要进 key 又要进 value"的情况，就会撞上字段名冲突。举个具体例子：风控侧希望消息 value 里也保留order_id，方便不解析 key 就能定位，但order_id同时又是主键的一部分。

默认行为是value.fields-include = 'ALL'，也就是 value 包含表里所有列（含主键列），此时 key 和 value 共存同一字段名不会有问题——因为它们走的是两套独立的序列化路径。真正会打架的是另一种场景：key 和 value 用了不同的 format，比如 key 用json、value 用avro，而 avro 的 schema 里又包含同名字段，这时部分版本的 schema 推导会报重复字段。

解决办法就是加前缀，把 key 用的列单独起一套名字：

CREATE TABLE dwd_order_item_sink_v2 ( k_order_id STRING, k_item_id STRING, order_id STRING, item_id STRING, amount DECIMAL(18, 2), update_time TIMESTAMP(3), PRIMARY KEY (k_order_id, k_item_id) NOT ENFORCED ) WITH ( 'connector' = 'upsert-kafka', 'topic' = 'dwd_order_item_v2', 'properties.bootstrap.servers' = 'broker-a:9092', 'key.fields-prefix' = 'k_', 'value.fields-include' = 'EXCEPT_KEY', 'key.format' = 'json', 'value.format' = 'json' );

配上'key.fields-prefix' = 'k_'之后，主键列k_order_id、k_item_id在序列化时前缀会被剥掉，写进 Kafka 的 key 依然是{"order_id":...,"item_id":...}，下游完全无感。而'value.fields-include' = 'EXCEPT_KEY'让 value 里排除掉所有带k_前缀的列，于是 value 只剩业务字段。

这套组合还有个隐性好处：它强制你把 key 字段和 payload 字段在表结构上区分开，代码读起来一目了然，不会出现"这个字段到底是主键还是普通列"的困惑。我在模拟项目里第二版就换成了这种写法，后面加字段时再没出现过 schema 冲突。要注意的是EXCEPT_KEY排除的是加了前缀的那批列，不带前缀的同名列会正常保留在 value 里，正好符合"key 和 payload 都要有 order_id"的诉求。

3. 让 exactly-once 真的生效的参数组合

3.1 transaction.timeout.ms 与 checkpoint 间隔的死锁式耦合

听到"upsert-kafka 支持 exactly-once"就以为配个 connector 就完事，是新手最容易栽的坑。它的实现依赖 Kafka 事务：每次 checkpoint 时提交一次事务，checkpoint 之间写入的数据都处于未提交状态。于是参数之间形成了一条强耦合链。

Kafka producer 有一个transaction.timeout.ms，默认 1 小时；broker 端有一个transaction.max.timeout.ms，默认 15 分钟。producer 的值必须小于等于 broker 的值，否则连接阶段就会抛The transaction timeout is larger than the maximum value allowed by the broker。这是我在模拟项目里遇到的第一个 exactly-once 相关报错，当时以为是集群权限问题，查了半天才发现是两边默认值天生不匹配。

调整思路有两个方向，要看你们的基础设施约束：

• 改 broker 的transaction.max.timeout.ms，把它抬到比 Flink 侧的值大，这需要集群运维配合，改完要滚动重启；
• 改 Flink 侧的properties.transaction.timeout.ms，压到 15 分钟以内，这是作业级配置，自己就能改。

我一般选后者，把 Flink 侧设成900000（15 分钟），然后确保 checkpoint 间隔远小于它。如果 checkpoint 间隔是 3 分钟，那 15 分钟的事务超时意味着最多允许 5 次 checkpoint 失败不推进而不触发事务超时；如果 checkpoint 连续卡住超过 15 分钟，事务被 broker 中止，作业会失败重启。所以这条链的合理取值是：

一个反直觉的经验：checkpoint 间隔不是越短越好。曾有个模拟项目把间隔设成 20 秒，结果 Kafka 里瞬间堆出大量未提交事务，broker 的事务协调器压力陡增，写入延迟反而上去了，还触发过Too many open transactions类的告警。后来调到 2 分钟，吞吐稳定下来。事务开销是真实存在的，别拿 OLTP 的思维去调流作业。

3.2 事务前缀、并行度与 checkpoint 并发数

exactly-once 生效还有一个前提：每个 sink 并发的写入事务要能被正确识别和恢复。这和sink.transactional-id-prefix有关——它给每个 subtask 的 transactional.id 加前缀，保证作业重启后能接续未完成的事务，而不是创建一批新事务把老的遗留成僵尸。

要点有两个。其一，前缀必须在同一作业的所有实例间保持一致，如果多个作业写同一个 topic 却不加区分，事务 ID 会冲突，作业直接起不来。其二，前缀改了之后要确保旧事务已经清理，否则 broker 侧会残留悬挂事务，把read_committed消费者的 LSO（last stable offset）卡住，表现为"数据写进去了但下游一直读不到"。这个现象非常迷惑人，我在第 3.3 节专门讲。

并行度和分区数的关系也值得说。建议sink.parallelism不超过 topic 分区数，超出部分会空闲等待，白占资源；低于分区数则单并发要写多个分区，吞吐受限。我通常让它等于分区数，配合前面说的 key 哈希分区，每个 subtask 稳定负责一批 key，行为最好预测。

还有个小细节：Flink 的事务写入在 checkpoint 期间才提交，如果你在作业运行时用命令行去看 topic，会看到"数据量比预期少"，甚至完全为空——这不是数据丢了，是事务还没提交。别急着改配置，先看 checkpoint 是否正常完成。

3.3 下游读不到数据的排查：isolation.level 与未提交事务

这条我单独拎出来讲，因为它是最容易让人怀疑"数据丢了"的假故障。用事务写入时，消息在提交前对read_committed的消费者是不可见的。而 Kafka 自带的kafka-console-consumer默认是read_uncommitted，所以你在命令行能看到数据，下游应用却读不到——两边的可见性视图根本不同。

排查顺序我是这么走的，按成本从低到高：

1. 确认作业 checkpoint 是否在正常完成，看 Flink UI 的 checkpoint 页面，有没有连续失败；
2. 确认下游消费者的isolation.level，如果是read_committed，检查 topic 的 LSO 是否被悬挂事务卡住，可以用kafka-consumer-groups或管理工具查看各分区的 LSO 与最新 offset 的差值；
3. 如果 LSO 明显落后，基本可以判定有未提交事务，去查是否有历史作业异常退出、transactional.id 前缀变更、或者事务超时被中止但 producer 没感知；
4. 确认下游消费者的auto.offset.reset和消费组位点，别在错误的方向上查半天。

还有一类情况是消费端把 tombstone 过滤掉了。开启read_committed且消费逻辑里写了if (value == null) continue;，那删除操作就永远传不到下游，表现为"源表删了，下游还在"。这个问题在第 4.2 节展开。

4. 四个真实报错背后的排查链路

4.1 doesn't support consuming update changes：从 DDL 到执行计划逐层定位

这个报错的全貌大概是Table sink 'xxx' doesn't support consuming update changes but its production is update changes。我的定位路径分三层，按顺序走基本不会绕弯。

第一层看 connector 类型。八成的情况就是用了kafka而不是upsert-kafka。这是最常见的，没有之一。

第二层看主键声明。如果 connector 已经是upsert-kafka还报类似错，那多半是主键没写或者写错了位置。upsert-kafka 要求PRIMARY KEY声明在列定义里（表级约束），带NOT ENFORCED。有同学把主键写在 WITH 里，或者用key.fields手工指定却忘了声明主键，都会失败。后者其实是可以工作的——key.fields能替代主键声明来确定 key 列，但没有主键时 planner 认为这是 append 流，仍然可能拒绝变更输入。稳妥做法是两个都写，主键声明为准，key.fields只在需要显式控制顺序时补上。

第三层看上游是不是真能产出 upsert 流。这一层最深，也最有意思。有些 SQL 写的其实是 retract 流（包含-U），而 upsert-kafka 期望的是 upsert 流（只有+I、+U、-D，不混用-U）。两者在 planner 里的表示不同，混用时 Flink 会自动插入一个SinkUpsertMaterializer算子来做归一化，但如果这个算子因为状态后端配置或内存不足起不来，也会报奇怪的错。

判断方法是用EXPLAIN看执行计划，找 sink 前一个节点的 changelog 模式。我在模拟项目里就靠这一步定位到一个隐藏问题：上游ROW_NUMBER()的窗口没有按主键分区，产出的其实是 retract 流，加了 materialize 之后状态暴涨，最后改成先在子查询里按主键聚合再输出，状态量降了一个数量级。顺带说，table.exec.sink.upsert-materialize这个参数可以控制该算子的行为（auto/force/none），碰到状态异常时值得调出来看，但具体可用性要看版本，我后面的版本差异一节有说明。

4.2 下游把 tombstone 当成脏数据丢掉

删除操作在 upsert-kafka 里表现为"value 为 null 的消息"，这一点如果没提前和下游对齐，几乎必然出问题。我在模拟项目上线第一周就撞上了：风控侧的消息处理逻辑里有一句空值判断，直接continue掉了，结果源端作废的订单在下游一直显示有效，对账时差了三百多条。

根因不在 Flink 侧，而在语义约定没有传递下去。修复做法有两层：一是下游必须显式处理 null value，把它解释为删除并同步更新自己的状态存储；二是如果下游实在不方便改，可以在链路中间加一层 Flink 作业，把 upsert 流转换成带op字段的普通变更流再写出，用显式字段代替 null 语义。

还有一种变体是用changelog-json格式。它会在 value 外面包一层{"+I": {...}}这样的标记，删除就是{"-D": {...}}，语义显式、可读性好。但我实测下来不建议在 upsert-kafka 上用它，原因有两个：一是语义重复，删除既是 null tombstone 又是-D标记，下游容易双重处理；二是部分版本对 changelog-json 与 upsert 语义的组合处理有差异，升级时容易踩坑。如果确实需要显式 op 字段，我更倾向于在业务字段里自己加一列op，值域自己定义，可控性强得多。

顺带提一个验证技巧：测试时专门构造一条"插入—更新—删除"的完整序列，然后用命令行消费 topic，逐条确认 key 正确、更新只写+U、删除写的是 null。这条三步验证我每次上线前都做，五分钟能省掉后面几小时的扯皮。

4.3 主键带 null 导致同一条订单散落在多个分区

这个坑相当隐蔽，因为它在测试环境几乎不会暴露——测试数据总是"干净"的。逻辑是这样的：Kafka 生产者的默认分区器，当消息 key 为 null 时会走轮询分区，而 key 非 null 时才走哈希。如果主键里的某个字段值恰好是 null，整个 key 序列化出来可能仍然是合法的（比如 JSON 里{"order_id":null,"item_id":"I02"}），但某些 format 下会退化成空字节，进而被当作 null key 处理，轮询到不同分区。

一旦同一逻辑主键的两条消息落进不同分区，"同分区有序"这个前提就没了，下游折叠时可能拿到旧值覆盖新值，表现为"数据偶尔回退"。这种问题极难复现，因为依赖 Kafka 的内部实现和时间顺序。

处理办法是在源头堵住。声明主键的列，务必在 DDL 上补NOT NULL，或者在上游 SQL 里用COALESCE兜底成一个确定值（比如空字符串或-1）。别指望 NOT ENFORCED 的主键帮忙做非空校验，它真的只是元数据。我在模拟项目里给主键列统一加了NOT NULL，并在上游 ETL 里做了空值清洗，之后这个现象再没出现过。

还有一个相关问题值得提：如果主键字段类型是DECIMAL或TIMESTAMP，不同 format 下的序列化结果差异较大，json可能输出带引号的字符串，avro是二进制。同一个 topic 的 key 序列化方式必须全局统一，不能 A 作业写 json、B 作业写 avro，否则同一业务主键会算出两种字节序列，落进不同分区。这在多团队共用一个 topic 时特别容易发生。

4.4 数据倾斜：单分区写入打满而其他分区闲着

按 key 哈希分区天然会倾斜，这不算 bug，但当主键分布极不均匀时，问题会被放大。我在一个模拟项目里见过极端情况：主键是"商户号 + 日期"，而某个大商户占了 40% 的订单量，结果对应分区持续打满，其他五个分区利用率不到 20%。表现是整体吞吐上不去、部分 subtask 的busyTime长期接近 100%，checkpoint 也会被最慢的那个 subtask 拖住。

解决思路按侵入性从小到大排：

• 加盐：在主键里掺一个随机后缀，写入时打散，读取时再做一次聚合。代价是下游逻辑变复杂，一般不用在核心链路上。
• 改主键粒度：从"商户号"细化到"商户号 + 订单号"，让分区更均匀。这是我首选的做法，前提是业务上确实需要这个粒度。
• 提高分区数：只能缓解不能根治，因为倾斜来自 key 分布本身，分区再多，热点 key 还是打同一个分区。
• 前置预聚合：把热点商户的聚合在 Flink 里做掉，减少写入量，sink 端压力自然下降。

监控上我习惯看两个指标：每个 sink subtask 的numRecordsOutPerSecond，以及 Kafka 各分区的写入速率。两者对照着看，能很快判断是算子级倾斜还是分区级倾斜。前者调并行度，后者只能动 key 设计。

5. 上线前怎么验证数据真的按 key 覆盖了

5.1 用命令行逐条比对 key 与 value

作业跑起来不等于数据对。我上线前的第一件事永远是手工验证，用的命令很朴素：

kafka-console-consumer.sh \ --bootstrap-server broker-a:9092 \ --topic dwd_order_item \ --property print.key=true \ --property print.partition=true \ --property key.separator=' | ' \ --from-beginning \ --max-messages 20

三个property是重点：print.key打开才会显示 key，print.partition能看出同一主键是否稳定落在同一分区，key.separator让 key 和 value 分开显示，可读性提升明显。

判读的时候看三件事：

• 同一个order_id+item_id的多次更新，partition是不是完全一致；
• 每次更新的 value 里字段是否完整，有没有出现只有部分字段的"半截消息"（这说明上游 schema 有问题）；
• 删除操作对应的 value 是否为 null，key 是否正常输出。

我还会额外看一眼消息的时间戳和 offset 是否单调递增，确认顺序没有乱。这一步花五分钟，但能挡掉后面八成的问题反馈。

5.2 源表与目标 topic 的对账 SQL 写法

手工抽检只能看个大概，量级验证还得靠对账。思路是把目标 topic 当成一张 upsert 表读回来，和源端聚合后的结果做比对。读回来用upsert-kafka作为 source，这时它的语义是"每条消息当作 upsert，value 为 null 视为删除"，正好契合。

CREATE TABLE dwd_order_item_source ( order_id STRING, item_id STRING, amount DECIMAL(18, 2), order_status STRING, PRIMARY KEY (order_id, item_id) NOT ENFORCED ) WITH ( 'connector' = 'upsert-kafka', 'topic' = 'dwd_order_item', 'properties.bootstrap.servers' = 'broker-a:9092', 'properties.isolation.level' = 'read_committed', 'properties.auto.offset.reset' = 'earliest', 'key.format' = 'json', 'value.format' = 'json' );

对账时我一般做三个比对，按严格程度递进：

只要前两项对得上，基本可以判断链路健康。第三项如果对不上但前两项正常，优先怀疑精度问题（DECIMAL精度不一致）或者聚合时机不一致，而不是链路丢数据。

这里有个必须提醒的点：用 upsert-kafka 当 source 时，topic 一定要配 compaction，否则历史版本会无限累积，读回来越来越慢。这个约束在 next section 细说。

5.3 一个够用的压测脚本与观察指标

容量验证我不搞复杂工具，直接用一个 Flink SQL 作业造数，往目标 topic 灌变更流，观察 sink 端表现：

INSERT INTO dwd_order_item_sink SELECT CONCAT('O', CAST(n AS STRING)) AS order_id, CONCAT('I', CAST(n % 5 AS STRING)) AS item_id, CONCAT('U', CAST(n % 1000 AS STRING)) AS user_id, CAST(n % 20 AS INT) AS quantity, CAST(n * 1.5 AS DECIMAL(18, 2)) AS amount, CASE WHEN n % 7 = 0 THEN 'CANCELLED' ELSE 'PAID' END AS order_status, CURRENT_TIMESTAMP AS update_time FROM ( SELECT * FROM TABLE( UNNEST_SEQUENCE(1, 5000000) ) );

（UNNEST_SEQUENCE不是标准语法，实际项目里可用 datagen 连接器或按业务表复制放大，这里只是示意造数逻辑。）

压测期间我盯四个指标，比盯 CPU 有用得多：

• sink subtask 的numRecordsOutPerSecond是否存在明显长尾，长尾说明分区倾斜；
• checkpoint 的alignment和duration，事务写入会放大对齐时间，超过超时阈值就必须调参数；
• Kafka 各分区的写入速率和磁盘增长曲线，判断 compaction 能不能跟上；
• 端到端延迟，从源消息时间戳到落库时间，这个最贴近业务体感。

我一般跑两轮，第一轮固定主键分布看基线，第二轮刻意造倾斜（比如 80% 的记录主键集中在 1000 个值里），看极限在哪。两轮数据合在一起，才敢给出容量结论。

6. 版本差异与不该用 Upsert Kafka 的场景

6.1 几个关键版本的行为变化

upsert-kafka 这几年的实现改过几轮，行为差异不小，升级前务必核对。我按自己踩过的几处总结：

• exactly-once 的实现方式：早期版本基于FlinkKafkaProducer的旧实现，后续切换到统一的 KafkaSink。切换之后，事务相关参数的生效方式有变化，部分版本默认投递语义并不是 exactly-once，需要sink.delivery-guarantee显式声明。这一点最坑，因为它不报错，只是默默退化成了 at-least-once，你只有通过"重启后是否出现重复数据"才能发现。
• upsert materialize 算子：SinkUpsertMaterializer的引入让 retract 流转 upsert 流成为可能，但它的状态存储在早期版本里没有很好的清理策略，长跑作业容易出现状态膨胀。对应参数table.exec.sink.upsert-materialize的可用性和取值也需要核对。
• key.fields-prefix的支持范围：这个配置在部分小版本上对 source 和 sink 的行为不完全对称，用作 sink 时表现正常、用作 source 时前缀处理有偏差。我现在的做法是只要用了前缀，就固定同一个大版本不随便升级。

核查方法很直接：跑一个最小的验证作业，写入"插入—更新—删除"三条记录，重启一次，看是否出现重复或丢失。这个用例五分钟能跑完，比翻 Release Note 快得多。

6.2 topic 配置：compaction 与 retention 的取舍

upsert-kafka 作为sink时，对 topic 的清理策略没有硬性要求，但作为source读回来时，必须配cleanup.policy=compact（或compact,delete），否则同一个 key 的历史版本会全部保留，读回来的数据量随时间线性膨胀，最终不可用。

我的经验配置是这样：

关于 tombstone 的保留：Kafka 的 compaction 会保留删除标记一段时间（delete.retention.ms，默认 24 小时）再物理清理。这个窗口必须长于最慢消费者的重启周期，否则一个离线超过窗口的消费者回来后，就再也看不到那条删除记录，导致它本地的旧数据永远删不掉。这个细节很少人提，但在有批处理回刷的架构里很致命。

还有一个常见误区：以为配了 compact 之后 topic 里就只有每个 key 一条数据了。不是的，compaction 是后台异步的，触发时机不固定，你会看到同一个 key 有多条记录并存，只是最终会收敛。下游实现必须是"按键取最新"的逻辑，不能假设只有一条。

6.3 什么时候老老实实用别的 sink

写到这里，我得说句实在话：upsert-kafka 不是万能的，有些场景硬套它反而添乱。

第一种，下游只需要最终结果、不做流式消费。比如定时报表，那直接落 JDBC 或 OLAP 库更省事，用 Kafka 中转反而多了一层需要维护的组件，还要操心 compaction 和消费者位点。

第二种，更新频率极高而 key 基数很小。比如几百个设备的状态每秒更新几十次，整个 topic 99% 都是同一个 key 的覆盖写入，那 Kafka 的日志结构完全不划算，写入放大严重、清理压力大，用 Redis 或时序库更合适。

第三种，需要按非主键维度查询。upsert-kafka 只保证按 key 定位，如果下游要按user_id或时间范围查询，那本质上需要一个支持索引的存储，Kafka 帮不了你。

第四种，团队没有事务与位点运维能力。exactly-once 带来的是运维复杂度：悬挂事务、LSO 卡住、事务超时调参，每一个都需要有人能看懂。如果这块没人兜底，老实退到 at-least-once 加下游幂等，稳定性反而更高。

我自己的取舍标准是：只要"一份变更流要被多个下游分别消费"这个诉求成立，且下游能接受按键取最新的约定，那就用 upsert-kafka；否则优先考虑直接写库。架构选择上没有银弹，把一个 append-only 的日志系统当成数据库用，代价迟早会以某种形式还回来。

最后分享一个我在实际操作里养成的小习惯：每个用 upsert-kafka 的作业，我都会在 DDL 旁边写一段注释，记清楚主键字段顺序、key 的序列化格式、tombstone 的处理约定这三件事，并把这段注释同步给下游消费方。听上去是体力活，但它挡掉的沟通成本，远比写这几行字的时间高。这套东西本质上不是配置问题，是上下游之间的一份契约，契约写清楚了，链路才真的稳。