Kafka Engine ClickHouse: atomicity without losses

# Kafka Engine in ClickHouse: Ensuring Atomic Inserts Without Data Loss

Frequent individual inserts into MergeTree tables (10,000 INSERT/s) cause performance to drop to 59 rows/s. For ReplicatedMergeTree, it's down to 6 rows/s, and with quorum, just 2 rows/s. This is due to creating a multitude of small data parts, which are periodically merged based on the max_merge_selecting_sleep_ms setting. The merging process overloads CPU and I/O, reducing overall speed.

Buffer tables seem like a solution: data accumulates in RAM and is flushed to the target table in batches (by time, size, or row count). This minimizes the number of files and merge load. However, if ClickHouse crashes, unflushed data in memory is lost.

Buffer Tables: Mechanism and Risks

A buffer table uses the Buffer engine, accumulating data in memory until flush conditions are met: max_time_to_flush (e.g., 100 s), min_time_to_flush, max_rows_to_flush (1 million rows), max_bytes_to_flush (1 GB). The flush happens atomically into MergeTree, but only after writing to disk.

Key risk: If the server crashes (kill -9, OOM), the buffer contents in RAM are lost forever. There's no replication or WAL for buffers, making this approach unreliable for high-load streams.

Switching to Kafka Engine solves the issue: data is persistent in Kafka, and ClickHouse only commits the offset after successful processing.

Setting Up the Kafka Engine Pipeline

Simple stack: Kafka + ZooKeeper + ClickHouse in Docker. Create the test_topic_json topic with JSON messages:

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

In ClickHouse, the pipeline consists of three components:

• Kafka Engine table for reading:

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;

• Materialized view as consumer:

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;

• Target MergeTree:

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

Demonstrating Atomicity

After sending messages, the consumer offset for my-group4 is 2 (End=2). Now change the column type in message_json_parse to UInt64 — parsing the JSON string "gg" will fail:

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

Send {"message": "gg"}. The offset stays at 2 (End=3), ClickHouse log: Cannot parse string 'gg' as UInt64. The offset is committed only after successful write to MV → MergeTree.

ClickHouse reads the message into RAM but doesn't commit the offset on error. New messages are blocked until fixed. After correcting the type back to String:

• The "gg" message is inserted.
• Offset updates to 3.
• No data loss even after ClickHouse restart.

Key Points

• Atomicity: Kafka offset is committed post-insert into the materialized view, ensuring exactly-once semantics.
• Recovery: After failure/error, processing resumes from the last committed offset without duplicates or skips.
• Performance: Batch processing from Kafka minimizes small parts in MergeTree.
• Reliability: Kafka stores data persistently; ClickHouse acts as a stateless consumer.
• Limitations: Requires Kafka infrastructure; set kafka_skip_broken_messages=0 for strict handling.

Implementation Recommendations

• Use kafka_num_consumers >1 for parallelism.
• Set max_block_size=2MB+ for batching.
• Monitor offsets: SELECT * FROM system.kafka_consumers.
• For JSON: JSONEachRow with strict parsing.

— Editorial Team