ClickHouse Kafka Connect Sink

ClickHouse Kafka Connect Sink est le connecteur Kafka qui achemine les données d’un topic Kafka vers une table ClickHouse.

Le Kafka Connector Sink est distribué sous licence Apache 2.0

Le framework Kafka Connect v2.7 ou une version ultérieure doit être installé dans l'environnement et s'exécuter sur Java 11 ou une version ultérieure.

Le connecteur refuse de démarrer avec des serveurs ClickHouse antérieurs à la version 23.3. Utilisez toujours la dernière version, sauf si vous devez utiliser une version antérieure.

• Livré avec une sémantique exactly-once prête à l'emploi. Il repose sur une nouvelle fonctionnalité du cœur de ClickHouse appelée KeeperMap (utilisée comme magasin d'état par le connecteur) et permet une architecture minimaliste.
• Prise en charge des magasins d'état tiers : utilise actuellement par défaut le stockage en mémoire, mais peut aussi utiliser KeeperMap (Redis sera bientôt ajouté).
• Intégration au cœur : développée, maintenue et prise en charge par ClickHouse.
• Testé en continu sur ClickHouse Cloud.
• Insertions de données avec ou sans schéma déclaré.
• Prise en charge de tous les types de données de ClickHouse.

Pour vous connecter à ClickHouse via HTTP(S), vous avez besoin des informations suivantes :

Les informations de votre service ClickHouse Cloud sont disponibles dans la console ClickHouse Cloud. Sélectionnez un service, puis cliquez sur Connect :

Choisissez HTTPS. Les détails de connexion s’affichent dans un exemple de commande curl.

Si vous utilisez ClickHouse autogéré, les détails de connexion sont définis par votre administrateur ClickHouse.

Le connecteur est distribué sous la forme d'un unique fichier JAR contenant tous les fichiers de classes nécessaires à l'exécution du plugin.

Pour installer le plugin, suivez les étapes suivantes :

• Téléchargez une archive ZIP contenant le fichier JAR du connecteur depuis la page Releases du dépôt ClickHouse Kafka Connect Sink.
• Extrayez le contenu du fichier ZIP et copiez-le à l'emplacement souhaité.
• Ajoutez à la configuration plugin.path de votre fichier de propriétés Connect le chemin du répertoire du plugin, afin que Confluent Platform puisse le trouver.
• Indiquez dans la config le nom du topic, le hostname de l'instance ClickHouse et le mot de passe.

connector.class=com.clickhouse.kafka.connect.ClickHouseSinkConnector
tasks.max=1
topics=<topic_name>
ssl=true
jdbcConnectionProperties=?sslmode=STRICT
security.protocol=SSL
hostname=<hostname>
database=<database_name>
password=<password>
ssl.truststore.location=/tmp/kafka.client.truststore.jks
port=8443
value.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
exactlyOnce=true
username=default
schemas.enable=false

• Redémarrez Confluent Platform.
• Si vous utilisez Confluent Platform, connectez-vous à l’UI du Confluent Control Center pour vérifier que le ClickHouse Sink figure dans la liste des connecteurs disponibles.

Pour connecter le ClickHouse Sink au serveur ClickHouse, vous devez fournir :

• les informations de connexion : nom d’hôte (obligatoire) et port (facultatif)
• les identifiants utilisateur : mot de passe (obligatoire) et nom d’utilisateur (facultatif)
• la classe du connecteur : com.clickhouse.kafka.connect.ClickHouseSinkConnector (obligatoire)
• topics ou topics.regex : les topics Kafka à interroger ; les noms de topic doivent correspondre aux noms de table (obligatoire)
• les convertisseurs de clé et de valeur : à définir selon le type de données de votre topic. Obligatoire s’ils ne sont pas déjà définis dans la configuration du worker.

Le tableau complet des options de configuration :

ClickHouse Connect Sink lit les messages des topics Kafka et les écrit dans les tables appropriées. ClickHouse Connect Sink écrit les données dans des tables existantes. Veuillez vous assurer qu'une table cible avec un schéma adapté a été créée dans ClickHouse avant de commencer à y insérer des données.

Chaque topic nécessite une table cible dédiée dans ClickHouse. Le nom de la table cible doit correspondre au nom du topic source.

