背景#

这条链路是业务服务写 Kafka（单 topic、8 个分区，峰值每秒约 1 万条），Confluent Cloud 托管的 ClickHouse sink（clickhouse-kafka-connect v1.3.9）消费后写进 Aiven 上的三副本 ClickHouse 25.3。sink 是至少一次投递，没开 exactlyOnce；明细表是 ReplicatedMergeTree，ORDER BY (settle_ms, id, rev)，按结算日分区。

这条链路上的重复有三种，都是除写入时间 create_time 之外整行相同：

来源 见过的情形 特征
producer 重试 现有集群，一百多行 集中在一个 45 分钟的窗口里，每个键正好 2 份，broker 里存了两条 offset 不同的消息
sink 超时重投 现有集群，两百多行 同一批原样再写一次，间隔 32 到 124 秒
sink 重启重放 lab 上造过（实验 21） task 重启后从上一个提交点重读，批次边界会变

人工补发是第四个入口：发送超时的记录事后补发时，其中一部分可能早已写进去。

超时重投怎么产生、为什么没被服务端拦下，见超时重投和块级去重窗口；producer 重试和重启重放见四个入口。这篇回答的是选了什么方案：明细表在对账口径上没有重复，下游报表不因为重复多算，出了重复能发现、能修。

方案在 clickhouse-lab 的三副本实验台上验过（实验 02、03、14、18、21、23、24、25、27、29），还没在现有集群上执行。

为什么不靠 sink 的 exactlyOnce#

官方 sink 有一个 exactlyOnce 开关，按 offset 区间加一张 KeeperMap 状态表判断一批该不该写。lab 上用真的 Kafka Connect 测过两件事（实验 21、23）：

  • 现有集群上 234 行那一种「同一批原样重投」，它管不住。INSERT 在服务端提交了、客户端超时，状态停在 BEFORE_PROCESSING（插入结果不明），同一区间再来时源码的处理就是再写一遍，交给 ClickHouse 的块级去重拦，开不开结果一样。
  • 它防得住批次边界变了的重放，但打开之后，worker 崩溃重启时状态常常对不上，默认配置下 task 直接停住；配了 errors.tolerance=all 而没配 DLQ 时，有一段数据会丢。

细节写在四个入口那篇。根子在 ClickHouse 没法把 Kafka 的 offset 和数据放进同一个事务提交：它的事务还是实验特性，只支持 Atomic 库里不带复制的 MergeTree（Transactional (ACID) support）。所以官方 connector 和 ClickPipes 的 exactly-once 做的是同一件事：记下每一批的 offset 区间，失败后按原区间重放、带同一个 token，交给块级去重拦下。ClickPipes 的 token 是 topic:partition:firstOffset-lastOffset，文档写明 token 出了去重窗口之后再重放「can be inserted again」，并建议按最坏的重放延迟调大窗口（ClickPipes for Kafka best practices）。开不开 exactlyOnce，同一批的重投都要靠窗口兜着。

同类系统怎么取舍#

ClickHouse 本身不提供「只写一次」的开关，官方去重策略页列的是 ReplacingMergeTree 和两种 Collapsing 引擎，去重在合并时发生，任何时刻表里都可能还有重复（Deduplication strategies）。能查到的几种做法按去重放在哪一层分开看：

  • 接受至少一次，靠表引擎按键去重，接受最终一致。Sentry 的 Snuba 文档原话是 consumer 保证每批至少一次交给 ClickHouse，选对表引擎去重，「we can achieve exactly once semantics if we accept eventual consistency」（Snuba architecture）。PostHog 用 ReplicatedReplacingMergeTree，同时提醒去重不是保证（PostHog 的 ClickHouse 说明）。
  • 重复直接等于多收钱的，写入之前就去重。OpenMeter 做用量计费，自己写了消费端（原因之一是一条坏记录会让 Kafka Connect 整批进 DLQ），先在批内去重，再查 Redis 去掉见过的事件，然后才写 ClickHouse（OpenMeter，2023-10）。代价是自己养一套消费端和去重存储。
  • 目标库能把 offset 和数据一起提交，才是真正的 exactly-once。Snowflake 的 Snowpipe Streaming 把 offset token 和数据在同一个 channel 里提交（Snowpipe Streaming channels）。ClickHouse 不在这一类。

