Kafka Engine ClickHouse : atomicité sans pertes

Moteur Kafka dans ClickHouse : Garantir des insertions atomiques sans perte de données

Les insertions individuelles fréquentes dans les tables MergeTree (10 000 INSERT/s) font chuter les performances à 59 lignes/s. Pour ReplicatedMergeTree, cela descend à 6 lignes/s, et avec quorum, à seulement 2 lignes/s. Cela est dû à la création d'une multitude de petites parties de données, qui sont périodiquement fusionnées en fonction du paramètre max_merge_selecting_sleep_ms. Le processus de fusion surcharge le CPU et les E/S, réduisant la vitesse globale.

Les tables Buffer semblent être une solution : les données s'accumulent en RAM et sont vidées vers la table cible par lots (par temps, taille ou nombre de lignes). Cela minimise le nombre de fichiers et la charge de fusion. Cependant, si ClickHouse plante, les données non vidées en mémoire sont perdues.

Tables Buffer : Mécanisme et risques

Une table tampon utilise le moteur Buffer, accumulant les données en mémoire jusqu'à ce que les conditions de vidage soient remplies : max_time_to_flush (par ex., 100 s), min_time_to_flush, max_rows_to_flush (1 million de lignes), max_bytes_to_flush (1 GB). Le vidage se produit de manière atomique dans MergeTree, mais seulement après écriture sur disque.

Risque principal : Si le serveur plante (kill -9, OOM), le contenu du tampon en RAM est perdu à jamais. Il n'y a pas de réplication ni de WAL pour les tampons, rendant cette approche peu fiable pour les flux à forte charge.

Passer au moteur Kafka résout le problème : les données sont persistantes dans Kafka, et ClickHouse ne valide l'offset qu'après un traitement réussi.

Configuration du pipeline du moteur Kafka

Stack simple : Kafka + ZooKeeper + ClickHouse dans Docker. Créer le topic test_topic_json avec des messages JSON :

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

Dans ClickHouse, le pipeline comprend trois composants :

• Table moteur Kafka pour la lecture :

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;

• Vue matérialisée en tant que consommateur :

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 cible :

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

Démonstration de l'atomicité

Après envoi des messages, l'offset du consommateur pour my-group4 est de 2 (End=2). Changez maintenant le type de colonne dans message_json_parse en UInt64 — l'analyse de la chaîne JSON "gg" échouera :

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

Envoyez {"message": "gg"}. L'offset reste à 2 (End=3), journal ClickHouse : Cannot parse string 'gg' as UInt64. L'offset n'est validé qu'après une écriture réussie dans MV → MergeTree.

ClickHouse lit le message en RAM mais ne valide pas l'offset en cas d'erreur. Les nouveaux messages sont bloqués jusqu'à correction. Après correction du type en String :

• Le message "gg" est inséré.
• L'offset passe à 3.
• Pas de perte de données même après redémarrage de ClickHouse.

Points clés

• Atomicité : L'offset Kafka est validé après insertion dans la vue matérialisée, garantissant une sémantique exactly-once.
• Récupération : Après panne/erreur, le traitement reprend à partir du dernier offset validé, sans doublons ni omissions.
• Performances : Le traitement par lots depuis Kafka minimise les petites parties dans MergeTree.
• Fiabilité : Kafka stocke les données de manière persistante ; ClickHouse agit comme un consommateur sans état.
• Limites : Nécessite l'infrastructure Kafka ; définissez kafka_skip_broken_messages=0 pour un traitement strict.

Recommandations de mise en œuvre

• Utilisez kafka_num_consumers >1 pour le parallélisme.
• Définissez max_block_size=2MB+ pour le traitement par lots.
• Surveillez les offsets : SELECT * FROM system.kafka_consumers.
• Pour JSON : JSONEachRow avec analyse stricte.

— Editorial Team