Kafka Engine ClickHouse：无损失原子性

ClickHouse 中的 Kafka 引擎：确保无数据丢失的原子插入

频繁向 MergeTree 表执行单个插入（每秒 10,000 次 INSERT）会导致性能下降至每秒 59 行。对于 ReplicatedMergeTree，降至每秒 6 行，使用 quorum 时仅为每秒 2 行。这是由于产生了大量小数据部分，这些部分会根据 max_merge_selecting_sleep_ms 设置定期合并。合并过程会过度占用 CPU 和 I/O，从而降低整体速度。

Buffer 表看似是一个解决方案：数据在内存中累积，然后按时间、大小或行数批量刷新到目标表。这可以最小化文件数量和合并负载。然而，如果 ClickHouse 崩溃，内存中未刷新的数据就会丢失。

Buffer 表：机制与风险

Buffer 表使用 Buffer 引擎，在内存中累积数据，直到满足刷新条件：max_time_to_flush（例如 100 秒）、min_time_to_flush、max_rows_to_flush（100 万行）、max_bytes_to_flush（1 GB）。刷新会原子性地写入 MergeTree，但仅在写入磁盘后完成。

关键风险： 如果服务器崩溃（kill -9、OOM），内存中的 Buffer 内容将永久丢失。Buffer 没有复制或 WAL，这使得该方法不适合高负载流。

切换到 Kafka 引擎即可解决问题：数据在 Kafka 中持久化，ClickHouse 仅在成功处理后提交 offset。

设置 Kafka 引擎管道

简单栈：Kafka + ZooKeeper + ClickHouse 在 Docker 中。创建 test_topic_json 主题，包含 JSON 消息：

{"message": "qq"}
{"message": "bb"}

在 ClickHouse 中，管道由三个组件组成：

• 用于读取的 Kafka 引擎表：

drop table if exists test_kafka_json_parse;
create table test_kafka_json_parse
(
    message String
)
engine = Kafka
SETTINGS kafka_broker_list = 'kafka:29092'
    , kafka_topic_list = 'test_topic_json'
    , kafka_group_name = 'my-group4'
    , kafka_format = 'JSONEachRow'
    , kafka_max_block_size = 2097152;

• 作为消费者的物化视图：

drop view if exists test_kafka_json_parse_view;
create materialized view test_kafka_json_parse_view
TO default.message_json_parse
(
 `message` String
)
AS SELECT message FROM test_kafka_json_parse;

• 目标 MergeTree：

create table message_json_parse
(
    message String
)
engine = MergeTree
order by message;

演示原子性

发送消息后，my-group4 的消费者 offset 为 2（End=2）。现在将 message_json_parse 中的列类型改为 UInt64 —— 解析 JSON 字符串 "gg" 将失败：

drop table if exists message_json_parse;
create table message_json_parse
(
    message UInt64
)
engine = MergeTree
order by message;

发送 {"message": "gg"}。offset 保持在 2（End=3），ClickHouse 日志：Cannot parse string 'gg' as UInt64。offset 仅在成功写入 MV → MergeTree 后提交。

ClickHouse 将消息读入内存，但在出错时不提交 offset。新消息将被阻塞，直到问题修复。类型改回 String 后：

• "gg" 消息被插入。
• offset 更新为 3。
• 即使 ClickHouse 重启，也无数据丢失。

关键点

• 原子性： Kafka offset 在插入物化视图后提交，确保精确一次语义。
• 恢复： 故障/错误后，从最后一个已提交的 offset 恢复处理，无重复或跳过。
• 性能： 从 Kafka 批量处理，最小化 MergeTree 中的小数据部分。
• 可靠性： Kafka 持久存储数据；ClickHouse 作为无状态消费者。
• 限制： 需要 Kafka 基础设施；为严格处理设置 kafka_skip_broken_messages=0。

实现推荐

• 使用 kafka_num_consumers >1 以实现并行。
• 设置 max_block_size=2MB+ 以进行批量处理。
• 监控 offset：SELECT * FROM system.kafka_consumers。
• 对于 JSON：使用 JSONEachRow 并严格解析。

— Editorial Team