背景#

这条链路是业务服务写 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 重启重放 几万行 几分钟内的一段 offset 整段重放，批次边界会变

人工补发是第四个入口，那次补发前先按键查过 ClickHouse，没有补出新的重复。

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

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

为什么不靠 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 变了 拦不住，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）。

FINAL 是要付读代价的。实验 14 测到它跟着重叠范围走，按排序键范围查大约贵一倍，比查询里手写 GROUP BY 去重省一个数量级以上的内存，具体数字在查询路径那篇。合并是异步的，没合并的重复在不带 FINAL 的读里一直能看到。超过 45 天、已经搬到对象存储上的老分区不太再合并，补发进去的重复可能长期留在物理层，靠 FINAL 读时折叠，这一点是推断，还没验过。

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

留下哪一份#

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

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

入口：调大去重窗口，核实 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 配置）。超时之后应用层的重发，还是要靠存储层按键折叠。这套系统里 producer 的实际配置还要核。

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

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）。

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

发现和修复#

对账作业常态化：按小时切窗，算 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）
定期 OPTIMIZE … FINAL DEDUPLICATE BY … 每次重写整个分区；同一组留下哪一行由合并决定，执行前看不到结果，也没有回滚
定期轻量删除 25.3 上必须带 IN PARTITION；按 create_time 删分不开同一秒写进来的两份；没有回滚（实验 18，见轻量删除那篇）

实施顺序和还没验的#

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

除了上面标出的两处（老分区还合不合并、递减的版本列），lab 上的机制都验过了，没验的是量级：调到 2 万之后 Keeper 的负担，宽表上 FINAL 的代价，迁移一天、重算一天要多少内存和多久，以及分区在对象存储上时 ATTACH 和 REPLACE 还走不走硬链接。这些要在更大的机器上按真实量级再跑一遍。

小结#

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

参考资料#

相关文章#