为什么不用 Kafka engine，也不写分布式表
2167 字 · 约 5 分钟
经常有人问我两个问题。ClickHouse 自带 Kafka 表引擎，为什么还要在外面再做一个导入工具？就算用外部工具，为什么不直接写分布式表，让 ClickHouse 自己分发，非要去写本地表？
这两个问题看着不相干，其实问的是同一件事：Kafka 到 ClickHouse 这条写入路径，该由谁来控制。
需求到底是什么
先把要解决的问题说清楚，不然很容易变成比谁的功能多。
日志和指标场景下，Kafka 到 ClickHouse 这条链路要同时做到几件事。一是高吞吐的大批量写入，MergeTree 的性能来自 merge，每次 INSERT 产生一个 part，批次一小，part 数就顶上去了，merge 压力跟着涨，官方建议一次至少 1000 行，最好在 1 万到 10 万行之间。二是数据在集群里落得均衡，不能某个节点的磁盘先满。三是能按业务语义路由，去重字段、join key、聚合维度要落在同一个分片上，这件事为什么重要，我在 扩容之后为什么不自动做数据均衡 里讲过。四是副本故障时写入不中断，分片里某个副本挂了，能自动切到同分片的其他副本。五是至少一次语义，数据确认写进去了才提交 offset。
这五条里，第二、三、四条官方 Kafka engine 做不了，它压根没有这个能力。
Kafka engine 的代价
多了一跳，还不归你管
在 ClickHouse 里，engine 决定了一张表的数据怎么组织，直接影响读写能力。稀疏索引、分区裁剪、merge、TTL 分层，这些全在 MergeTree 家族上。
用 Kafka engine，数据得先进一张 Kafka 表，再靠物化视图转进 MergeTree。多了一跳，而且这一跳的行为你控制不了。sinker 直接写 MergeTree 的本地表，数据一步到位。
出了问题不好查
Kafka engine 底层用的是 librdkafka。库本身质量不差，可它是嵌在 server 进程里的第三方组件。出了问题，日志和 server 的其他日志混在一起；消费行为能看到多少，取决于 engine 暴露了什么；想调一个 librdkafka 的参数，还得看 ClickHouse 有没有把它透出来。
懂的都懂，不出问题还好，一旦出了问题，基本没法有效排查。外部工具就不一样了，它是个独立进程，日志、指标都是自己的，可以单独重启，可以上 pprof。
数据落在哪，由 Kafka 说了算
Kafka engine 部署在每个节点上，各自消费分到的分区。数据最后落在哪个节点，取决于 Kafka 的分区分配，跟数据本身没关系。Kafka 各分区的数据量不均，ClickHouse 节点就跟着不均，均衡与否全看运气。按业务语义路由更是无从谈起，你没办法让同一个去重 key 永远落在同一个分片上。
节点挂了也是一样的道理。某个节点上的 consumer 没了，Kafka 会把它的分区 rebalance 给别的 consumer，数据就这么落到了别的节点、别的分片上。写入倒是没停，可分片内副本之间的切换这个概念在这里根本不存在，数据去哪还是 Kafka 说了算。
sinker 的配置里 hosts 是个两层数组，外层是 shard，内层是 replica。某个副本不可用了，就切到同一个分片里的其他副本，数据还是写进它该去的那个分片。
为什么写本地表，不写分布式表
就算不用 Kafka engine，也还有一个选择：外部工具写 Distributed 表，让 ClickHouse 自己去分发。我也不这么干。
先看分布式表 INSERT 的时候到底发生了什么。默认是异步模式（distributed_foreground_insert = 0，老版本里叫 insert_distributed_sync）：
① 数据先全量落到「接收节点」的本地临时目录
data/<db>/<table>/shard1_replica1/ 等，按分片键拆开
② 后台线程异步把各分片的数据发送到对应节点
③ 每个分片收到后，若是 ReplicatedMergeTree，
再经 Keeper 在副本间复制三跳，不是一跳。代价有三个。
一是节点之间磁盘用得不均。接收节点得先把全部数据接下来，再往外分发，分发要时间，数据量一大，很容易某一个节点的磁盘占用异常地高。轮着换接收节点能缓解，可高峰那一刻的压力，照样全压在当班的那个节点上。
二是 Keeper 的压力。批次越小，part 越多，每个 part 都要走一遍 Keeper 的复制协调。写入量大、批次又小的时候，Keeper 会忙得不可开交，某次操作 30 秒没响应，会话一断，表就进了 READ_ONLY，整张表都写不进去。Keeper 的负载是 ClickHouse 集群里最容易被低估的，我在 126 万 znode 是怎么长出来的 里写过一个完整的生产案例。
三是内网带宽。数据在集群里多走一跳，占的是内网带宽，而查询时分片之间交换数据用的也是同一条内网。写入挤占查询带宽，这种事压测里很难暴露，到了生产高峰才冒出来。
写本地表的做法是，外部工具在写入之前就算好每一行该去哪个分片，然后直连那个分片的节点写本地表：
sinker 算好分片 → 直连目标分片写本地表 → 分片内副本复制没有接收节点，也就没有中转。磁盘不均和内网带宽这两个问题直接没了；Keeper 的压力因为批次大（bufferSize 默认 26 万行）也缓解了一大截。
写本地表自己的代价
上面说的都是别人的代价，sinker 这条路也有它的。
路由和均衡得自己扛。sinker 把 sharding 规则拿到了写入侧，好处是能按业务语义路由，可这个规则从此就归你维护了。分片数一变，路由规则跟着变，存量数据得 rehash，这是个很重的操作。规则写错了，数据就分布错了，还不报错。
指定了 shardingKey 以后，故障域也变了。一个分片里的副本要是全挂了，路由到这个分片的数据就永远写不进去，sinker 会一直失败，直到集群恢复。没指定 shardingKey 的时候（按 offset 均摊），理论上还能把这批数据改投到别的分片；指定了就不行，改投就等于把路由规则破坏了。按业务语义路由换来的查询优化，是拿可用性从集群级降到分片级换的。
还多了一个要运维的组件。sinker 是独立进程，要部署、要监控、要升级、要管配置（本地文件或者 Nacos 托管）。Kafka engine 起码不用操这些心，它就在 server 里。一套新组件就是一套新的故障面，这个账得算进去。
我会选什么
说这么多，得交代一句：我是 clickhouse_sinker 的主要维护者，推荐它就是在推荐我自己维护的东西，你读的时候可以打个折。
规模化的日志和指标写入，我选外部工具写本地表。开头那五条需求里，有三条官方 engine 做不到，而这三条恰恰都是规模上去以后才要命的。
不过也有我会改主意的时候：
| 情况 | 选择 |
|---|---|
| 集群只有一个分片 | Kafka engine 或分布式表都行。均衡、路由、分发这三个问题全都不存在，多一个组件纯属负担 |
| 写入量小、批次天然就大 | 分布式表够用。Keeper 压力和磁盘不均都出不来，让 ClickHouse 自己分发省事得多 |
| 团队没能力维护额外的组件 | 用 engine。一个跑不稳的外部工具，比一个能力有限但简单的内建方案更糟 |
| 数据没有业务语义上的分片需求 | 差别小了很多。这时候选外部工具，理由就只剩可观测性和排障方便，不再是「做不到」 |
反过来说，多分片、写入量大、有按业务语义分片的需求、团队又养得起一个额外组件，这几条都占了，写本地表才明显更好。少一条，就得重新算账。