Kafka Engine ClickHouse: Atomizität ohne Verluste

Kafka-Engine in ClickHouse: Atomare Einfügungen ohne Datenverlust sicherstellen

Häufige einzelne Einfügungen in MergeTree-Tabellen (10.000 INSERT/s) lassen die Performance auf 59 Zeilen/s sinken. Bei ReplicatedMergeTree sind es nur 6 Zeilen/s, und mit Quorum sogar nur 2 Zeilen/s. Der Grund ist die Erzeugung unzähliger kleiner Datenparts, die periodisch basierend auf der Einstellung max_merge_selecting_sleep_ms zusammengeführt werden. Dieser Merge-Prozess überlastet CPU und I/O und reduziert die Gesamtgeschwindigkeit.

Buffer-Tabellen scheinen eine Lösung zu sein: Die Daten sammeln sich im RAM an und werden in Batches (nach Zeit, Größe oder Zeilenanzahl) in die Zieltabelle geschrieben. Das minimiert die Anzahl der Dateien und die Merge-Last. Allerdings gehen bei einem Absturz von ClickHouse nicht geschriebene Daten im Speicher verloren.

Buffer-Tabellen: Mechanismus und Risiken

Eine Buffer-Tabelle verwendet den Buffer-Engine, der Daten im Speicher ansammelt, bis Flush-Bedingungen erfüllt sind: max_time_to_flush (z. B. 100 s), min_time_to_flush, max_rows_to_flush (1 Million Zeilen), max_bytes_to_flush (1 GB). Der Flush erfolgt atomar in MergeTree, aber erst nach dem Schreiben auf Disk.

Hauptrisiko: Bei Serverabsturz (kill -9, OOM) gehen die Buffer-Inhalte im RAM unwiderruflich verloren. Es gibt keine Replikation oder WAL für Buffer, was diesen Ansatz für hochbelastete Streams unzuverlässig macht.

Der Wechsel zur Kafka-Engine löst das Problem: Die Daten sind persistent in Kafka, und ClickHouse commitet den Offset erst nach erfolgreicher Verarbeitung.

Einrichtung der Kafka-Engine-Pipeline

Einfacher Stack: Kafka + ZooKeeper + ClickHouse in Docker. Erstellen Sie das test_topic_json-Topic mit JSON-Nachrichten:

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

In ClickHouse besteht die Pipeline aus drei Komponenten:

• Kafka-Engine-Tabelle zum Lesen:

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;

• Materialisierte View als 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;

• Zieltabelle MergeTree:

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

Demonstration der Atomicität

Nach dem Senden der Nachrichten beträgt der Consumer-Offset für my-group4 2 (End=2). Ändern Sie nun den Spaltentyp in message_json_parse zu UInt64 – das Parsen der JSON-Zeichenkette „gg“ schlägt fehl:

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

Senden Sie {"message": "gg"}. Der Offset bleibt bei 2 (End=3), ClickHouse-Log: Cannot parse string 'gg' as UInt64. Der Offset wird erst nach erfolgreichem Schreiben in MV → MergeTree committed.

ClickHouse liest die Nachricht in den RAM, commitet den Offset bei Fehlern aber nicht. Neue Nachrichten werden blockiert, bis der Fehler behoben ist. Nach Korrektur des Typs zurück zu String:

• Die „gg“-Nachricht wird eingefügt.
• Offset aktualisiert sich auf 3.
• Kein Datenverlust selbst nach ClickHouse-Neustart.

Wichtige Punkte

• Atomicität: Der Kafka-Offset wird nach dem Einfügen in die materialisierte View committed, was exactly-once-Semantik gewährleistet.
• Wiederherstellung: Nach Fehler/Ausfall wird die Verarbeitung ab dem letzten committed Offset fortgesetzt, ohne Duplikate oder Lücken.
• Performance: Batch-Verarbeitung aus Kafka minimiert kleine Parts in MergeTree.
• Zuverlässigkeit: Kafka speichert Daten persistent; ClickHouse agiert als stateless Consumer.
• Einschränkungen: Erfordert Kafka-Infrastruktur; setzen Sie kafka_skip_broken_messages=0 für strenge Handhabung.

Implementierungsempfehlungen

• Verwenden Sie kafka_num_consumers >1 für Parallelität.
• Setzen Sie max_block_size=2MB+ für Batching.
• Überwachen Sie Offsets: SELECT * FROM system.kafka_consumers.
• Für JSON: JSONEachRow mit strict Parsing.

— Editorial Team