Si vous devez transformer les messages sortants avant leur envoi à ClickHouse Kafka Connect Sink, utilisez les transformations Kafka Connect.

Avec un schéma déclaré :

• (1) - JSON est pris en charge uniquement lorsque les paramètres ClickHouse incluent input_format_binary_read_json_as_string=1. Cela fonctionne uniquement avec la famille de formats RowBinary, et ce paramètre affecte toutes les colonnes de la requête d'insertion ; elles doivent donc toutes être de type chaîne. Dans ce cas, le connecteur convertit STRUCT en chaîne JSON.

• (2) - Lorsqu'une struct contient des unions comme oneof, le convertisseur doit être configuré pour NE PAS ajouter de préfixe/suffixe aux noms de champs. Il existe le paramètre generate.index.for.unions=false pour ProtobufConverter.

Sans schéma déclaré :

Un enregistrement est converti en JSON et envoyé à ClickHouse comme valeur au format JSONEachRow.

Voici quelques exemples de configuration courants pour vous aider à démarrer rapidement.

La configuration minimale pour démarrer : elle suppose que vous exécutez Kafka Connect en mode distribué, qu’un serveur ClickHouse s’exécute sur localhost:8443 avec SSL activé, et que les données sont au format JSON sans schéma.

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "tasks.max": "1",
    "consumer.override.max.poll.records": "5000",
    "consumer.override.max.partition.fetch.bytes": "5242880",
    "database": "default",
    "errors.retry.timeout": "60000",
    "exactlyOnce": "false",
    "hostname": "localhost",
    "port": "8443",
    "ssl": "true",
    "jdbcConnectionProperties": "?ssl=true&sslmode=strict",
    "username": "default",
    "password": "<PASSWORD>",
    "topics": "<TOPIC_NAME>",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "clickhouseSettings": ""
  }
}

Le connecteur peut consommer des données à partir de plusieurs topics

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "topics": "SAMPLE_TOPIC, ANOTHER_TOPIC, YET_ANOTHER_TOPIC",
    ...
  }
}

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "<DLQ_TOPIC>",
    "errors.deadletterqueue.context.headers.enable": "true",
  }
}

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "<SCHEMA_REGISTRY_HOST>:<PORT>",
    "value.converter.schemas.enable": "true",
  }
}

La correspondance de types ci-dessous est définie par io.confluent.connect.avro.AvroConverter, l’implémentation officielle de sérialisation/désérialisation Avro de Kafka Connect. Consultez la documentation de Kafka Connect pour plus d’informations sur la logique de conversion avancée.

✅ : Pris en charge

❌ : Non pris en charge

️⚠️ : Partiellement pris en charge

Reportez-vous à Types de données pris en charge pour la correspondance entre les types Kafka Connect et les types ClickHouse.

Les schémas Avro suivants ne sont pas pris en charge par le connecteur :

• type logique decimal de type fixed

{"name": "decimal_18_4", "type": "fixed", "size": 8, "logicalType": "decimal", "precision": 18, "scale": 4}

• unions Nullable

{"name": "mixed_union", "type": ["null", "string", "int"], "default": null}

• unions d’enregistrements

{
  "name": "record_union",
  "type": [
    {
      "type": "record",
      "name": "TypeA",
      "fields": [
        {
          "name": "label",
          "type": "string"
        }
      ]
    },
    {
      "type": "record",
      "name": "TypeB",
      "fields": [
        {
          "name": "count",
          "type": "int"
        }
      ]
    }
  ]
}

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
    "value.converter.schema.registry.url": "<SCHEMA_REGISTRY_HOST>:<PORT>",
    "value.converter.schemas.enable": "true",
  }
}

Veuillez noter : si vous rencontrez des problèmes de classes manquantes, tous les environnements n’incluent pas le convertisseur Protobuf et vous devrez peut-être utiliser une autre version du fichier JAR fournie avec les dépendances.

La correspondance des types ci-dessous est définie par io.confluent.connect.protobuf.ProtobufConverter, l’implémentation officielle du sérialiseur/désérialiseur Protobuf dans Kafka Connect. Consultez la documentation de Kafka Connect pour plus d’informations sur la logique de conversion.

