Motor de tabla de NATS

Este motor permite integrar ClickHouse con NATS. NATS permite:

Publicar o suscribirse a subjects de NATS.
Procesar mensajes nuevos a medida que estén disponibles.

Crear una tabla

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_credentials = '-----BEGIN NATS USER JWT----- ...',]
    [nats_startup_connect_tries = 5,]
    [nats_max_rows_per_message = 1,]
    [nats_commit_on_select = false,]
    [nats_handle_error_mode = 'default']

Parámetros obligatorios:

nats_url – host:port (por ejemplo, localhost:4222)..
nats_subjects – Lista de subject de la tabla NATS a los que suscribirse o en los que publicar. Admite subjects comodín como foo.*.bar o baz.>
nats_format – Formato del mensaje. Usa la misma notación que la función SQL FORMAT, como JSONEachRow. Para obtener más información, consulta la sección Formats.

Parámetros opcionales:

nats_schema – Parámetro que debe usarse si el formato requiere una definición de schema. Por ejemplo, Cap’n Proto requiere la ruta al archivo de schema y el nombre del objeto raíz schema.capnp:Message.
nats_stream – El nombre de un stream existente en NATS JetStream.
nats_consumer_name – El nombre de un consumidor pull duradero existente en NATS JetStream.
nats_num_consumers – El número de consumidores por tabla. Valor predeterminado: 1. Especifica más consumidores si el throughput de un consumidor no es suficiente, solo para NATS core.
nats_queue_group – Nombre del queue group de los suscriptores de NATS. El valor predeterminado es el nombre de la tabla.
nats_max_reconnect – Deprecated y no tiene efecto; la reconexión se realiza permanentemente con el timeout nats_reconnect_wait.
nats_reconnect_wait – Tiempo de espera en milisegundos entre cada intento de reconexión. Valor predeterminado: 2000.
nats_server_list - Lista de servidores para la conexión. Puede especificarse para conectarse a un cluster de NATS.
nats_skip_broken_messages - Tolerancia del parser de mensajes NATS a mensajes incompatibles con el schema por bloque. Valor predeterminado: 0. Si nats_skip_broken_messages = N, el motor omite N mensajes NATS que no pueden parsearse (un mensaje equivale a una fila de datos).
nats_max_block_size - Número de filas recopiladas por poll(s) para el flushing de datos desde NATS. Valor predeterminado: max_insert_block_size.
nats_flush_interval_ms - Timeout para el flushing de los datos leídos desde NATS. Valor predeterminado: stream_flush_interval_ms.
nats_wait_for_flush_interval - Si es true, un ciclo de streaming en segundo plano permanece abierto durante todo el intervalo de flushing (nats_flush_interval_ms o, de lo contrario, stream_flush_interval_ms) en lugar de finalizar tan pronto como se vacía la cola del consumidor, lo que permite que se acumulen más mensajes en un único bloque a costa de hasta un intervalo de flushing de latencia de ingestión adicional. Valor predeterminado: false (comportamiento de vaciado y continuación de baja latencia).
nats_username - Nombre de usuario de NATS. Cuando se almacena en una colección con nombre definida en el archivo de configuración del servidor, la consulta no puede sobrescribir el nats_url ni el nats_server_list de la colección.
nats_password - Contraseña de NATS. Cuando se almacena en una colección con nombre definida en el archivo de configuración del servidor, la consulta no puede sobrescribir el nats_url ni el nats_server_list de la colección.
nats_token - Token de autenticación de NATS. Cuando se almacena en una colección con nombre definida en el archivo de configuración del servidor, la consulta no puede sobrescribir el nats_url ni el nats_server_list de la colección.
nats_credential_file - Ruta a un archivo de credentials de NATS. Solo se acepta desde una colección con nombre definida en el archivo de configuración del servidor cuyo nats_url y nats_server_list no sean sobrescritos por la consulta, porque el servidor abre la ruta con sus propios privilegios. En una consulta, pasa en su lugar el contenido del archivo en nats_credentials.
nats_credentials - Contenido de las credentials de NATS (la misma carga útil que en un archivo .creds con JWT de usuario y seed). Dado que es la única forma que puede usar una consulta, reemplaza un nats_credential_file heredado de una colección con nombre en lugar de entrar en conflicto con él, a menos que el operador haya bloqueado esa ruta con <nats_credential_file overridable="false">. No se le puede asignar la cadena vacía para eliminar las credentials que contiene una colección con nombre.
nats_ca_file - Ruta a un archivo con los certificados de CA de confianza usados para verificar el certificado del servidor NATS. Requiere nats_secure. Al igual que nats_credential_file, solo se acepta desde una colección con nombre definida en el archivo de configuración del servidor cuyo nats_url y nats_server_list no sean sobrescritos por la consulta, porque el servidor abre la ruta con sus propios privilegios.
nats_client_cert_file - Ruta al certificado de client presentado al servidor NATS. Requiere nats_secure y nats_client_key_file. Se acepta desde las mismas fuentes que nats_ca_file.
nats_client_key_file - Ruta a la clave privada de nats_client_cert_file. Se acepta desde las mismas fuentes que nats_ca_file.
nats_startup_connect_tries - Número de intentos de conexión al inicio. Valor predeterminado: 5.
nats_max_rows_per_message — El número máximo de filas escritas en un mensaje NATS para formatos basados en filas. (valor predeterminado: 1).
nats_commit_on_select - Hace commit de los mensajes cuando se realiza una consulta. Solo se aplica a JetStream; NATS core no tiene confirmaciones. Valor predeterminado: 0.
nats_handle_error_mode — Cómo gestionar los errores del motor NATS. Valores posibles: default (se lanzará una excepción si no se puede parsear un mensaje), stream (el mensaje de excepción y el mensaje sin procesar se guardarán en las columnas virtuales _error y _raw_message).

