ClickHouse 报表的对账和重算

背景#

报表表由物化视图在明细每次插入时增量聚合（建法在用物化视图做交易报表）。物化视图只看每次新插入的块，看不到之后的合并、FINAL 和改数据，所以明细表折得掉的重复，报表不一定折得掉。偏了就要能发现、能修，修的时候还不能丢掉迟到写入、不能拖慢写入。

结论主要来自 clickhouse-lab 的实验 24、25、26（25.3.14.14，三副本），窗口那部分另有实验 02。明细表 events 是 ReplicatedReplacingMergeTree(create_time)，报表表 rpt 是按 UTC 半小时分桶、按天分区的 ReplicatedAggregatingMergeTree，这是改过之后的报表链路，还没部署到 Aiven 上的现有集群。

哪些重复会让报表多算#

实验 25 造了三天、每天 30 万行的明细，所有写入都带 token，并且开着 deduplicate_blocks_in_dependent_materialized_views = 1（下一节细说）。然后按现有集群上见过的几种重复各注入一次，看报表偏了多少：

偏差的量和注入的行数、金额完全一致。超时重投如果超出了去重窗口，报表同样多算，依据是实验 02 和实验 24 第五节。明细这边按键折叠之后是对的，报表上的偏差只能靠重算改回来：实验 24 里先写一行、再换 token 重放、再重发金额从 100 改成 150 的新版本，明细 FINAL 读到 150，物化视图累加成 350，对明细 OPTIMIZE … FINAL、两张表都合并之后还是 350。

让物化视图跟着源表去重#

窗口能拦下的那种重投，可以让报表也不多算，办法是写入时带上 deduplicate_blocks_in_dependent_materialized_views = 1。实验 24 把它的边界一条条测了：

• 默认值 0 时，源表拦下了重投，物化视图照样又算一次，带不带 token 都一样。25.3 源码里这个设置的说明写的是「If an INSERTed block is skipped due to deduplication in the source table, there will be no insertion into attached materialized views」（Settings.cpp @ v25.3.13.19-lts），和实测相反。上游在 26.2 把默认值改成了 1（SettingsChangesHistory.cpp）。现有集群的 profile 没改这一项，还是默认的 0；sink 的 INSERT 有没有自己带上，还要核。
• 设成 1 要对所有写入一律生效。首写用 0、重投用 1，物化视图照样算两次，因为首写时目标表没有记下去重标记。所以要写进 sink 的 clickhouseSettings 或者 sink 用户的 profile。
• 物化视图目标表自己的去重窗口也要盖得住重投。lab 里把目标表的窗口压到 2，等它的裁剪线程把旧记录裁掉之后再重投，源表拦下了，报表还是多算了一次。调大窗口时，目标表要和明细表一起调，不能比明细表小。
• 在 25.3 上不能和 async_insert 一起开，两个都开时 INSERT 直接报 Code: 344，由 throw_if_deduplication_in_dependent_materialized_views_enabled_with_async_insert 控制。connector 默认 async_insert=0（ClickHouseSinkConfig.java:227），别在 clickhouseSettings 里打开它。这个开关在 26.1 起作废，26.2 起 async_insert 的默认值也改成了 1（SettingsChangesHistory.cpp），升级时这一条要重新看。
• 不会误伤：两个不同的源批次聚合出一模一样的块，设成 1 也都算进去了，带不带 token 都一样。目标表的去重标记是从源批次派生的，不按物化视图输出的内容算哈希。

它只管窗口内的重投，表里另外三行多算的情形它管不了，要靠下面的对账和重算。

按天对账#

报表按桶聚合，和「明细 FINAL 按同样的桶现算」用 FULL JOIN 逐桶比，只输出有对不上的桶的那几天。{from} 和 {from_ms} 是同一个时刻，一个写成时间、一个写成毫秒：

SELECT toYYYYMMDD(bucket) AS d,
       countIf(r_rows != t_rows OR r_amount != t_amount OR r_users != t_users) AS bad_buckets,
       toInt64(sum(r_rows)) - toInt64(sum(t_rows))                              AS rows_diff,
       sum(r_amount) - sum(t_amount)                                            AS amount_diff
FROM (SELECT bucket, tenant_id, sum(rows) AS r_rows, sum(amount) AS r_amount, uniqExactMerge(users) AS r_users
      FROM rpt WHERE bucket >= {from} GROUP BY bucket, tenant_id) AS r