✅ : Pris en charge

❌ : Non pris en charge

️⚠️ : Partiellement pris en charge

Reportez-vous à Types de données pris en charge pour la correspondance entre les types Kafka Connect et les types ClickHouse.

Le connecteur ne prend pas en charge la traduction des unions Protobuf (oneof) vers le type Variant de ClickHouse. Indiquez plutôt les champs oneof séparément comme des champs nullable dans le schéma de votre table ClickHouse.

Par exemple :

syntax = "proto3";

package com.clickhouse.kafka.connect.proto.test;

message StringIntUnion {
  oneof mixed {
    string mixed_string = 2;
    int32 mixed_int = 3;
  }
}

se traduit par la définition de la table ClickHouse suivante :

CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
    mixed_string Nullable(String),
    mixed_int Nullable(Int32)
) ENGINE = ...;

Les schémas Protobuf suivants ne sont pas pris en charge par le connecteur :

• unions multi-messages (avant la version 26.1 de CH)

syntax = "proto3";

package com.clickhouse.kafka.connect.proto.test;

message TwoRecords {
  oneof payload {
    TypeA type_a = 2;
    TypeB type_b = 3;
  }

  // translates to Nullable(Tuple(label String)) in ClickHouse, which is unsupported
  message TypeA {
    string label = 1;
  }

  // translates to Nullable(Tuple(count Int32)) in ClickHouse, which is unsupported
  message TypeB {
    int32 count = 1;
  }
}

À partir de la version 26.1 de CH, ce schéma est pris en charge si allow_experimental_nullable_tuple_type=1 (voir cette page de la documentation).

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  }
}

Le connecteur prend en charge le convertisseur String avec différents formats ClickHouse : JSON, CSV et TSV.

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "customInsertFormat": "true",
    "insertFormat": "CSV"
  }
}

La mise en mémoire tampon interne permet à la tâche sink d’accumuler des enregistrements issus de plusieurs appels à poll() et de les envoyer à ClickHouse sous forme de batches plus volumineux. Cela peut améliorer le débit dans les workloads où chaque interrogation produit de nombreux petits batches par partition.

Comportement clé :

• bufferCount contrôle le nombre d’enregistrements mis en mémoire tampon avant leur envoi.
• bufferFlushTime définit le temps d’attente maximal (en millisecondes) avant l’envoi des enregistrements mis en mémoire tampon.
• bufferFlushTime n’a d’effet que lorsque bufferCount > 0.
• bufferCount=0 et bufferFlushTime=0 laissent la mise en mémoire tampon désactivée (comportement par défaut).
• La mise en mémoire tampon n’est pas prise en charge lorsque exactlyOnce=true.

Pourquoi la mise en mémoire tampon est incompatible avec le mode exactly-once : La mise en mémoire tampon modifie les limites des batches, ce qui perturbe la déduplication des blocs dans ClickHouse ainsi que la machine à états des offsets du connecteur. Pour corriger cela, désactivez soit le mode exactly-once avec exactlyOnce=false dans la config de votre connecteur, soit la mise en mémoire tampon avec bufferCount=0.

Exemple :

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "exactlyOnce": "false",
    "bufferCount": "5000",
    "bufferFlushTime": "2000"
  }
}

La journalisation est fournie automatiquement par la plateforme Kafka Connect. La destination et le format des logs peuvent être configurés via le fichier de configuration de Kafka Connect.

Si vous utilisez Confluent Platform, vous pouvez consulter les logs en exécutant une commande CLI :

confluent local services connect log

Pour en savoir plus, consultez le tutoriel.

ClickHouse Kafka Connect expose des métriques d’exécution via Java Management Extensions (JMX). JMX est activé par défaut dans Kafka Connector.

Le connecteur expose des métriques personnalisées via le nom MBean suivant :

com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}

Le connecteur expose les métriques standard des producteurs et consommateurs Kafka, qui donnent de la visibilité sur le flux de données, le débit et les performances.

Métriques au niveau du topic :

• records-sent-total : Nombre total d’enregistrements envoyés au topic
• bytes-sent-total : Nombre total d’octets envoyés au topic
• record-send-rate : Taux moyen d’enregistrements envoyés par seconde
• byte-rate : Nombre moyen d’octets envoyés par seconde
• compression-rate : Ratio de compression obtenu