Conexión SSL: Para una conexión segura, usa nats_secure = 1. La verificación de certificados se controla mediante la variable de entorno CLICKHOUSE_NATS_TLS_SECURE; Si el certificado está expirado, es autofirmado, falta o no es válido por algún otro motivo, desactiva la verificación estableciendo CLICKHOUSE_NATS_TLS_SECURE=0. Un certificado de servidor firmado por una CA privada se verifica indicando el certificado de CA en nats_ca_file, lo cual es preferible a desactivar la verificación. Cuando el servidor requiere certificados de cliente, proporciónalos con nats_client_cert_file y nats_client_key_file. Los tres son settings del operador: provienen de una colección con nombre definida en el archivo de configuración del servidor. Cada archivo se lee cuando la tabla se conecta, por lo que uno ilegible o malformado hace que falle la consulta en lugar del handshake. Escritura en tabla NATS: Si la tabla lee de un solo subject, cualquier insert publicará en ese mismo subject. Sin embargo, si la tabla lee de varios subjects, es necesario especificar en qué subject queremos publicar. Por eso, al insertar en una tabla con varios subjects, se necesita el setting stream_like_engine_insert_queue. Puedes seleccionar uno de los subjects de los que lee la tabla y publicar allí tus datos. Por ejemplo:

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);

También se pueden añadir settings de formato junto con los settings relacionados con NATS. Ejemplo:

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 configuración del servidor NATS se puede añadir mediante el archivo de configuración de ClickHouse. En concreto, puedes añadir tu contraseña para el motor NATS:

<nats>
    <user>click</user>
    <password>house</password>
    <token>clickhouse</token>
</nats>

Descripción

SELECT no es especialmente útil para leer mensajes (salvo para depuración), porque cada mensaje solo puede leerse una vez. Es más práctico crear flujos en tiempo real mediante vistas materializadas. Para ello:

Use el motor para crear un consumidor de NATS y considérelo un flujo de datos.
Cree una tabla con la estructura deseada.
Cree una vista materializada que convierta los datos del motor y los inserte en una tabla creada previamente.

Cuando la MATERIALIZED VIEW se conecta al motor, comienza a recopilar datos en segundo plano. Esto le permite recibir continuamente mensajes de NATS y convertirlos al formato requerido mediante SELECT. Una tabla de NATS puede tener tantas vistas materializadas como desee; no leen datos de la tabla directamente, sino que reciben nuevos registros (en bloques). De este modo, puede escribir en varias tablas con distintos niveles de detalle (con agrupación - agregación y sin ella). Ejemplo:

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;

Para dejar de recibir datos de los streams o cambiar la lógica de conversión, desvincula la vista materializada:

DETACH TABLE consumer;
ATTACH TABLE consumer;

Si quiere cambiar la tabla de destino mediante ALTER, le recomendamos deshabilitar la vista materializada para evitar discrepancias entre la tabla de destino y los datos de la vista.

Columnas virtuales

_subject - subject del mensaje de NATS. Tipo de dato: String.

Columnas virtuales adicionales cuando nats_handle_error_mode='stream':

_raw_message - Mensaje sin procesar que no pudo analizarse correctamente. Tipo de dato: Nullable(String).
_error - Mensaje de excepción producido durante un error de análisis. Tipo de dato: Nullable(String).

