`). La configuration globale est appliquée en premier, puis la configuration au niveau du topic est appliquée (si elle existe).
```xml theme={null}
cgrp
3000
logs
4000
smallest
logs
100000
stats
50000
logs
250
stats
400
```
Pour obtenir la liste des options de configuration possibles, consultez la [référence de configuration de librdkafka](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). Dans la configuration de ClickHouse, utilisez le caractère de soulignement (`_`) à la place d’un point. Par exemple, `check.crcs=true` deviendra `true`.
### Prise en charge de Kerberos
Pour utiliser Kafka avec Kerberos, ajoutez l’élément enfant `security_protocol` avec la valeur `sasl_plaintext`. Cela suffit si le ticket d’octroi de tickets Kerberos est obtenu et mis en cache par les mécanismes du système d’exploitation.
ClickHouse peut gérer les informations d’identification Kerberos à l’aide d’un fichier keytab. Tenez compte des éléments enfants `sasl_kerberos_service_name`, `sasl_kerberos_keytab` et `sasl_kerberos_principal`.
Exemple :
```xml theme={null}
SASL_PLAINTEXT
/home/kafkauser/kafkauser.keytab
kafkauser/kafkahost@EXAMPLE.COM
```
## Colonnes virtuelles
* `_topic` — Topic Kafka. Type de données : `LowCardinality(String)`.
* `_key` — Clé du message. Type de données : `String`.
* `_offset` — Offset du message. Type de données : `UInt64`.
* `_timestamp` — Horodatage du message. Type de données : `Nullable(DateTime)`.
* `_timestamp_ms` — Horodatage du message en millisecondes. Type de données : `Nullable(DateTime64(3))`.
* `_partition` — Partition du topic Kafka. Type de données : `UInt64`.
* `_headers.name` — Tableau des clés d'en-tête du message. Type de données : `Array(String)`.
* `_headers.value` — Tableau des valeurs d'en-tête du message. Type de données : `Array(String)`.
Colonnes virtuelles supplémentaires lorsque `kafka_handle_error_mode='stream'` :
* `_raw_message` - Message brut qui n'a pas pu être analysé correctement. Type de données : `String`.
* `_error` - Message d'exception généré lors d'une erreur d'analyse. Type de données : `String`.
Remarque : les colonnes virtuelles `_raw_message` et `_error` sont renseignées uniquement en cas d'exception pendant l'analyse ; elles sont toujours vides lorsque le message a été analysé correctement.
## Correspondance entre les colonnes et les métadonnées des messages Kafka
Lors de la production de messages avec `INSERT INTO`, le moteur Kafka utilise toujours une colonne nommée `_key` (de type `String`) comme clé du message Kafka et une colonne nommée `_timestamp` (de type `DateTime`) comme horodatage du message Kafka — si ces colonnes existent dans la table. Par défaut, ces colonnes apparaissent également dans la charge utile du message produit, aux côtés des autres colonnes.
Avec `kafka_map_virtual_columns_on_write = 1`, le comportement change :
* `_key` (type `String`) — associé à la clé du message Kafka.
* `_timestamp` (type `DateTime`) — associé à l’horodatage du message Kafka.
* `_headers.name` (type `Array(String)`) et `_headers.value` (type `Array(String)`) — associés aux en-têtes des messages Kafka. Chaque paire `(_headers.name[i], _headers.value[i])` devient un en-tête Kafka. Comme `_headers.name` et `_headers.value` partagent le préfixe Nested `_headers`, ClickHouse exige que les deux tableaux aient la même taille pour chaque ligne.
Les colonnes portant ces noms sont **exclues de la charge utile du message** uniquement si leurs types correspondent à ceux indiqués ci-dessus ; sinon, elles restent dans la charge utile, de sorte que les schémas qui réutilisent ces noms par hasard pour des données sans rapport continuent de fonctionner.
Exemple :
```sql theme={null}
CREATE TABLE kafka_out
(
event_json String,
`_key` String,
`_timestamp` DateTime,
`_headers.name` Array(String),
`_headers.value` Array(String)
)
ENGINE = Kafka
SETTINGS
kafka_broker_list = 'broker:9092',
kafka_topic_list = 'events',
kafka_group_name = 'events-producer',
kafka_format = 'JSONEachRow',
kafka_map_virtual_columns_on_write = 1;
INSERT INTO kafka_out VALUES
('{\"a\":1}', 'session-42', now(), ['source', 'trace_id'], ['api', 'abc-123']);
```
Le message Kafka produit contient la charge utile `{"event_json":"{\"a\":1}"}`, la clé `session-42`, l’horodatage actuel et deux en-têtes `source=api` et `trace_id=abc-123`.
## Prise en charge des formats de données
Le moteur Kafka prend en charge tous les [formats](/fr/reference/formats/index) pris en charge par ClickHouse.
Le nombre de lignes dans un message Kafka dépend du fait que le format soit basé sur les lignes ou sur les blocs :
* Pour les formats basés sur les lignes, le nombre de lignes dans un message Kafka peut être contrôlé en définissant `kafka_max_rows_per_message`.
* Pour les formats basés sur les blocs, il n’est pas possible de diviser un bloc en parties plus petites, mais le nombre de lignes dans un bloc peut être contrôlé par le paramètre général [max\_block\_size](/fr/reference/settings/session-settings#max_block_size).
## Moteur pour stocker les offsets validés dans ClickHouse Keeper
Si `allow_experimental_kafka_offsets_storage_in_keeper` est activé, deux paramètres supplémentaires peuvent être spécifiés pour le moteur de table Kafka :
* `kafka_keeper_path` spécifie le chemin vers la table dans ClickHouse Keeper
* `kafka_replica_name` spécifie le nom de la réplique dans ClickHouse Keeper
Soit les deux paramètres doivent être spécifiés, soit aucun des deux. Lorsque les deux sont spécifiés, un nouveau moteur Kafka expérimental est utilisé. Ce nouveau moteur ne dépend pas du stockage des offsets validés dans Kafka, mais les stocke dans ClickHouse Keeper. Il essaie toujours de valider les offsets dans Kafka, mais il ne dépend de ces offsets qu’au moment de la création de la table. Dans tous les autres cas (si la table est redémarrée ou restaurée après une erreur), les offsets stockés dans ClickHouse Keeper sont utilisés pour reprendre la consommation des messages. En plus de l’offset validé, il stocke également le nombre de messages consommés dans le dernier lot, de sorte que, si l’insert échoue, le même nombre de messages sera consommé, ce qui permet la déduplication si nécessaire.
Exemple :
```sql theme={null}
CREATE TABLE experimental_kafka (key UInt64, value UInt64)
ENGINE = Kafka('localhost:19092', 'my-topic', 'my-consumer', 'JSONEachRow')
SETTINGS
kafka_keeper_path = '/clickhouse/{database}/{uuid}',
kafka_replica_name = '{replica}'
SETTINGS allow_experimental_kafka_offsets_storage_in_keeper=1;
```
### Limites connues
Comme le nouveau moteur est expérimental, il n’est pas encore prêt pour une utilisation en production. L’implémentation présente quelques limites connues :
* Supprimer puis recréer rapidement la table, ou spécifier le même chemin ClickHouse Keeper pour différents moteurs, peut entraîner des problèmes. Comme bonne pratique, vous pouvez utiliser `{uuid}` dans `kafka_keeper_path` pour éviter les conflits de chemins.
* Pour garantir des lectures répétables, les messages ne peuvent pas être consommés à partir de plusieurs partitions sur un seul thread. En revanche, les consommateurs Kafka doivent être interrogés régulièrement pour être maintenus actifs. En raison de ces deux contraintes, nous avons décidé d’autoriser la création de plusieurs consommateurs uniquement si `kafka_thread_per_consumer` est activé, sinon il est trop compliqué d’éviter les problèmes liés à l’interrogation régulière des consommateurs.
**Voir aussi**
* [Colonnes virtuelles](/fr/reference/engines/table-engines/index#table_engines-virtual_columns)
* [background\_message\_broker\_schedule\_pool\_size](/fr/reference/settings/server-settings/settings#background_message_broker_schedule_pool_size)
* [system.kafka\_consumers](/fr/reference/system-tables/kafka_consumers)