取舍看三件事：目标库能不能原子地提交 offset；重复的代价多大，分析类容得下短时间的重复，计费类容不下；批次边界在不在自己手里，托管 connector 不在。这套是 ClickHouse 加托管 connector，用途是分析和报表，所以走「至少一次 + 按键折叠」。

这里有一件要业务拍板的事：报表金额能不能接受最终一致。明细按键折叠之后，多出来的那份在合并或 FINAL 之后就看不到了；报表是增量累加的，窗口外的重复会先让它多算，等对账、重算之后才改回来。要求报表金额任何时刻都不能多算，就得走写入前去重，这个方案没有设计这一条。

每一层管得住哪种重复#

层 做法 sink 超时重投 sink 重启重放 producer 重试
入口 块级去重加 token，窗口调大；或改用异步插入，token 记进 async_blocks/ 窗口够大就拦得住 拦不住，token 变了 拦不住，offset 和 token 都不同
入口 producer 幂等 — — 按默认已开；防得住会话内的重试，超时报失败之后应用再发的那类防不住
存储加查询 ReplacingMergeTree 加 FINAL 能 能 能
派生数据 物化视图跟着去重；报表能重算 物化视图去重能拦 只能靠重算 只能靠重算
修复 REPLACE PARTITION runbook 事后清 事后清 事后清

只有按键折叠那一层同时管住三种，所以它是方案的主体，入口那几层是尽力而为的第一道过滤。按键折叠管不到的是派生数据：物化视图在插入时就触发，看不到后来的合并。

明细表换成 ReplicatedReplacingMergeTree，键不变#

排序键 (settle_ms, id, rev) 本身就是业务键。三种重复都是除 create_time 外整行相同，排序键相同，合并时会折成一行；人工补发的迟到数据按 settle_ms 落进原来那个日分区，同样会被折叠。没合并之前的重复，读的时候带 FINAL 或在 profile 里设 final = 1（Settings.cpp）就看不到。lab 上一千万行混进 50 万行重投，物理行数 1050 万，FINAL 读出来正好 1000 万（实验 14）。官方博客讲这类引擎时提醒过一条限制：ReplacingMergeTree 只能按排序键认定同一行，排序键要变就得重建表（How we built fast UPDATEs – Part 1）。

FINAL 是要付读代价的。实验 14 测到它跟着重叠范围走，按排序键范围查大约贵一倍，比查询里手写 GROUP BY 去重省一个数量级以上的内存。后来在 25.8 上用 3 亿行的单日分区又测了一轮：全分区 count() FINAL 的峰值内存是 149 到 336 MiB，精确去重要约 18 GiB（实验 30–34）。具体数字和不能外推的部分在查询路径那篇。合并是异步的，没合并的重复在不带 FINAL 的读里一直能看到。

迁移的路子是 ATTACH PARTITION … FROM 把旧表的分区挂进新表（本地盘上是硬链接），停 sink、排空、补齐之后用 EXCHANGE TABLES 原子切换，迁移前就有的重复在新表上带 FINAL 读时被折掉了（实验 27）。整套步骤和回滚写在在线换引擎那篇。

留下哪一份#

这一项要和报表的使用方一起定。create_time 是写入时由默认值 now() 填的，三种重复里后到的那份 create_time 不早于先到的；producer 重试那批约三分之二是同一秒写进来的，两份相等。用 create_time 做版本列，留下的是版本最大的那份，版本相同时留最后写入的；不带版本列，留下的是最后写入的那份（ReplacingMergeTree 的 ver 参数）。两种规则都留下后到的那一份。

报表按 settle_ms 分桶时，留哪一份不影响报表，重复的几份落在同一个桶里。报表还按 create_time 的窗口切的话，这一行最后会算进更晚的窗口；要它算在第一次写入的窗口，得加一列随时间递减的版本，旧表也要先加上这一列才能 ATTACH。

入口：调大去重窗口，核实 producer 幂等#

块级去重的窗口在 25.3 上默认是 1000 个块。窗口至少要等于「建块速率 × 最长的重投间隔」（机制见实验 02）：一轮超时重投最长约 90 秒，现有集群上见过 124 秒，是两轮。两亿行那套部署每秒约 124 个块，124 × 124 ≈ 1.5 万，方案取 2 万；三亿多行那套每秒约 40 个块，取 1 万。物化视图的目标表也要一起调，不能比明细表小（实验 24，原因在报表对账那篇）。

ALTER TABLE events MODIFY SETTING replicated_deduplication_window = 20000;