Métriques au niveau de la partition :

• records-sent-total : Nombre total d’enregistrements envoyés à la partition
• bytes-sent-total : Nombre total d’octets envoyés à la partition
• records-lag : Lag actuel de la partition
• records-lead : Avance actuelle de la partition
• replica-fetch-lag : Informations de lag des répliques

Métriques de connexion au niveau du nœud :

• connection-creation-total : Nombre total de connexions créées vers le nœud Kafka
• connection-close-total : Nombre total de connexions fermées
• request-total : Nombre total de requêtes envoyées au nœud
• response-total : Nombre total de réponses reçues du nœud
• request-rate : Taux moyen de requêtes par seconde
• response-rate : Taux moyen de réponses par seconde

Ces métriques aident à surveiller :

• Débit : Suivre les débits d’ingestion des données
• Lag : Identifier les goulots d’étranglement et les retards de traitement
• Compression : Mesurer l’efficacité de la compression des données
• État des connexions : Surveiller la connectivité réseau et la stabilité

Le connecteur s’intègre au Kafka Connect Framework et expose des métriques sur le cycle de vie des tâches et le suivi des erreurs.

Métriques d’état des tâches :

• task-count : Nombre total de tâches dans le connecteur
• running-task-count : Nombre de tâches actuellement en cours d’exécution
• paused-task-count : Nombre de tâches actuellement en pause
• failed-task-count : Nombre de tâches en échec
• destroyed-task-count : Nombre de tâches détruites
• unassigned-task-count : Nombre de tâches non attribuées

Les valeurs d’état des tâches incluent : running, paused, failed, destroyed, unassigned

Métriques d’erreur :

• deadletterqueue-produce-failures : Nombre d’écritures dans la DLQ ayant échoué
• deadletterqueue-produce-requests : Nombre total de tentatives d’écriture dans la DLQ
• last-error-timestamp : Horodatage de la dernière erreur
• records-skip-total : Nombre total d’enregistrements ignorés en raison d’erreurs
• records-retry-total : Nombre total d’enregistrements ayant fait l’objet d’une nouvelle tentative
• errors-total : Nombre total d’erreurs rencontrées

Métriques de performance :

• offset-commit-failures : Nombre d’échecs de commit des offsets
• offset-commit-avg-time-ms : Temps moyen des commits d’offset
• offset-commit-max-time-ms : Temps maximal des commits d’offset
• put-batch-avg-time-ms : Temps moyen de traitement d’un batch
• put-batch-max-time-ms : Temps maximal de traitement d’un batch
• source-record-poll-total : Nombre total d’enregistrements récupérés

1. Surveillez le consumer lag : suivez records-lag par partition afin d’identifier les goulots d’étranglement du traitement
2. Suivez les taux d’erreur : surveillez errors-total et records-skip-total pour détecter les problèmes de qualité des données
3. Vérifiez l’état des tâches : surveillez les métriques d’état des tâches pour vous assurer qu’elles s’exécutent correctement
4. Mesurez le débit : utilisez records-send-rate et byte-rate pour suivre les performances d’ingestion
5. Surveillez l’état des connexions : vérifiez les métriques de connexion au niveau des nœuds afin de détecter d’éventuels problèmes réseau
6. Suivez l’efficacité de la compression : utilisez compression-rate pour optimiser le transfert de données

Pour des définitions détaillées des métriques JMX et de l’intégration avec Prometheus, consultez le fichier de configuration jmx-export-connector.yml.

• Les suppressions ne sont pas prises en charge.
• La taille du lot est héritée des propriétés du consommateur Kafka.
• Lorsque vous utilisez KeeperMap pour exactly-once et que l'offset est modifié ou réinitialisé à une valeur antérieure, vous devez supprimer le contenu de KeeperMap pour ce topic spécifique. (Consultez le guide de dépannage ci-dessous pour plus de détails)

Cette section présente des stratégies de réglage des performances pour le ClickHouse Kafka Connect Sink. Le réglage des performances est essentiel lorsque vous traitez des cas d’usage à haut débit ou que vous devez optimiser l’utilisation des ressources et minimiser la latence.

