去年有个做实时数仓的朋友拿着一版 SQL 来找我说 Flink 在 sink 端往 Kafka 写数据怎么调都报Table sink doesnt 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_AFTERU。普通 kafka sink 只声明自己支持INSERTplanner 一看对不上直接抛错作业连提交都提交不了。这里有个容易被忽略的点报错发生在编译期不是运行期。所以你在日志里看不到任何数据只有一句干巴巴的异常。很多人第一反应是去改 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 的选型对照实际项目里往哪里落变更流通常有三条路我按自己的使用频次做个对照方便你判断什么时候该选谁。对比维度普通 kafka sinkupsert-kafka sinkJDBC/OLAP sink支持的变更类型仅 INSERTINSERT / UPDATE_AFTER / DELETE全部是否要求主键否是NOT ENFORCED视库而定通常建议有物理覆盖无无逻辑覆盖有真实 UPDATEexactly-once 实现事务写入事务写入 key 分区事务或幂等写典型下游日志采集、埋点实时宽表、维表、CDC 汇聚报表库、明细查询数据膨胀与写入量一致与写入量一致靠 compaction 清理只保留最新状态选型上我的判断逻辑很直接如果下游还要用 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 被采集工具转成 changelogI、-U、U、-D四种类型齐全字段里通常还带一个op标识列。形态二Flink SQL 聚合产出。比如按订单号做GROUP BY算总金额输出的是 retract 流包含-U和U或者用ROW_NUMBER()做去重取最新一条输出的是 upsert 流。这两种形态对 sink 的要求不完全一样但都超出了普通 kafka sink 的能力。我在模拟项目里用的是形态二因为它更容易暴露问题——聚合流的状态更大、乱序更明显正好用来验证 sink 端配置是否扎实。下游消费侧的要求也很明确按order_iditem_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 侧设成90000015 分钟然后确保 checkpoint 间隔远小于它。如果 checkpoint 间隔是 3 分钟那 15 分钟的事务超时意味着最多允许 5 次 checkpoint 失败不推进而不触发事务超时如果 checkpoint 连续卡住超过 15 分钟事务被 broker 中止作业会失败重启。所以这条链的合理取值是参数建议取值说明checkpoint 间隔1 ~ 3 分钟太短会造成大量小事务太长则故障恢复回放多checkpoint 超时间隔的 3 倍左右给慢 checkpoint 留余量transaction.timeout.ms检查点间隔的 3 ~ 5 倍且 ≤ broker 上限保证正常 checkpoint 不会被误判超时transaction.max.timeout.ms (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消费者的 LSOlast 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所以你在命令行能看到数据下游应用却读不到——两边的可见性视图根本不同。排查顺序我是这么走的按成本从低到高确认作业 checkpoint 是否在正常完成看 Flink UI 的 checkpoint 页面有没有连续失败确认下游消费者的isolation.level如果是read_committed检查 topic 的 LSO 是否被悬挂事务卡住可以用kafka-consumer-groups或管理工具查看各分区的 LSO 与最新 offset 的差值如果 LSO 明显落后基本可以判定有未提交事务去查是否有历史作业异常退出、transactional.id 前缀变更、或者事务超时被中止但 producer 没感知确认下游消费者的auto.offset.reset和消费组位点别在错误的方向上查半天。还有一类情况是消费端把 tombstone 过滤掉了。开启read_committed且消费逻辑里写了if (value null) continue;那删除操作就永远传不到下游表现为源表删了下游还在。这个问题在第 4.2 节展开。4. 四个真实报错背后的排查链路4.1 doesnt support consuming update changes从 DDL 到执行计划逐层定位这个报错的全貌大概是Table sink xxx doesnt 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.keytrue \ --property print.partitiontrue \ --property key.separator | \ --from-beginning \ --max-messages 20三个property是重点print.key打开才会显示 keyprint.partition能看出同一主键是否稳定落在同一分区key.separator让 key 和 value 分开显示可读性提升明显。判读的时候看三件事同一个order_iditem_id的多次更新partition是不是完全一致每次更新的 value 里字段是否完整有没有出现只有部分字段的半截消息这说明上游 schema 有问题删除操作对应的 value 是否为 nullkey 是否正常输出。我还会额外看一眼消息的时间戳和 offset 是否单调递增确认顺序没有乱。这一步花五分钟但能挡掉后面八成的问题反馈。5.2 源表与目标 topic 的对账 SQL 写法手工抽检只能看个大概量级验证还得靠对账。思路是把目标 topic 当成一张 upsert 表读回来和源端聚合后的结果做比对。读回来用upsert-kafka作为 source这时它的语义是每条消息当作 upsertvalue 为 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 );对账时我一般做三个比对按严格程度递进比对项SQL 思路说明条数比对两边各自COUNT(*)后相减最快能发现大范围丢失状态分布比对按order_status分组计数后 FULL JOIN能发现删除或更新丢失金额汇总比对两边SUM(amount)相减容忍极小浮点误差能发现字段级错乱只要前两项对得上基本可以判断链路健康。第三项如果对不上但前两项正常优先怀疑精度问题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.policycompact或compact,delete否则同一个 key 的历史版本会全部保留读回来的数据量随时间线性膨胀最终不可用。我的经验配置是这样配置项建议值理由cleanup.policycompact按 key 保留最新版本是 upsert 语义的基础min.cleanable.dirty.ratio0.5 左右太低会频繁触发清理影响 IOsegment.ms按数据量通常 1 小时段文件太大清理不及时太小文件数爆炸retention.ms有业务回溯需求时另设纯 compact 会自动清理旧版本但需要保留删除标记一段时间关于 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 的处理约定这三件事并把这段注释同步给下游消费方。听上去是体力活但它挡掉的沟通成本远比写这几行字的时间高。这套东西本质上不是配置问题是上下游之间的一份契约契约写清楚了链路才真的稳。