窗口只管 sink 超时重投这一种。代价在 Keeper：每张表多存这么多个块节点，裁剪线程每轮多删这么多。上游在 25.9 已经把默认值提到了 1 万，对两亿行那套仍然不够（块级去重窗口那篇）。

producer 重试那一种，producer 端的幂等只能挡一部分。这条链路的 producer 用的是 kafka-clients 3.4.0，没有覆盖 acks 和 enable.idempotence，按默认幂等是开着的，重复照样出现了：幂等只管一个 producer 会话内部的重试，超过 delivery.timeout.ms 报失败的那条写没写进 broker 不确定，应用再把它发一次，就不在幂等的范围里（四个入口那篇）。所以这一项做的是核实：enable.idempotence=true、acks=all、max.in.flight.requests.per.connection 不超过 5，没有冲突的配置把幂等关掉。Kafka 的文档写明，设了冲突的配置、又没有显式打开时，幂等会被关掉（producer 配置）。超时之后应用层的重发，还是要靠存储层按键折叠。

服务端异步插入：三个开关要一起开#

源表上没有靠 deduplicate_blocks_in_dependent_materialized_views 去重的物化视图时（原因见下一节），入口还能让服务端把 sink 的小 INSERT 攒成大 part。sink 每次 poll 按分区各发一条十几行的 INSERT，每条都是一个新 part、一条复制日志；开了异步插入，同一个刷新窗口里的 INSERT 先进服务端的缓冲区，到时间或攒够大小再一起写（Asynchronous inserts）。实验 29 在 25.3 的三副本上，20 条并发的小 INSERT 合成了 1 个 part，复制日志只多了 1 条。

sink 的每条 INSERT 都带 insert_deduplication_token，异步插入要不要按它去重，由 async_insert_deduplicate 决定，25.3 上默认是 0（Settings.cpp:1717）。只开 async_insert = 1 时，一个刷新桶落盘只拿桶里第一条 INSERT 的 token 去 blocks/ 查重、记账，后面几条的 token 被丢掉。实验 29 第一段测出两个后果：

  • 桶里第二条 INSERT 被原样重投，blocks/ 里没有它的 token，第二份照样写进表里。
  • 重投的那条正好第一个进桶时，同桶里其他全新的 INSERT 跟着它一起被判成重复，一行也没写进去。

第二种情况服务端照常返回成功，sink 不会再发，这些行就丢了。三个开关一起开时，实验里两种情况都没再出现：

SETTINGS async_insert = 1, wait_for_async_insert = 1, async_insert_deduplicate = 1

这时每条 INSERT 的 token 各自记进 Keeper 的 async_blocks/。实验 29 第二段造了三种重投，都拦住了：同一个桶里出现两条相同的 token，只写进一份；第一次和别的 INSERT 合批写入之后，单独重投到另一个副本；旧 token 的重投和新 token 的首投落进同一个桶，旧的被剔除，新的照常写入。桶里有一条数据解析出错时，只有它自己报 Code: 27 回滚。wait_for_async_insert = 1 让 INSERT 等数据写进表里才返回，sink 收到成功时数据已经落盘，和同步插入一样。

用的时候还有三处要注意：

  • token 要放在请求的设置里传，sink 本来就是这么传的（ClickHouseWriter.java:1000-1003）。写成 SQL 里的 SETTINGS insert_deduplication_token = '…'，每条 INSERT 的文本都不一样，服务端按 INSERT 的文本分桶，攒批就失效了，实验里两条拆成了 2 个 part。
  • 同步插入的 token 记在 blocks/，异步的记在 async_blocks/，两边互不查对方。切换之前同步写进去的一批，切换之后重投不会被认出来，反过来也一样，所以开关要挑没有重投积压的时候切。
  • 异步这边的窗口是另一个设置 replicated_deduplication_window_for_async_inserts，25.3 默认 10000（MergeTreeSettings.cpp:150），按 async_blocks/ 下的节点数裁剪，一条 INSERT 的 token 占一个节点。sink 发 INSERT 的次数没变，按两亿行那套每秒约 124 条推算，10000 条只盖得住 80 秒左右，90 秒以上的重投照样漏过去，这个窗口也要按前面同步窗口的算法调大。

实验 29 在 25.8.33.6 上复跑过，默认值和结果都一样。实验里这几个设置是放在 HTTP 参数里传的；落到现有集群上，打算在 sink 用的账号上用 ALTER USER … SETTINGS 下发，sink 那边不用改，这一步还没验过，Aiven 上能不能改用户级设置也要先核。