Nota: las columnas virtuales _raw_message y _error se rellenan solo en caso de excepción durante el análisis; siempre son NULL cuando el mensaje se ha analizado correctamente.

Compatibilidad con formatos de datos

El motor NATS es compatible con todos los formatos compatibles con ClickHouse. El número de filas en un mensaje de NATS depende de si el formato está basado en filas o en bloques:

En los formatos basados en filas, el número de filas en un mensaje de NATS puede controlarse mediante la configuración nats_max_rows_per_message.
En los formatos basados en bloques, no es posible dividir un bloque en partes más pequeñas, pero el número de filas de un bloque puede controlarse mediante la configuración general max_block_size.

Uso de JetStream

Antes de usar el motor NATS con NATS JetStream, debe crear un stream de NATS y un consumidor pull duradero. Para ello, puede usar, por ejemplo, la utilidad nats del paquete NATS CLI:

creación de stream

$ 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

creación de consumidor pull duradero

$ 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

Después de crear el stream y el consumidor pull duradero, podemos crear una tabla con el motor NATS. Para ello, debe inicializar: nats_stream, nats_consumer_name y 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';

Las tablas de JetStream ofrecen entrega al menos una vez: un mensaje se confirma solo después de insertarse en las vistas materializadas dependientes, por lo que un mensaje cuya inserción falla o se interrumpe permanece sin confirmar y se vuelve a entregar. NATS Core (sin JetStream) no tiene confirmación ni reproducción, por lo que ofrece entrega como máximo una vez y un mensaje interrumpido se pierde.

Durabilidad de los datos

Esta sección se aplica únicamente a JetStream. Core NATS no tiene confirmaciones y ofrece entrega como máximo una vez, como se describió anteriormente, por lo que no existe ninguna ventana en la que pueda perderse un mensaje confirmado. Una tabla de JetStream puede perder silenciosamente filas ya consumidas si la caché de páginas del SO se descarta antes de que los datos insertados se escriban en disco. Después de enviar un lote a las vistas materializadas dependientes, el consumidor confirma esos mensajes, lo que permite que el stream avance más allá de ellos. Sin embargo, las filas insertadas solo son duraderas cuando la parte de destino se sincroniza mediante fsync, algo que, de forma predeterminada, no ocurre de manera síncrona (fsync_after_insert = 0). Si la caché de páginas se pierde después de la confirmación pero antes de que la parte de destino se sincronice mediante fsync, los mensajes ya no se vuelven a entregar, por lo que las filas se pierden sin que se produzca ningún error y count() simplemente es menor. La terminación forzada de un proceso no pone esto de manifiesto, porque el kernel conserva la caché de páginas y acaba escribiéndola en disco. En cambio, la pérdida de la caché de páginas sí lo revela; por ejemplo, una pérdida de alimentación a nivel de dispositivo o un reinicio no limpio del host o del kernel. Para la ruta recomendada de consumo mediante vistas materializadas (la confirmación se envía solo después de que finalice toda la canalización de inserción), establecer fsync_after_insert = 1 (y fsync_part_directory = 1) en las tablas MergeTree de destino garantiza que las partes insertadas sean duraderas antes de enviar la confirmación, lo que reduce considerablemente esta ventana. La configuración debe habilitarse en cada tabla MergeTree en la que se inserte el lote, incluidos los destinos de vistas materializadas en cascada; cualquier tabla de este tipo que mantenga el valor predeterminado aún puede perder su parte. Los intermediarios asíncronos no obtienen durabilidad únicamente con esta configuración: por ejemplo, un destino Distributed inserta en segundo plano cuando distributed_foreground_insert = 0, que es el valor predeterminado fuera de ClickHouse Cloud, por lo que requiere su propia configuración de durabilidad o una inserción síncrona. Esta mitigación tampoco se aplica a un INSERT ... SELECT ... FROM <nats_table> directo con nats_commit_on_select = 1, donde los mensajes se confirman cuando la lectura llega a su fin, en lugar de después de que el destino haya escrito una parte duradera.

Referencia de SQL

Tipos de datos

Motores

Funciones

Formatos

Configuración

Tablas del sistema

Lagos de datos

Crear una tabla

Descripción

Columnas virtuales

Compatibilidad con formatos de datos

Uso de JetStream

Durabilidad de los datos

​Crear una tabla

​Descripción

​Columnas virtuales

​Compatibilidad con formatos de datos

​Uso de JetStream

​Durabilidad de los datos

Crear una tabla

Descripción

Columnas virtuales

Compatibilidad con formatos de datos

Uso de JetStream

Durabilidad de los datos