L’optimisation des performances est généralement nécessaire dans les cas suivants :

• Charges de travail à haut débit : lorsque vous traitez des millions d’événements par seconde depuis des topics Kafka
• Consumer lag : lorsque votre connecteur ne parvient pas à suivre le rythme de production des données, ce qui entraîne un décalage croissant
• Contraintes de ressources : lorsque vous devez optimiser l’utilisation du CPU, de la mémoire ou du réseau
• Topics multiples : lorsque vous consommez simultanément plusieurs topics à fort volume
• Petites tailles de message : lorsque vous traitez de nombreux petits messages qui bénéficieraient d’un regroupement côté serveur par lots

L’optimisation des performances n’est généralement PAS nécessaire lorsque :

• Vous traitez des volumes faibles à modérés (< 10 000 messages/seconde)
• Le consumer lag est stable et acceptable pour votre cas d’usage
• Les paramètres par défaut du connecteur répondent déjà à vos exigences de débit
• Votre cluster ClickHouse peut facilement absorber la charge entrante

Avant tout réglage, il est important de comprendre comment les données circulent dans le connecteur :

1. Kafka Connect Framework récupère en arrière-plan les messages des topics Kafka
2. Le connecteur interroge le tampon interne du framework pour récupérer les messages
3. Le connecteur regroupe en lots les messages selon la taille du poll
4. ClickHouse reçoit l’insertion par lot via HTTP/S
5. ClickHouse traite l’insertion (de manière synchrone ou asynchrone)

Les performances peuvent être optimisées à chacune de ces étapes.

Le premier niveau d’optimisation consiste à contrôler la quantité de données que le connecteur reçoit de Kafka à chaque lot.

Kafka Connect (le framework) récupère les messages des topics Kafka en arrière-plan, indépendamment du connecteur :

• fetch.min.bytes : quantité minimale de données avant que le framework ne transmette les valeurs au connecteur (par défaut : 1 octet)
• fetch.max.bytes : quantité maximale de données à récupérer dans une seule requête (par défaut : 52428800 / 50 MB)
• fetch.max.wait.ms : temps d’attente maximal avant le renvoi des données si fetch.min.bytes n’est pas atteint (par défaut : 500 ms)

Le connecteur effectue des polls pour récupérer les messages depuis le tampon du framework :

• max.poll.records : Nombre maximal d’enregistrements renvoyés lors d’une seule opération de poll (par défaut : 500)
• max.partition.fetch.bytes : Quantité maximale de données par partition (par défaut : 1048576 / 1 MB)

Pour des performances optimales avec ClickHouse, privilégiez des lots plus importants :

# Increase the number of records per poll
consumer.override.max.poll.records=5000

# Increase the partition fetch size (5 MB)
consumer.override.max.partition.fetch.bytes=5242880

# Optional: Increase minimum fetch size to wait for more data (1 MB)
consumer.override.fetch.min.bytes=1048576

# Optional: Reduce wait time if latency is critical
consumer.override.fetch.max.wait.ms=300

Important : les paramètres de récupération de Kafka Connect portent sur des données compressées, tandis que ClickHouse reçoit des données non compressées. Ajustez ces paramètres en fonction de votre ratio de compression.

Compromis :

• Lots plus volumineux = meilleures performances d’ingestion dans ClickHouse, moins de parts, moins de surcharge
• Lots plus volumineux = utilisation mémoire plus élevée, augmentation potentielle de la latence de bout en bout
• Lots trop volumineux = risque de dépassement de délai, d’erreurs OutOfMemory ou de dépassement de max.poll.interval.ms

Plus de détails : documentation Confluent | documentation Kafka

Les insertions asynchrones sont particulièrement utiles lorsque le connecteur envoie des lots relativement petits ou lorsque vous souhaitez optimiser davantage l’ingestion en confiant à ClickHouse la gestion du traitement par lots.

Envisagez d’activer les insertions asynchrones dans les cas suivants :

