Kafka Engine ClickHouse: atomarita bez ztrát

Kafka Engine v ClickHouse: záruka atomarity vložení bez ztráty dat

Časté jednotlivé vložení do tabulek MergeTree (10 000 INSERT/s) vede k poklesu výkonu na 59 řádků/s, v ReplicatedMergeTree na 6 řádků/s, s kvórem na 2 řádky/s. To je způsobeno vytvářením mnoha malých částí dat, které se periodicky slučují (merge) podle nastavení max_merge_selecting_sleep_ms. Proces slučování přetěžuje CPU a I/O, čímž snižuje celkovou rychlost.

Bufferové tabulky se zdají být řešením: data se hromadí v RAM a flushují se do cílové tabulky dávkami (podle času, objemu nebo počtu řádků). To minimalizuje počet souborů a zátěž na merge. Nicméně při pádu ClickHouse se nesflushovaná data z paměti ztratí.

Bufferové tabulky: mechanismus a rizika

Bufferová tabulka používá engine Buffer, hromadí data v paměti do podmínek flush: max_time_to_flush (např. 100 s), min_time_to_flush, max_rows_to_flush (1 milion řádků), max_bytes_to_flush (1 GB). Flush probíhá atomárně do MergeTree, ale až po zápisu na disk.

Klíčové riziko: při krachu serveru (kill -9, OOM) obsah bufferu v RAM zmizí bez možnosti obnovy. Neexistuje replikace ani WAL pro buffery, což činí tento přístup nespolehlivým pro streamy s vysokou zátěží.

Přechod na Kafka Engine tento problém řeší: data jsou persistentní v Kafka, ClickHouse commituje offset až po úspěšném zpracování.

Nastavení pipeline Kafka Engine

Jednoduchý stack: Kafka + ZooKeeper + ClickHouse v Dockeru. Vytvoříme topic test_topic_json s JSON zprávami:

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

V ClickHouse pipeline ze tří komponent:

• Tabulka Kafka Engine pro čtení:

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;

• Materializované zobrazení jako 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;

• Cílová MergeTree:

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

Demonstrace atomarity

Po odeslání zpráv je offset consumeru my-group4 roven 2 (End=2). Nyní změníme typ sloupce v message_json_parse na UInt64 – parsování JSON řetězce "gg" selže:

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

Odešleme {"message": "gg"}. Offset zůstane 2 (End=3), log ClickHouse: Cannot parse string 'gg' as UInt64. Offset se commituje až po úspěšném zápisu do MV → MergeTree.

ClickHouse načte zprávu do RAM, ale necommituje offset při chybě. Nové zprávy jsou blokovány do opravy. Po opravě typu na String:

• Zpráva "gg" se zapíše.
• Offset se aktualizuje na 3.
• Žádné ztráty i po restartu ClickHouse.

Co je důležité

• Atomarita: Kafka offset se commituje po insertu do materializovaného zobrazení a zaručuje exactly-once sémantiku.
• Obnova: Po selhání/chybě zpracování pokračuje z posledního committed offsetu bez duplicit nebo propustů.
• Výkon: Dávkové zpracování z Kafka minimalizuje malé části v MergeTree.
• Spolehlivost: Kafka ukládá data persistentně, ClickHouse je stateless consumer.
• Omezení: Vyžaduje Kafka infrastrukturu; nastavte kafka_skip_broken_messages=0 pro striktní zpracování.

Doporučení pro implementaci

• Používejte kafka_num_consumers >1 pro paralelizmus.
• Nastavte max_block_size=2MB+ pro batching.
• Monitorujte offsets: SELECT * FROM system.kafka_consumers.
• Pro JSON: JSONEachRow se striktním parsováním.

— Editorial Team