FULL OUTER JOIN
     (SELECT toStartOfInterval(toDateTime(intDiv(settle_ms, 1000), 'UTC'), INTERVAL 30 MINUTE) AS bucket, tenant_id,
             count() AS t_rows, sum(amount) AS t_amount, uniqExact(user_id) AS t_users
      FROM events FINAL WHERE settle_ms >= {from_ms} GROUP BY bucket, tenant_id) AS t
USING (bucket, tenant_id)
GROUP BY d HAVING bad_buckets > 0 ORDER BY d;

实验 25 里这条查询只找出了被注入过重复的那两天（20260308、20260309），偏差的行数和金额也和注入的一致；人工补发过 500 行正常迟到数据的 20260307 不在其中。

现有集群上见过人工补发写进十几天前的分区，对账至少要覆盖最近两周，更早的分区会不会再被写还没查过。一次查一天（加上 _partition_id 条件），只查到昨天为止，带上 max_threads；明细 FINAL 在两三亿行的日分区上要多少资源，还没测过。

重算一天#

从明细 FINAL 把一天重算进临时表，再 REPLACE PARTITION 换进报表。思路沿用清理重复行的那套 runbook：记下快照时刻，加一道闸挡住快照之后的写入，再换分区，事后再核一次（REPLACE PARTITION runbook）。区别是报表的「快照」就是明细 FINAL，不用另建快照表。

-- 1. 临时表，结构和报表表一样（Replicated 库不写 ON CLUSTER）；重来时先删掉上一轮的
DROP TABLE IF EXISTS rpt_fix SYNC;
CREATE TABLE rpt_fix AS rpt;

-- 2. 记下时刻 t0，执行节点先把明细追平
SELECT now64(6);                     -- t0
SYSTEM SYNC REPLICA events;

-- 3. 从明细 FINAL 重算这一天，限两个线程；列名写全，不按位置对
INSERT INTO rpt_fix (bucket, tenant_id, rows, amount, users)
SELECT toStartOfInterval(toDateTime(intDiv(settle_ms, 1000), 'UTC'), INTERVAL 30 MINUTE) AS bucket, tenant_id,
       count(), sum(amount), uniqExactState(user_id)
FROM events FINAL WHERE _partition_id = '{D}' GROUP BY bucket, tenant_id
SETTINGS max_threads = 2;

-- 4. 闸：t0 之后明细这一天在任一副本上有没有新写进来的 part，不算被去重拦下的；
--    不是 0 就从第 1 步重来。t0 往前留 5 秒，抵消副本之间的时钟差
SYSTEM FLUSH LOGS ON CLUSTER default;
SELECT count() FROM clusterAllReplicas('default', system.part_log)
WHERE database = '{db}' AND table = 'events' AND partition_id = '{D}'
  AND event_type = 'NewPart' AND error = 0
  AND event_time_microseconds >= toDateTime64('{t0}', 6) - INTERVAL 5 SECOND;

-- 5. 换分区，等三个副本都换完
SELECT now64(6);                     -- t_swap
ALTER TABLE rpt REPLACE PARTITION ID '{D}' FROM rpt_fix SETTINGS alter_sync = 2;

-- 6. 再跑一次第 4 步，加上 AND event_time_microseconds < '{t_swap}'：
--    闸和换分区之间那一瞬间的写入只能事后查出来，不是 0 就重来
-- 7. DROP TABLE rpt_fix SYNC;

几处说明：

• 第 2、3 步放在同一个会话里跑，读的就是刚追平的这个副本。现有集群的连接随机落到某个节点，换了连接、落到还没追平的节点上，会漏读 t0 之前已经写进去、还没复制过来的行，第 4 步的闸只看 t0 之后的新 part，拦不住这种情况。
• 第 1 步的 CREATE TABLE … AS 和 runbook 的 R0 一样，前提是报表表的 Keeper 路径里带表自己的 uuid，查法见 runbook。
• 第 2 步的 SYSTEM SYNC REPLICA 和第 4 步的 SYSTEM FLUSH LOGS ON CLUSTER 都要有对应的权限，Aiven 文档没写给不给，执行前按在线换引擎那篇的办法确认。
• 第 6 步加上界，是因为换分区之后写进明细的行会经物化视图加进新换进来的分区，不会丢，不加上界会把它们也报出来，触发没必要的重来。这一条是推演，实验 25 换分区之后没有写入，没暴露出来。