• Nombreux petits lots : votre connecteur envoie fréquemment de petits lots (< 1000 lignes par lot)
• Concurrence élevée : plusieurs tâches du connecteur écrivent dans la même table
• Déploiement distribué : vous exécutez de nombreuses instances du connecteur sur différents hôtes
• Surcoût lié à la création de parts : vous rencontrez des erreurs « too many parts »
• Charge de travail mixte : vous combinez une ingestion en temps réel avec des charges liées aux requêtes

N’utilisez PAS les insertions asynchrones dans les cas suivants :

• Vous envoyez déjà de grands lots (> 10,000 lignes par lot) à une fréquence maîtrisée
• Vous avez besoin d’une visibilité immédiate des données (les requêtes doivent voir les données instantanément)
• La sémantique d’exactement une fois avec wait_for_async_insert=0 est incompatible avec vos exigences
• Votre cas d’usage peut plutôt tirer parti d’améliorations du batching côté client

Avec l’insertion asynchrone activée, ClickHouse :

1. Reçoit la requête d’insertion du connecteur
2. Écrit les données dans un tampon en mémoire (au lieu de les écrire immédiatement sur le disque)
3. Renvoie une confirmation de succès au connecteur (si wait_for_async_insert=0)
4. Vide le tampon sur le disque lorsqu’une de ces conditions est remplie :
   • Le tampon atteint async_insert_max_data_size (par défaut : 100 MB)
   • async_insert_busy_timeout_ms millisecondes se sont écoulées depuis le premier insert (par défaut : 1000 ms)
   • Le nombre maximal de requêtes accumulées est atteint (async_insert_max_query_number, par défaut : 100)

Cela réduit considérablement le nombre de parts créées et améliore le débit global.

Ajoutez les paramètres des async inserts au paramètre de configuration clickhouseSettings :

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
  }
}

Paramètres clés :

• async_insert=1 : Active les insertions asynchrones
• wait_for_async_insert=1 (recommandé) : Le connecteur attend que les données soient écrites dans le stockage ClickHouse avant d’envoyer un accusé de réception. Cela garantit la livraison.
• wait_for_async_insert=0 : Le connecteur envoie immédiatement un accusé de réception après la mise en mémoire tampon. Les performances sont meilleures, mais les données peuvent être perdues en cas de plantage du serveur avant leur écriture sur le stockage.

Vous pouvez affiner le comportement de flush des async insert :

"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=104857600,async_insert_busy_timeout_ms=1000"

Paramètres de réglage courants :

• async_insert_max_data_size (par défaut : 104857600 / 100 MB) : Taille maximale du tampon avant vidage
• async_insert_busy_timeout_ms (par défaut : 1000) : Temps maximal (ms) avant vidage
• async_insert_stale_timeout_ms (par défaut : 0) : Temps (ms) écoulé depuis la dernière insertion avant vidage
• async_insert_max_query_number (par défaut : 100) : Nombre maximal de requêtes avant vidage

Compromis :

• Avantages : Moins de parts, meilleures performances de fusion, surcharge CPU réduite, débit amélioré en cas de forte concurrence
• Considérations : Les données ne peuvent pas être interrogées immédiatement, latence de bout en bout légèrement plus élevée
• Risques : Perte de données en cas de plantage du serveur si wait_for_async_insert=0, risque de pression sur la mémoire avec de grands tampons

Lorsque vous utilisez exactlyOnce=true avec l’insertion asynchrone :

{
  "config": {
    "exactlyOnce": "true",
    "clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
  }
}

Important : utilisez toujours wait_for_async_insert=1 avec exactly-once afin de garantir que les commits d’offsets n’ont lieu qu’une fois les données persistées.

Pour plus d’informations sur l’insertion asynchrone, consultez la documentation ClickHouse sur l’insertion asynchrone.

Augmentez le parallélisme pour améliorer le débit :

"tasks.max": "4"

Chaque tâche traite un sous-ensemble des partitions du topic. Plus de tâches = plus de parallélisme, mais :

• Nombre maximal de tâches réellement efficaces = nombre de partitions du topic
• Chaque tâche maintient sa propre connexion à ClickHouse
• Plus de tâches = plus de surcoût et un risque accru de contention sur les ressources

Recommandation : commencez avec tasks.max égal au nombre de partitions du topic, puis ajustez en fonction des métriques de CPU et de débit.

