` | MAP | Map(K, V) | ✅ | |
| oneof | STRUCT | Tuple / Variant | ⚠️ | Voir la section ci-dessous sur la traduction de oneof en schéma ClickHouse |
| google.protobuf.DoubleValue | FLOAT64 | Nullable(Float64) | ✅ | |
| google.protobuf.FloatValue | FLOAT32 | Nullable(Float32) | ✅ | |
| google.protobuf.Int64Value | INT64 | Nullable(Int64) | ✅ | |
| google.protobuf.UInt64Value | INT64 | Nullable(UInt64) | ✅ | |
| google.protobuf.UInt32Value | INT64 | Nullable(UInt32) | ✅ | |
| google.protobuf.Int32Value | INT32 | Nullable(Int32) | ✅ | |
| google.protobuf.BoolValue | BOOLEAN | Nullable(Bool) | ✅ | |
| google.protobuf.StringValue | STRING | Nullable(String) | ✅ | |
| google.protobuf.BytesValue | BYTES | Nullable(String) | ✅ | |
| google.protobuf.Timestamp | org.apache.kafka.connect.data.Timestamp | DateTime64(3) | ✅ | |
| google.type.Date | org.apache.kafka.connect.data.Date | Date | ✅ | |
| google.type.TimeOfDay | org.apache.kafka.connect.data.Time | Int32 / Int64 | ✅ | |
| google.protobuf.Duration | STRUCT | Tuple(`seconds` Int64, `nano` Nullable(Int32)) | ✅ | |
| google.protobuf.Any | *N/A* | *N/A* | ❌ | |
| google.protobuf.Empty | *N/A* | *N/A* | ❌ | |
Reportez-vous à [Types de données pris en charge](#supported-data-types) pour la correspondance entre les types Kafka Connect et les types ClickHouse.
#### Remarque sur la traduction des champs `oneof` en colonnes 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 :
```protobuf theme={null}
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 :
```sql theme={null}
CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
mixed_string Nullable(String),
mixed_int Nullable(Int32)
) ENGINE = ...;
```
#### Schémas Protobuf non pris en charge
Les schémas Protobuf suivants ne sont pas pris en charge par le connecteur :
* unions multi-messages (**avant la version 26.1 de CH**)
```protobuf theme={null}
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](/docs/fr/reference/settings/session-settings#allow_experimental_nullable_tuple_type)).
### Prise en charge du schéma JSON
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
}
}
```
### Prise en charge de String
Le connecteur prend en charge le convertisseur String avec différents formats ClickHouse : [JSON](/docs/fr/reference/formats/JSON/JSONEachRow), [CSV](/docs/fr/reference/formats/CSV/CSV) et [TSV](/docs/fr/reference/formats/TabSeparated/TabSeparated).
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"customInsertFormat": "true",
"insertFormat": "CSV"
}
}
```
### Mise en mémoire tampon interne
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 :
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"exactlyOnce": "false",
"bufferCount": "5000",
"bufferFlushTime": "2000"
}
}
```
### Journalisation
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](https://docs.confluent.io/platform/current/connect/logging.html#log4j-properties-file) de Kafka Connect.
Si vous utilisez Confluent Platform, vous pouvez consulter les logs en exécutant une commande CLI :
```bash theme={null}
confluent local services connect log
```
Pour en savoir plus, consultez le [tutoriel](https://docs.confluent.io/platform/current/connect/logging.html).
### Supervision
ClickHouse Kafka Connect expose des métriques d’exécution via [Java Management Extensions (JMX)](https://www.oracle.com/technical-resources/articles/javase/jmx.html). JMX est activé par défaut dans Kafka Connector.
#### Métriques spécifiques à ClickHouse
Le connecteur expose des métriques personnalisées via le nom MBean suivant :
```java theme={null}
com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}
```
| Nom de la métrique | Type | Description |
| ---------------------- | ---- | ----------------------------------------------------------------------------------------------------------------------- |
| `receivedRecords` | long | Nombre total d’enregistrements reçus. |
| `recordProcessingTime` | long | Temps total, en nanosecondes, consacré au regroupement et à la conversion des enregistrements en une structure unifiée. |
| `taskProcessingTime` | long | Temps total, en nanosecondes, consacré au traitement et à l’insertion des données dans ClickHouse. |
#### Métriques des producteurs et consommateurs Kafka
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é
#### Métriques du Kafka Connect Framework
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
#### Bonnes pratiques de supervision
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](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/main/jmx-export-connector.yml).
### Limitations
* 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)
### Réglage des performances et optimisation du débit
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.
#### Quand l’optimisation des performances est-elle nécessaire ?
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
#### Comprendre le flux des données
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.
#### Réglage de la taille des lots dans Kafka Connect
Le premier niveau d’optimisation consiste à contrôler la quantité de données que le connecteur reçoit de Kafka à chaque lot.
##### Paramètres de récupération
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)
Sur Confluent Cloud, l’ajustement de ces paramètres nécessite l’ouverture d’un ticket de support via Confluent Cloud.
##### Paramètres de poll
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)
Sur Confluent Cloud, la modification de ces paramètres nécessite l’ouverture d’un ticket de support via Confluent Cloud.
##### Paramètres recommandés pour un débit élevé
Pour des performances optimales avec ClickHouse, privilégiez des lots plus importants :
```properties theme={null}
# 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
```
Les propriétés ci-dessus nécessitent d’activer les client overrides dans la configuration de votre worker via `connector.client.config.override.policy=All`. Consultez la [documentation Kafka Connect](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) pour plus d’informations.
**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](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) | [documentation Kafka](https://kafka.apache.org/documentation/#consumerconfigs)
#### Insertions asynchrones
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.
##### Quand utiliser les insertions asynchrones
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
##### Fonctionnement de l’insertion asynchrone
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.
##### Activer les async inserts
Ajoutez les paramètres des async inserts au paramètre de configuration `clickhouseSettings` :
```json theme={null}
{
"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.
##### Réglage du comportement des async insert
Vous pouvez affiner le comportement de flush des async insert :
```json theme={null}
"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
##### Insertion asynchrone avec une garantie d'exactly-once
Lorsque vous utilisez `exactlyOnce=true` avec l’insertion asynchrone :
```json theme={null}
{
"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](/docs/fr/concepts/best-practices/selecting-an-insert-strategy#asynchronous-inserts).
#### Parallélisme du connecteur
Augmentez le parallélisme pour améliorer le débit :
##### Tâches par connecteur
```json theme={null}
"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.
##### Ignorer les partitions lors du regroupement en lots
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 :
```json theme={null}
"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.
#### Plusieurs topics à haut débit
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](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/578ac07e8be1a920aaa3b26e49183595c3edd04b/src/main/java/com/clickhouse/kafka/connect/sink/ProxySinkTask.java#L95-L100).
**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.
#### Considérations relatives au moteur de table ClickHouse
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** :
```sql theme={null}
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 :
```json theme={null}
"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"
```
#### Pool de connexions et délais d’expiration
Le connecteur maintient des connexions HTTP avec ClickHouse. Ajustez les délais d’expiration pour les réseaux à forte latence :
```json theme={null}
"clickhouseSettings": "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élai d’attente avec des lots volumineux.
#### Surveillance et dépannage des performances
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](#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** :
| Symptôme | Cause possible | Solution |
| ------------------------- | ------------------------------------------ | --------------------------------------------------------------------------- |
| Consumer lag élevé | Lots trop petits | Augmentez `max.poll.records`, activez les insertions asynchrones |
| Erreurs "Too many parts" | Insertions fréquentes de petite taille | Activez les insertions asynchrones, augmentez la taille des lots |
| Erreurs de timeout | Taille de lot trop importante, réseau lent | Réduisez la taille des lots, augmentez `socket_timeout`, vérifiez le réseau |
| Utilisation CPU élevée | Trop de petites parts | Activez les insertions asynchrones, augmentez les paramètres de fusion |
| Erreurs OutOfMemory | Taille de lot trop importante | Réduisez `max.poll.records`, `max.partition.fetch.bytes` |
| Charge des tâches inégale | Répartition inégale des partitions | Rééquilibrez les partitions ou ajustez `tasks.max` |
#### Résumé des bonnes pratiques
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
#### Exemple : configuration à haut débit
Voici un exemple complet optimisé pour un débit élevé :
```json theme={null}
{
"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": "",
"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"
}
}
```
La configuration du connecteur ci-dessus exige que vous activiez les client overrides dans la configuration de votre worker via `connector.client.config.override.policy=All`. Consultez la [documentation de Kafka Connect](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) pour plus d’informations.
**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
### Dépannage
#### "Incohérence d’état pour le topic `[someTopic]` partition `[0]`"
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 :
```sql theme={null}
-- 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]
```
Cet ajustement peut avoir des conséquences sur la sémantique exactly-once.
#### "Quelles erreurs le connecteur retentera-t-il ?"
À 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.
#### "Toutes mes données sont vides/à zéro"
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 :
```properties theme={null}
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.).
#### "Je veux utiliser mes clés Kafka dans ClickHouse"
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`) :
```properties theme={null}
transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key
```
