Kafka Engine ClickHouse: atomowość bez strat

Silnik Kafka w ClickHouse: gwarancja atomowości wstawień bez utraty danych

Częste pojedyncze wstawki do tabel MergeTree (10 000 INSERT/s) powodują spadek wydajności do 59 wierszy/s, w ReplicatedMergeTree — do 6 wierszy/s, z kworum — do 2 wierszy/s. Jest to spowodowane tworzeniem wielu małych części danych, które okresowo są łączone (merge) według ustawienia max_merge_selecting_sleep_ms. Proces łączenia przeciąża CPU i I/O, obniżając ogólną prędkość.

Tabele buforowe wydają się rozwiązaniem: dane gromadzone są w RAM i zrzucane do tabeli docelowej partiami (według czasu, objętości lub liczby wierszy). Minimalizuje to liczbę plików i obciążenie merge. Jednak w przypadku awarii ClickHouse niezrzucone dane z pamięci są tracone.

Tabele buforowe: mechanizm i ryzyka

Tabela buforowa używa silnika Buffer, gromadząc dane w pamięci do spełnienia warunków zrzutu: max_time_to_flush (np. 100 s), min_time_to_flush, max_rows_to_flush (1 mln wierszy), max_bytes_to_flush (1 GB). Zrzut odbywa się atomowo do MergeTree, ale dopiero po zapisaniu na dysk.

Kluczowe ryzyko: w przypadku awarii serwera (kill -9, OOM) zawartość bufora w RAM znika bezpowrotnie. Brak replikacji lub WAL dla buforów czyni to podejście niewiarygodnym dla strumieni o dużym obciążeniu.

Przejście na Kafka Engine rozwiązuje problem: dane są trwałe w Kafka, ClickHouse commituje offset dopiero po pomyślnym przetworzeniu.

Konfiguracja potoku Kafka Engine

Prosty stos: Kafka + ZooKeeper + ClickHouse w Dockerze. Tworzymy topik test_topic_json z wiadomościami JSON:

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

W ClickHouse potok składa się z trzech komponentów:

• Tabela Kafka Engine do odczytu:

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;

• Widok materializowany jako konsument:

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;

• Docelowa MergeTree:

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

Demonstracja atomowości

Po wysłaniu wiadomości offset konsumenta my-group4 wynosi 2 (End=2). Teraz zmieniamy typ kolumny w message_json_parse na UInt64 — parsowanie ciągu JSON „gg” się nie powiedzie:

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

Wysyłamy {"message": "gg"}. Offset pozostaje 2 (End=3), log ClickHouse: Cannot parse string 'gg' as UInt64. Offset jest commitowany dopiero po pomyślnym zapisie do MV → MergeTree.

ClickHouse odczytuje wiadomość do RAM, ale nie commituje offsetu w przypadku błędu. Nowe wiadomości są blokowane do naprawy. Po poprawieniu typu na String:

• Wiadomość „gg” jest zapisywana.
• Offset jest aktualizowany do 3.
• Brak strat nawet po restarcie ClickHouse.

Co ważne

• Atomowość: Offset Kafka jest commitowany po wstawieniu do widoku materializowanego, gwarantując semantykę exactly-once.
• Odzyskiwanie: Po awarii/błędzie przetwarzanie kontynuowane jest od ostatniego committed offset bez duplikatów lub pominięć.
• Wydajność: Pakietowe przetwarzanie z Kafka minimalizuje małe części w MergeTree.
• Niezawodność: Kafka przechowuje dane trwale, ClickHouse — bezstanowy konsument.
• Ograniczenia: Wymaga infrastruktury Kafka; ustaw kafka_skip_broken_messages=0 dla ścisłego przetwarzania.

Zalecenia wdrożeniowe

• Używaj kafka_num_consumers >1 dla paralelizmu.
• Ustaw max_block_size=2MB+ dla batczingu.
• Monitoruj offsety: SELECT * FROM system.kafka_consumers.
• Dla JSON: JSONEachRow z surowym parsowaniem.

— Editorial Team