Par défaut, le connecteur regroupe les messages en lots par partition. Pour un débit plus élevé, vous pouvez regrouper les lots sur plusieurs partitions :

"ignorePartitionsWhenBatching": "true"

** Avertissement** : à utiliser uniquement lorsque exactlyOnce=false. Ce paramètre peut améliorer le débit en créant des lots plus volumineux, mais supprime les garanties d’ordre par partition.

Si votre connecteur est configuré pour s’abonner à plusieurs topics, que vous utilisez topic2TableMap pour associer les topics aux tables et que vous rencontrez un goulot d’étranglement lors des insertions, entraînant un consumer lag, envisagez plutôt de créer un connecteur par topic.

La principale raison est qu’actuellement, les lots sont insérés dans chaque table en série.

Recommandation : pour plusieurs topics à fort volume, déployez une instance de connecteur par topic afin de maximiser le débit des insertions en parallèle.

Choisissez le moteur de table ClickHouse adapté à votre cas d’utilisation :

• MergeTree : idéal pour la plupart des cas d’utilisation, offre un bon équilibre entre les performances de requête et d’insertion
• ReplicatedMergeTree : requis pour la haute disponibilité, ajoute une surcharge liée à la réplication
• *MergeTree avec un ORDER BY approprié : optimisez en fonction de vos modèles de requêtes

Paramètres à prendre en compte :

CREATE TABLE my_table (...)
ENGINE = MergeTree()
ORDER BY (timestamp, id)
SETTINGS 
    -- Increase max insert threads for parallel part writing
    max_insert_threads = 4,
    -- Allow inserts with quorum for reliability (ReplicatedMergeTree)
    insert_quorum = 2

Pour les paramètres d’insertion au niveau du connecteur :

"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"

Le connecteur maintient des connexions HTTP avec ClickHouse. Ajustez les dépassements de délai pour les réseaux à forte latence :

"jdbcConnectionProperties": "?socket_timeout=300000,connection_timeout=30000"

• socket_timeout (par défaut : 30000 ms) : Temps maximal pour les opérations de lecture
• connection_timeout (par défaut : 10000 ms) : Temps maximal pour établir une connexion

Augmentez ces valeurs si vous rencontrez des erreurs de dépassement de délai avec des lots volumineux.

Lors de l’utilisation du client V2 ("client_version": "V2"), les données sont copiées entre le socket et la mémoire de l’application via un tampon contrôlé par client_network_buffer_size (en octets). Sa taille par défaut est de 300000 octets. Un tampon plus grand peut améliorer le débit sur des liaisons à forte latence ou à large bande passante, mais augmente l’utilisation de la mémoire par tâche de connexion.

Toutes les options du client se configurent via jdbcConnectionProperties plutôt que via clickhouseSettings. Par exemple :

"jdbcConnectionProperties": "?client_network_buffer_size=1000000"

Si vous devez configurer plusieurs propriétés client, séparez-les par & :

"jdbcConnectionProperties": "?ssl=true&client_network_buffer_size=1000000"

Surveillez ces métriques clés :

1. Consumer lag : utilisez les outils de monitoring Kafka pour suivre le lag par partition
2. Métriques du connecteur : surveillez receivedRecords, recordProcessingTime, taskProcessingTime via JMX (voir Monitoring)
3. Métriques ClickHouse :
   • system.asynchronous_inserts : surveillez l’utilisation du buffer des insertions asynchrones
   • system.parts : surveillez le nombre de parts pour détecter les problèmes de fusion
   • system.merges : surveillez les fusions actives
   • system.events : suivez InsertedRows, InsertedBytes, FailedInsertQuery

Problèmes de performances courants :

1. Commencez par les valeurs par défaut, puis mesurez et ajustez en fonction des performances réelles
2. Privilégiez des lots plus volumineux : visez 10,000-100,000 lignes par insertion lorsque c’est possible
3. Utilisez l’insertion asynchrone lorsque vous envoyez de nombreux petits lots ou en cas de forte concurrence
4. Utilisez toujours wait_for_async_insert=1 avec une sémantique exactly-once
5. Effectuez une mise à l’échelle horizontale : augmentez tasks.max jusqu’au nombre de partitions
6. Un connecteur par topic à fort volume pour un débit maximal
7. Surveillez en continu : suivez le consumer lag, le nombre de parts et l’activité de merge
8. Testez de manière approfondie : testez toujours les changements de configuration sous une charge réaliste avant un déploiement en production