派生数据：物化视图跟着去重，报表能重算#

ReplacingMergeTree 救不了物化视图喂的报表。lab 上同一行先正常写一次，再换 token 重放一次（模拟窗口外的重放），再按 ReplacingMergeTree 的惯例重发一条新版本、金额从 100 改成 150：明细 FINAL 读到的是 150，物化视图累加成 100 + 100 + 150 = 350，两张表都合并之后还是 350（实验 24）。所以派生数据要两样一起上：

  • 在 sink 的 clickhouseSettings 里加 deduplicate_blocks_in_dependent_materialized_views=1，让窗口拦下的重投不再在物化视图里多算一次。25.3 默认是 0，这时源表拦下了重投，物化视图照样又算一次。
  • 窗口外的重复和改数据，物化视图一定会多算，靠按天对账、从明细 FINAL 重算、过闸之后换分区来修（实验 25）。

这两条都有前提和坑，比如这个设置要对所有写入一律生效、在 25.3 上不能和 async_insert 一起开（两个都开时 INSERT 报 Code: 344），重算不过闸会抹掉迟到写入，都写在报表的对账和重算。还有一个前提要先核：现有集群上的下游聚合是不是物化视图喂的。如果是定时跑的 INSERT … SELECT，第一条没用，聚合读到的是明细表里已经存在的重复；这时也就没有和 async_insert 冲突的设置，入口可以改用上一节的异步插入。

发现和修复#

对账作业常态化：按小时切窗，算 count() - uniqExact(id, settle_ms, rev)，口径只用精确的 uniqExact。同时盯 system.part_log 里 event_type = 'NewPart' AND error = 389 的条数，它就是窗口拦下了多少次重投（实验 03）；条数突增说明有重投在发生，窗口拦不住的那些不会出现在这里，要靠对账发现。

已经写进去的历史重复，用快照加 REPLACE PARTITION 的 runbook 清（REPLACE PARTITION runbook），清完之后受影响的报表按天重算。

不当常规手段的几种做法#

做法 原因
sink 开 exactlyOnce 见上文「为什么不靠 sink 的 exactlyOnce」
查询里手写 GROUP BY、LIMIT 1 BY 去重 比 FINAL 更贵，内存差一个数量级以上（实验 14）；内存跟着键数涨，3 亿行的分区上精确去重约 18 GiB，4 vCPU / 16 GB 的节点在 2 亿行时就超限（实验 31–33）
定期 OPTIMIZE … FINAL DEDUPLICATE BY … 每次重写整个分区；同一组留下哪一行由合并决定，执行前看不到结果，也没有回滚
定期轻量删除 25.3 上必须带 IN PARTITION；按 create_time 删分不开同一秒写进来的两份；没有回滚（实验 18，见轻量删除那篇）

实施顺序#

  1. 先只读地核对这套系统：下游聚合是物化视图还是定时作业，sink 的 INSERT 现在带没带物化视图去重的设置，窗口有没有被改过，producer 的配置，Aiven 上能不能 MODIFY SETTING。再和报表的使用方定两件事：报表金额能不能接受最终一致，桶按 settle_ms 还是 create_time 切。
  2. 低成本、能马上见效的：物化视图跟着去重、核实 producer 幂等、对账和告警；调大窗口（明细表和物化视图的目标表一起调）也放在这一步。下游不是物化视图的话，物化视图那一条换成异步插入。
  3. 结构性的：换引擎和报表重算一起做，先定留下哪一份。
  4. 历史重复用 runbook 清，带上重算报表那一步。

小结#

  • 这条链路只能做到至少一次，sink 的 exactlyOnce 管不住同一批原样重投（234 行那一种），打开之后还会引入状态冲突，所以不开。
  • 方案的主体是把明细表换成 ReplicatedReplacingMergeTree，键不变，读的时候带 FINAL 或设 final = 1，三种重复都能折掉。
  • 入口把块级去重窗口调到盖得住一两轮重投、核实 producer 幂等，定位是尽力而为的第一道过滤。源表上没有靠去重的物化视图时，也可以开服务端异步插入攒批，async_insert_deduplicate 要一起开。
  • 物化视图看不到合并：设置物化视图跟着去重，窗口外的重复靠对账和重算修。报表金额只能做到最终一致，这一点要业务确认。

参考资料#

相关文章#