Passer au contenu principal
Ce moteur permet d’intégrer ClickHouse à NATS. NATS vous permet de :
  • Publier sur des sujets de messages ou vous y abonner.
  • Traiter les nouveaux messages dès qu’ils sont disponibles.

​
Création d’une table

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = NATS SETTINGS
    nats_url = 'host:port',
    nats_subjects = 'subject1,subject2,...',
    nats_format = 'data_format'[,]
    [nats_schema = '',]
    [nats_num_consumers = N,]
    [nats_queue_group = 'group_name',]
    [nats_secure = false,]
    [nats_max_reconnect = N,]
    [nats_reconnect_wait = N,]
    [nats_server_list = 'host1:port1,host2:port2,...',]
    [nats_skip_broken_messages = N,]
    [nats_max_block_size = N,]
    [nats_flush_interval_ms = N,]
    [nats_username = 'user',]
    [nats_password = 'password',]
    [nats_token = 'clickhouse',]
    [nats_credential_file = '/var/nats_credentials',]
    [nats_startup_connect_tries = 5,]
    [nats_max_rows_per_message = 1,]
    [nats_handle_error_mode = 'default']
Paramètres requis :
  • nats_url – hôte:port (par exemple, localhost:4222)..
  • nats_subjects – Liste des sujets auxquels la table NATS doit s’abonner ou publier. Prend en charge les sujets génériques comme foo.*.bar ou baz.>
  • nats_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 :
  • nats_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 et le nom de l’objet racine schema.capnp:Message.
  • nats_stream – Nom d’un stream existant dans NATS JetStream.
  • nats_consumer_name – Nom d’un durable pull consommateur existant dans NATS JetStream.
  • nats_num_consumers – Nombre de consommateurs par table. Par défaut : 1. Spécifiez davantage de consommateurs si le throughput d’un consommateur est insuffisant, uniquement pour NATS core.
  • nats_queue_group – Nom du groupe de queue pour les abonnés NATS. La valeur par défaut est le nom de la table.
  • nats_max_reconnect – Obsolète et sans effet ; la reconnexion est effectuée en permanence avec le délai d’expiration nats_reconnect_wait.
  • nats_reconnect_wait – Durée d’attente, en millisecondes, entre chaque tentative de reconnexion. Par défaut : 2000.
  • nats_server_list - Liste de serveurs pour la connexion. Peut être spécifiée pour se connecter à un cluster NATS.
  • nats_skip_broken_messages - Tolérance de l’analyseur de messages NATS aux messages incompatibles avec le schéma par bloc. Par défaut : 0. Si nats_skip_broken_messages = N, alors le moteur ignore N messages NATS qui ne peuvent pas être analysés (un message équivaut à une ligne de données).
  • nats_max_block_size - Nombre de lignes collectées par poll(s) pour vider les données depuis NATS. Par défaut : max_insert_block_size.
  • nats_flush_interval_ms - Délai d’expiration pour le vidage des données lues depuis NATS. Par défaut : stream_flush_interval_ms.
  • nats_username - Nom d’utilisateur NATS.
  • nats_password - Mot de passe NATS.
  • nats_token - Jeton d’authentification NATS.
  • nats_credential_file - Chemin vers un fichier d’identifiants NATS.
  • nats_startup_connect_tries - Nombre de tentatives de connexion au démarrage. Par défaut : 5.
  • nats_max_rows_per_message — Nombre maximal de lignes écrites dans un message NATS pour les formats basés sur les lignes. (par défaut : 1).
  • nats_handle_error_mode — Comment gérer les erreurs pour le moteur NATS. Valeurs possibles : default (une exception sera levée en cas d’échec de l’analyse d’un message), stream (le message d’exception et le message brut seront enregistrés dans les colonnes virtuelles _error et _raw_message).
