Passer au contenu principal
Ce moteur permet d’intégrer ClickHouse à RabbitMQ. 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

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.
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 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.
  • rabbitmq_flush_interval_ms - Délai d’attente avant le vidage des données depuis RabbitMQ. Par défaut : 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 : 
  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 : 
 <rabbitmq>
    <username>root</username>
    <password>clickhouse</password>
 </rabbitmq>
Configuration supplémentaire : 
 <rabbitmq>
    <vhost>clickhouse</vhost>
 </rabbitmq>

​
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. 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. Si le bloc n’a pas été formé dans les stream_flush_interval_ms 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 : 
  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 (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 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.
Dernière modification le 29 juin 2026