Voici un exemple complet optimisé pour un débit élevé :

{
  "name": "clickhouse-high-throughput",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "tasks.max": "8",
    
    "topics": "high_volume_topic",
    "hostname": "my-clickhouse-host.cloud",
    "port": "8443",
    "database": "default",
    "username": "default",
    "password": "<PASSWORD>",
    "ssl": "true",
    
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    
    "exactlyOnce": "false",
    "ignorePartitionsWhenBatching": "true",
    
    "consumer.override.max.poll.records": "10000",
    "consumer.override.max.partition.fetch.bytes": "5242880",
    "consumer.override.fetch.min.bytes": "1048576",
    "consumer.override.fetch.max.wait.ms": "500",
    
    "clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=16777216,async_insert_busy_timeout_ms=1000,socket_timeout=300000"
  }
}

Cette configuration :

• Traite jusqu’à 10 000 enregistrements par poll
• Regroupe les données de plusieurs partitions en lots pour des inserts plus volumineux
• Utilise l’insertion asynchrone avec un buffer de 16 Mo
• Exécute 8 tasks en parallèle (à aligner sur votre nombre de partitions)
• Optimisée pour le débit plutôt que pour un ordre strict

Cela se produit lorsque l’offset stocké dans KeeperMap est différent de celui stocké dans Kafka, généralement lorsqu’un topic a été supprimé ou que l’offset a été modifié manuellement. Pour corriger cela, vous devez supprimer les anciennes valeurs stockées pour ce topic + cette partition :

-- First, identify the database used to store the data.
SELECT * FROM [database].connect_state

-- Identify the key that matches the topic and partition.
ALTER TABLE [database].connect_state DELETE WHERE key = [keyname]

À l'heure actuelle, l'accent est mis sur l'identification des erreurs transitoires pouvant faire l'objet d'une nouvelle tentative, notamment :

• ClickHouseException - Il s'agit d'une exception générique pouvant être levée par ClickHouse. Elle est généralement levée lorsque le serveur est surchargé, et les codes d'erreur suivants sont considérés comme particulièrement transitoires :
  • 3 - UNEXPECTED_END_OF_FILE
  • 107 - FILE_DOESNT_EXIST
  • 159 - TIMEOUT_EXCEEDED
  • 164 - READONLY
  • 202 - TOO_MANY_SIMULTANEOUS_QUERIES
  • 203 - NO_FREE_CONNECTION
  • 209 - SOCKET_TIMEOUT
  • 210 - NETWORK_ERROR
  • 241 - MEMORY_LIMIT_EXCEEDED
  • 242 - TABLE_IS_READ_ONLY
  • 252 - TOO_MANY_PARTS
  • 285 - TOO_FEW_LIVE_REPLICAS
  • 319 - UNKNOWN_STATUS_OF_INSERT
  • 425 - SYSTEM_ERROR
  • 999 - KEEPER_EXCEPTION
• SocketTimeoutException - Cette exception est levée lorsque le délai d'attente du socket est dépassé.
• UnknownHostException - Cette exception est levée lorsque l'hôte ne peut pas être résolu.
• IOException - Cette exception est levée en cas de problème réseau.

Il est probable que les champs de vos données ne correspondent pas à ceux de la table - c’est particulièrement fréquent avec le CDC (et le format Debezium). Une solution courante consiste à ajouter la transformation flatten à la configuration de votre connecteur :

transforms=flatten
transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
transforms.flatten.delimiter=_

Cela transformera vos données d’un JSON imbriqué en un JSON aplati (en utilisant _ comme délimiteur). Les champs de la table suivront alors le format "field1_field2_field3" (par ex. "before_id", "after_id", etc.).

Par défaut, les clés Kafka ne sont pas stockées dans le champ value, mais vous pouvez utiliser la transformation KeyToValue pour déplacer la clé dans le champ value (sous le nouveau nom de champ _key) :

transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key