Connexion SSL : Pour une connexion sécurisée, utilisez nats_secure = 1. La vérification du certificat est contrôlée par la variable d’environnement CLICKHOUSE_NATS_TLS_SECURE ; Si le certificat est expiré, autosigné, manquant ou autrement invalide, désactivez la vérification en définissant CLICKHOUSE_NATS_TLS_SECURE=0. Écriture dans la table NATS : Si la table lit uniquement à partir d’un seul sujet, toute insertion sera publiée sur ce même sujet. Cependant, si la table lit à partir de plusieurs sujets, nous devons préciser sur quel sujet nous voulons publier. C’est pourquoi, lors d’une insertion dans une table avec plusieurs sujets, le paramètre stream_like_engine_insert_queue est nécessaire. Vous pouvez sélectionner l’un des sujets depuis lesquels la table lit et y publier vos données. Par exemple : 
  CREATE TABLE queue (
    key UInt64,
    value UInt64
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1,subject2',
             nats_format = 'JSONEachRow';

  INSERT INTO queue
  SETTINGS stream_like_engine_insert_queue = 'subject2'
  VALUES (1, 1);
Des paramètres de format peuvent également être ajoutés en plus des paramètres liés à NATS. Exemple : 
  CREATE TABLE queue (
    key UInt64,
    value UInt64,
    date DateTime
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1',
             nats_format = 'JSONEachRow',
             date_time_input_format = 'best_effort';
La configuration du serveur NATS peut être ajoutée dans le fichier de configuration de ClickHouse. Plus précisément, vous pouvez y ajouter votre mot de passe pour le moteur NATS : 
<nats>
    <user>click</user>
    <password>house</password>
    <token>clickhouse</token>
</nats>

​
Description

SELECT n’est pas particulièrement utile pour lire les 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 NATS et le considérer 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 insère dans une table créée au préalable.
Lorsque la MATERIALIZED VIEW est rattachée au moteur, elle commence à collecter les données en arrière-plan. Cela vous permet de recevoir en continu des messages depuis NATS et de les convertir au format requis à l’aide de SELECT. Une table NATS peut avoir autant de vues matérialisées que vous le souhaitez ; elles ne lisent pas directement les données de la table, mais reçoivent les nouveaux enregistrements (par blocs). Vous pouvez ainsi écrire dans plusieurs tables avec différents niveaux de détail (avec regroupement/agrégation ou sans). Exemple : 
  CREATE TABLE queue (
    key UInt64,
    value UInt64
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1',
             nats_format = 'JSONEachRow',
             date_time_input_format = 'best_effort';

  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;
Pour ne plus recevoir de données de flux ou pour modifier la logique de conversion, détachez la vue matérialisée : 
  DETACH TABLE consumer;
  ATTACH TABLE consumer;
Si vous souhaitez modifier la table cible en utilisant ALTER, nous vous recommandons de désactiver la vue matérialisée afin d’éviter des écarts entre la table cible et les données de la vue.

​
Colonnes virtuelles

  • _subject - Sujet du message NATS. Type de données : String.
Colonnes virtuelles supplémentaires lorsque nats_handle_error_mode='stream' :
  • _raw_message - Message brut qui n’a pas pu être analysé avec succès. 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é avec succès.

​
Prise en charge des formats de données

Le moteur NATS prend en charge tous les formats pris en charge par ClickHouse. Le nombre de lignes dans un message NATS dépend du fait que le format soit basé sur des lignes ou sur des blocs :
  • Pour les formats basés sur des lignes, le nombre de lignes dans un message NATS peut être contrôlé en définissant nats_max_rows_per_message.
  • Pour les formats basés sur des 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.

​
Utilisation de JetStream

Avant d’utiliser le moteur NATS avec NATS JetStream, vous devez créer un stream NATS et un durable pull consommateur. Pour cela, vous pouvez par exemple utiliser l’utilitaire nats du paquet NATS CLI : 
$ nats stream add
? Stream Name stream_name
? Subjects stream_subject
? Storage file
? Replication 1
? Retention Policy Limits
? Discard Policy Old
? Stream Messages Limit -1
? Per Subject Messages Limit -1
? Total Stream Size -1
? Message TTL -1
? Max Message Size -1
? Duplicate tracking time window 2m0s
? Allow message Roll-ups No
? Allow message deletion Yes
? Allow purging subjects or the entire stream Yes
Stream stream_name was created

Information for Stream stream_name created 2025-10-03 14:12:51

                Subjects: stream_subject
                Replicas: 1
                 Storage: File

Options:

               Retention: Limits
         Acknowledgments: true
          Discard Policy: Old
        Duplicate Window: 2m0s
              Direct Get: true
       Allows Msg Delete: true
            Allows Purge: true
  Allows Per-Message TTL: false
          Allows Rollups: false

Limits:

        Maximum Messages: unlimited
     Maximum Per Subject: unlimited
           Maximum Bytes: unlimited
             Maximum Age: unlimited
    Maximum Message Size: unlimited
       Maximum Consumers: unlimited

State:

                Messages: 0
                   Bytes: 0 B
          First Sequence: 0
           Last Sequence: 0
        Active Consumers: 0

$ nats consumer add
? Select a Stream stream_name
? Consumer name consumer_name
? Delivery target (empty for Pull Consumers) 
? Start policy (all, new, last, subject, 1h, msg sequence) all
? Acknowledgment policy explicit
? Replay policy instant
? Filter Stream by subjects (blank for all) 
? Maximum Allowed Deliveries -1
? Maximum Acknowledgments Pending 0
? Deliver headers only without bodies No
? Add a Retry Backoff Policy No
Information for Consumer stream_name > consumer_name created 2025-10-03T14:13:51+03:00

Configuration:

                    Name: consumer_name
               Pull Mode: true
          Deliver Policy: All
              Ack Policy: Explicit
                Ack Wait: 30.00s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 512

State:

  Last Delivered Message: Consumer sequence: 0 Stream sequence: 0
    Acknowledgment Floor: Consumer sequence: 0 Stream sequence: 0
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 0
           Waiting Pulls: 0 of maximum 512
Après avoir créé le stream et le durable pull consommateur, nous pouvons créer une table avec le moteur NATS. Pour cela, vous devez renseigner : nats_stream, nats_consumer_name et nats_subjects : 
CREATE TABLE nats_jet_stream (
    key UInt64,
    value UInt64
  ) ENGINE NATS 
    SETTINGS  nats_url = 'localhost:4222',
              nats_stream = 'stream_name',
              nats_consumer_name = 'consumer_name',
              nats_subjects = 'stream_subject',
              nats_format = 'JSONEachRow';
Dernière modification le 29 juin 2026