> ## Documentation Index > Fetch the complete documentation index at: https://private-7c7dfe99-mintlify-fbfa8bee.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. > Ce moteur permet d’intégrer ClickHouse à RabbitMQ. # Moteur de table RabbitMQ Ce moteur permet d’intégrer ClickHouse à [RabbitMQ](https://www.rabbitmq.com). `RabbitMQ` vous permet de : * publier ou de vous abonner à des flux de données ; * traiter les flux dès qu’ils sont disponibles.
## Créer une table
```sql theme={null} CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster] ( name1 [type1], name2 [type2], ... ) ENGINE = RabbitMQ SETTINGS rabbitmq_host_port = 'host:port' [or rabbitmq_address = 'amqp(s)://guest:guest@localhost/vhost'], rabbitmq_exchange_name = 'exchange_name', rabbitmq_format = 'data_format'[,] [rabbitmq_exchange_type = 'exchange_type',] [rabbitmq_routing_key_list = 'key1,key2,...',] [rabbitmq_secure = 0,] [rabbitmq_schema = '',] [rabbitmq_num_consumers = N,] [rabbitmq_num_queues = N,] [rabbitmq_queue_base = 'queue',] [rabbitmq_persistent = 0,] [rabbitmq_skip_broken_messages = N,] [rabbitmq_max_block_size = N,] [rabbitmq_flush_interval_ms = N,] [rabbitmq_queue_settings_list = 'x-dead-letter-exchange=my-dlx,x-max-length=10,x-overflow=reject-publish',] [rabbitmq_queue_consume = false,] [rabbitmq_address = '',] [rabbitmq_vhost = '/',] [rabbitmq_username = '',] [rabbitmq_password = '',] [rabbitmq_commit_on_select = false,] [rabbitmq_max_rows_per_message = 1,] [rabbitmq_handle_error_mode = 'default'] ``` Paramètres requis : * `rabbitmq_host_port` – hôte:port (par exemple, `localhost:5672`). * `rabbitmq_exchange_name` – nom de l’exchange RabbitMQ. * `rabbitmq_format` – format du message. Utilise la même notation que la fonction SQL `FORMAT`, par exemple `JSONEachRow`. Pour plus d’informations, consultez la section [Formats](/fr/reference/formats/index). Paramètres facultatifs : * `rabbitmq_exchange_type` – Le type d’exchange RabbitMQ : `direct`, `fanout`, `topic`, `headers`, `consistent_hash`. Par défaut : `fanout`. * `rabbitmq_routing_key_list` – Une liste de clés de routage séparées par des virgules. * `rabbitmq_schema` – Paramètre à utiliser si le format nécessite une définition de schéma. Par exemple, [Cap'n Proto](https://capnproto.org/) nécessite le chemin du fichier de schéma ainsi que le nom de l’objet racine `schema.capnp:Message`. * `rabbitmq_num_consumers` – Le nombre de consommateurs par table. Spécifiez davantage de consommateurs si le débit d’un consommateur est insuffisant. Par défaut : `1` * `rabbitmq_num_queues` – Nombre total de files. Augmenter ce nombre peut améliorer considérablement les performances. Par défaut : `1`. * `rabbitmq_queue_base` - Spécifie un préfixe pour les noms de file. Les cas d’utilisation de ce paramètre sont décrits ci-dessous. * `rabbitmq_persistent` - Si défini sur 1 (true), le mode de remise de la requête d’insertion sera défini sur 2 (marque les messages comme 'persistent'). Par défaut : `0`. * `rabbitmq_skip_broken_messages` – Tolérance de l’analyseur de messages RabbitMQ aux messages incompatibles avec le schéma par bloc. Si `rabbitmq_skip_broken_messages = N`, le moteur ignore alors *N* messages RabbitMQ qui ne peuvent pas être analysés (un message équivaut à une ligne de données). Par défaut : `0`. * `rabbitmq_max_block_size` - Nombre de lignes collectées avant le vidage des données depuis RabbitMQ. Par défaut : [max\_insert\_block\_size](/fr/reference/settings/session-settings#max_insert_block_size). * `rabbitmq_flush_interval_ms` - Délai d’attente avant le vidage des données depuis RabbitMQ. Par défaut : [stream\_flush\_interval\_ms](/fr/reference/settings/session-settings#stream_flush_interval_ms). * `rabbitmq_queue_settings_list` - permet de définir les paramètres RabbitMQ lors de la création d’une file. Paramètres disponibles : `x-max-length`, `x-max-length-bytes`, `x-message-ttl`, `x-expires`, `x-priority`, `x-max-priority`, `x-overflow`, `x-dead-letter-exchange`, `x-queue-type`. Le paramètre `durable` est activé automatiquement pour la file. * `rabbitmq_address` - Adresse de connexion. Utilisez soit ce paramètre, soit `rabbitmq_host_port`. * `rabbitmq_vhost` - vhost RabbitMQ. Par défaut : `'/'`. * `rabbitmq_queue_consume` - Utilise des files définies par l’utilisateur et n’effectue aucune configuration RabbitMQ : déclaration des exchanges, des files et des liaisons. Par défaut : `false`. * `rabbitmq_username` - Nom d’utilisateur RabbitMQ. * `rabbitmq_password` - Mot de passe RabbitMQ. * `reject_unhandled_messages` - Rejette les messages (envoie un accusé de réception négatif RabbitMQ) en cas d’erreur. Ce paramètre est activé automatiquement si un `x-dead-letter-exchange` est défini dans `rabbitmq_queue_settings_list`. * `rabbitmq_commit_on_select` - Valide les messages lorsqu’une requête SELECT est effectuée. Par défaut : `false`. * `rabbitmq_max_rows_per_message` — Le nombre maximal de lignes écrites dans un message RabbitMQ pour les formats basés sur les lignes. Par défaut : `1`. * `rabbitmq_empty_queue_backoff_start_ms` — Point de départ du backoff pour replanifier la lecture si la file RabbitMQ est vide. * `rabbitmq_empty_queue_backoff_end_ms` — Point de fin du backoff pour replanifier la lecture si la file RabbitMQ est vide. * `rabbitmq_empty_queue_backoff_step_ms` — Pas du backoff pour replanifier la lecture si la file RabbitMQ est vide. * `rabbitmq_handle_error_mode` — Mode de gestion des erreurs du moteur RabbitMQ. Valeurs possibles : default (une exception sera levée si l’analyse d’un message échoue), stream (le message d’exception et le message brut seront enregistrés dans les colonnes virtuelles `_error` et `_raw_message`), dead\_letter\_queue (les données liées aux erreurs seront enregistrées dans `system.dead_letter_queue`).
### Connexion SSL
Utilisez soit `rabbitmq_secure = 1`, soit `amqps` dans l'adresse de connexion : `rabbitmq_address = 'amqps://guest:guest@localhost/vhost'`. Par défaut, la bibliothèque utilisée ne vérifie pas si la connexion TLS établie est suffisamment sécurisée. Que le certificat soit expiré, autosigné, absent ou invalide, la connexion est simplement autorisée. Une vérification plus stricte des certificats pourra être mise en œuvre à l'avenir. Des paramètres de format peuvent également être ajoutés avec les paramètres liés à RabbitMQ. Exemple : ```sql theme={null} CREATE TABLE queue ( key UInt64, value UInt64, date DateTime ) ENGINE = RabbitMQ SETTINGS rabbitmq_host_port = 'localhost:5672', rabbitmq_exchange_name = 'exchange1', rabbitmq_format = 'JSONEachRow', rabbitmq_num_consumers = 5, date_time_input_format = 'best_effort'; ``` La configuration du serveur RabbitMQ doit être ajoutée dans le fichier de configuration de ClickHouse. Configuration requise : ```xml theme={null} root clickhouse ``` Configuration supplémentaire : ```xml theme={null} clickhouse ```
## Description
`SELECT` n’est pas particulièrement utile pour lire des messages (sauf pour le débogage), car chaque message ne peut être lu qu’une seule fois. Il est plus pratique de créer des flux en temps réel à l’aide de [vues matérialisées](/fr/reference/statements/create/view). Pour ce faire : 1. Utilisez le moteur pour créer un consommateur RabbitMQ et considérez-le comme un flux de données. 2. Créez une table avec la structure souhaitée. 3. Créez une vue matérialisée qui convertit les données du moteur et les place dans une table créée précédemment. Lorsque la `MATERIALIZED VIEW` est liée au moteur, elle commence à collecter les données en arrière-plan. Cela vous permet de recevoir en continu des messages depuis RabbitMQ et de les convertir au format requis à l’aide de `SELECT`. Une table RabbitMQ peut avoir autant de vues matérialisées que vous le souhaitez. Les données peuvent être acheminées en fonction de `rabbitmq_exchange_type` et de la `rabbitmq_routing_key_list` spécifiée. Il ne peut y avoir qu’un seul exchange par table. Un exchange peut être partagé entre plusieurs tables, ce qui permet un routage vers plusieurs tables en même temps. Options de type d’exchange : * `direct` - Le routage est basé sur la correspondance exacte des clés. Exemple de liste de clés de table : `key1,key2,key3,key4,key5` ; la clé du message peut être égale à l’une d’elles. * `fanout` - Routage vers toutes les tables (où le nom de l’exchange est identique), quelles que soient les clés. * `topic` - Le routage est basé sur des patterns avec des clés séparées par des points. Exemples : `*.logs`, `records.*.*.2020`, `*.2018,*.2019,*.2020`. * `headers` - Le routage est basé sur des correspondances `key=value` avec un paramètre `x-match=all` ou `x-match=any`. Exemple de liste de clés de table : `x-match=all,format=logs,type=report,year=2020`. * `consistent_hash` - Les données sont réparties uniformément entre toutes les tables liées (où le nom de l’exchange est identique). Notez que ce type d’exchange doit être activé avec le plugin RabbitMQ : `rabbitmq-plugins enable rabbitmq_consistent_hash_exchange`. Le paramètre `rabbitmq_queue_base` peut être utilisé dans les cas suivants : * permettre à différentes tables de partager des files, afin que plusieurs consommateurs puissent être enregistrés pour les mêmes files, ce qui améliore les performances. Si vous utilisez les paramètres `rabbitmq_num_consumers` et/ou `rabbitmq_num_queues`, une correspondance exacte des files est obtenue si ces paramètres sont identiques. * pouvoir reprendre la lecture à partir de certaines files durables lorsque tous les messages n’ont pas été consommés avec succès. Pour reprendre la consommation à partir d’une file spécifique, définissez son nom dans le paramètre `rabbitmq_queue_base` et ne spécifiez pas `rabbitmq_num_consumers` ni `rabbitmq_num_queues` (valeur par défaut : 1). Pour reprendre la consommation à partir de toutes les files déclarées pour une table spécifique, indiquez simplement les mêmes paramètres : `rabbitmq_queue_base`, `rabbitmq_num_consumers`, `rabbitmq_num_queues`. Par défaut, les noms des files seront uniques pour chaque table. * réutiliser des files, puisqu’elles sont déclarées durables et ne sont pas supprimées automatiquement. (Elles peuvent être supprimées via n’importe lequel des outils CLI de RabbitMQ.) Pour améliorer les performances, les messages reçus sont regroupés en blocs de la taille de [max\_insert\_block\_size](/fr/reference/settings/session-settings#max_insert_block_size). Si le bloc n’a pas été formé dans les [stream\_flush\_interval\_ms](/fr/reference/settings/server-settings/settings) millisecondes, les données seront écrites dans la table, même si le bloc n’est pas complet. Si les paramètres `rabbitmq_num_consumers` et/ou `rabbitmq_num_queues` sont spécifiés avec `rabbitmq_exchange_type`, alors : * le plugin `rabbitmq-consistent-hash-exchange` doit être activé. * la propriété `message_id` des messages publiés doit être spécifiée (unique pour chaque message/lot). Pour la requête d’insertion, des métadonnées de message sont ajoutées pour chaque message publié : `messageID` et l’indicateur `republished` (`true` s’il a été publié plus d’une fois) - ils sont accessibles via les en-têtes du message. N’utilisez pas la même table pour les inserts et les vues matérialisées. Exemple : ```sql theme={null} CREATE TABLE queue ( key UInt64, value UInt64 ) ENGINE = RabbitMQ SETTINGS rabbitmq_host_port = 'localhost:5672', rabbitmq_exchange_name = 'exchange1', rabbitmq_exchange_type = 'headers', rabbitmq_routing_key_list = 'format=logs,type=report,year=2020', rabbitmq_format = 'JSONEachRow', rabbitmq_num_consumers = 5; CREATE TABLE daily (key UInt64, value UInt64) ENGINE = MergeTree() ORDER BY key; CREATE MATERIALIZED VIEW consumer TO daily AS SELECT key, value FROM queue; SELECT key, value FROM daily ORDER BY key; ```
## Colonnes virtuelles
* `_exchange_name` - Nom de l’exchange RabbitMQ. Type de données : `String`. * `_channel_id` - ChannelID sur lequel le consommateur ayant reçu le message a été déclaré. Type de données : `String`. * `_delivery_tag` - DeliveryTag du message reçu. Limité à ce canal. Type de données : `UInt64`. * `_redelivered` - indicateur `redelivered` du message. Type de données : `UInt8`. * `_message_id` - messageID du message reçu ; non vide s’il a été défini lors de la publication du message. Type de données : `String`. * `_timestamp` - horodatage du message reçu ; non vide s’il a été défini lors de la publication du message. Type de données : `UInt64`. Colonnes virtuelles supplémentaires lorsque `rabbitmq_handle_error_mode='stream'` : * `_raw_message` - Message brut qui n’a pas pu être analysé correctement. Type de données : `Nullable(String)`. * `_error` - Message d’exception survenu lors de l’échec de l’analyse. Type de données : `Nullable(String)`. Remarque : les colonnes virtuelles `_raw_message` et `_error` ne sont renseignées qu’en cas d’exception lors de l’analyse ; elles sont toujours `NULL` lorsque le message a été analysé correctement.
## Mises en garde
Même si vous pouvez spécifier des [expressions de colonne par défaut](/fr/reference/statements/create/table#default_values) (comme `DEFAULT`, `MATERIALIZED`, `ALIAS`) dans la définition de la table, elles seront ignorées. À la place, les colonnes seront renseignées avec les valeurs par défaut correspondant à leur type.
## Prise en charge des formats de données
Le moteur RabbitMQ prend en charge tous les [formats](/fr/reference/formats/index) pris en charge par ClickHouse. Le nombre de lignes dans un message RabbitMQ dépend du fait que le format soit orienté lignes ou orienté blocs : * Pour les formats orientés lignes, le nombre de lignes dans un message RabbitMQ peut être contrôlé en définissant `rabbitmq_max_rows_per_message`. * Pour les formats orientés blocs, il n’est pas possible de diviser un block en parties plus petites, mais le nombre de lignes dans un block peut être contrôlé par le paramètre général [max\_block\_size](/fr/reference/settings/session-settings#max_block_size).