实验 25 里重算偏了的两天之后，三个副本上报表和明细逐桶一致；同一天再重算一次，结果不变。

闸不能省。实验 25 第五节在临时表写好之后、换分区之前，往明细的这一天迟到写进 50 行，物化视图已经把它们加进了报表。不过闸直接换，报表里这 50 行被换分区抹掉了，这一天反而少了 50 行、金额少了 23939.59。过闸时同样的时机再来 50 行，闸拦下了，重来一次之后前后两批迟到的 100 行都在报表里。

闸有两个前提。一是 part_log 得开着：现有集群上 part_log 只留约 4 天，够一次重算用。二是 sink 正在写的那一天每次都过不了闸，要等这一天不再被写（按服务端时区过了零点）再修。第 4 步往前留的那 5 秒是推断，副本之间的时钟差在单机的 lab 上测不出来。

重算要限线程#

重算和写入跑在同一批节点上。实验 26 在 lab 上比了三种大查询，各自在不限、max_threads = 2、max_threads = 1 下跑，每种取三轮的中位数（M2 Pro 上 12 CPU 的虚拟机，一天 2000 万行）。前后跑了 5 次，下表是其中一次：

几条结论：

• max_threads = N 就不超过 N 个核，三种查询都是这样。
• 重算本来就不太并行，不限时也只用两个核左右，限到 2 之后慢 6% 到 39%：5 次运行里已经合并的那天慢 6% 到 16%，没合并的慢 34% 到 39%。
• 能并行的大聚合差别大，不限时用了 9 个多核，限到 2 慢 3.6 倍。
• 同一节点上 sink 那种小批写入的 p95，在不限线程的重算跑着时是 33 到 69 ms，限到 2 之后是 9 到 19 ms，空载时是 17 到 20 ms。5 次运行方向一致，但这组数的噪声在 10 ms 上下，只作记录。

max_threads 只管查询自己的线程，后台的 merge 和副本之间的复制用的是服务端自己的后台线程池，不归它管，这一句是按设置的作用范围推的。上面的倍数要在现有集群上另测，那里节点的核数和数据量都不一样。现有集群在 profile 里把 max_threads 钉在节点的 vCPU 数，设得更小可以，设得更大不行；每个节点的并发查询数也有上限，INSERT 算在里面，超出时报 Code: 202 TOO_MANY_SIMULTANEOUS_QUERIES。

小结#

• 物化视图看不到合并：窗口外的重放、producer 重试和重发新版本都会让报表多算，明细 OPTIMIZE … FINAL 也救不回来。
• deduplicate_blocks_in_dependent_materialized_views = 1 只管窗口内的重投，要对所有写入一律设，目标表的窗口不能比明细表小，25.3 上不能和 async_insert 一起开。25.3 默认是 0，上游 26.2 起默认是 1。
• 按天把报表和明细 FINAL 逐桶对账，lab 上准确找出了偏了的天和偏差的量。
• 重算一天：先删掉上一轮的临时表，从明细 FINAL 写临时表，查 part_log 过闸，再 REPLACE PARTITION，事后再过一次闸。不过闸会把重算期间的迟到写入从报表里抹掉。
• 重算带 max_threads = 2，在 lab 上慢 6% 到 39%；同节点写入的 p95 只作记录。

参考资料#

• Deduplicating inserts on retries：物化视图的去重要两个开关都允许
• Settings.cpp @ v25.3.13.19-lts：25.3 上 deduplicate_blocks_in_dependent_materialized_views 的说明
• SettingsChangesHistory.cpp @ v26.3.47.6-lts：26.2 起物化视图去重和 async_insert 的默认值都改为 true
• ClickHouseSinkConfig.java @ v1.3.9：connector 默认 async_insert=0
• ALTER TABLE … REPLACE PARTITION
• clickhouse-lab：实验 24（物化视图去重的边界）、25（对账和重算）、26（限线程）

相关文章#

• ClickHouse 交易明细系统的整体架构：系列总览和阅读顺序
• 用物化视图做 ClickHouse 交易报表
• ClickHouse 明细表的去重方案
• 清理 ClickHouse 重复行的 REPLACE PARTITION runbook