Движок таблицы NATS

Этот движок позволяет интегрировать ClickHouse с NATS для публикации сообщений в subject, подписки на них и обработки новых сообщений по мере их поступления.

Этот движок позволяет интегрировать ClickHouse с NATS.

NATS позволяет:

• Публиковать сообщения в subject и подписываться на них.
• Обрабатывать новые сообщения по мере их поступления.

​

Создание таблицы

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']

Обязательные параметры:

• nats_url – host:port (например, localhost:4222)..
• nats_subjects – Список subject для таблицы NATS, на которые нужно подписываться или в которые нужно публиковать сообщения. Поддерживаются subject с подстановочными символами, например foo.*.bar или baz.>
• nats_format – Формат сообщений. Используется та же нотация, что и в SQL-функции FORMAT, например JSONEachRow. Подробнее см. в разделе Форматы.

Необязательные параметры:

• nats_schema – Параметр, который необходимо использовать, если формат требует определения схемы. Например, Cap’n Proto требует путь к файлу схемы и имя корневого объекта schema.capnp:Message.
• nats_stream – Имя существующего stream в NATS JetStream.
• nats_consumer_name – Имя существующего durable pull consumer в NATS JetStream.
• nats_num_consumers – Количество consumers на таблицу. По умолчанию: 1. Укажите больше consumers, если пропускной способности одного consumer недостаточно; применяется только к NATS core.
• nats_queue_group – Имя queue group для подписчиков NATS. По умолчанию используется имя таблицы.
• nats_max_reconnect – Устарел и не имеет эффекта; переподключение выполняется постоянно с тайм-аутом nats_reconnect_wait.
• nats_reconnect_wait – Время ожидания в миллисекундах между попытками переподключения. По умолчанию: 2000.
• nats_server_list - Список серверов для подключения. Можно указать для подключения к кластеру NATS.
• nats_skip_broken_messages - Допустимое число несовместимых со схемой сообщений для parser сообщений NATS на блок. По умолчанию: 0. Если nats_skip_broken_messages = N, то движок пропускает N сообщений NATS, которые не удаётся разобрать (одно сообщение соответствует одной строке данных).
• nats_max_block_size - Число строк, собираемых с помощью poll(s) для сброса данных из NATS. По умолчанию: max_insert_block_size.
• nats_flush_interval_ms - Тайм-аут для сброса данных, прочитанных из NATS. По умолчанию: stream_flush_interval_ms.
• nats_wait_for_flush_interval - Если true, фоновый цикл стриминга остаётся открытым на весь интервал сброса (nats_flush_interval_ms или, в противном случае, stream_flush_interval_ms) вместо завершения сразу после опустошения очереди consumer, что позволяет накопить больше сообщений в одном блоке ценой дополнительной задержки ингестии до одного интервала сброса. По умолчанию: false (поведение с низкой задержкой: опустошить очередь и завершить).
• nats_username - Имя пользователя NATS. Если оно хранится в именованной коллекции, определённой в файле конфигурации сервера, запрос не может переопределить nats_url или nats_server_list этой коллекции.
• nats_password - Пароль NATS. Если он хранится в именованной коллекции, определённой в файле конфигурации сервера, запрос не может переопределить nats_url или nats_server_list этой коллекции.
• nats_token - Токен аутентификации NATS. Если он хранится в именованной коллекции, определённой в файле конфигурации сервера, запрос не может переопределить nats_url или nats_server_list этой коллекции.
• nats_credential_file - Путь к файлу учётных данных NATS. Принимается только из именованной коллекции, определённой в файле конфигурации сервера, если её nats_url и nats_server_list не переопределяются запросом, поскольку сервер открывает этот путь со своими правами. Вместо этого в запросе передайте содержимое файла в nats_credentials.
• nats_credentials - Содержимое учётных данных NATS (та же полезная нагрузка, что и в файле .creds с JWT пользователя и seed). Поскольку это единственный вариант, доступный в запросе, он заменяет nats_credential_file, унаследованный из именованной коллекции, а не конфликтует с ним, если только оператор не заблокировал этот путь с помощью <nats_credential_file overridable="false">. Ему нельзя присвоить пустую строку, чтобы удалить учётные данные, содержащиеся в именованной коллекции.
• nats_ca_file - Путь к файлу с доверенными CA‑сертификатами, используемыми для проверки сертификата сервера NATS. Требует nats_secure. Как и nats_credential_file, принимается только из именованной коллекции, определённой в файле конфигурации сервера, если её nats_url и nats_server_list не переопределяются запросом, поскольку сервер открывает этот путь со своими правами.
• nats_client_cert_file - Путь к клиентскому сертификату, предоставляемому серверу NATS. Требует nats_secure и nats_client_key_file. Принимается из тех же источников, что и nats_ca_file.
• nats_client_key_file - Путь к закрытому ключу nats_client_cert_file. Принимается из тех же источников, что и nats_ca_file.
• nats_startup_connect_tries - Количество попыток подключения при запуске. По умолчанию: 5.
• nats_max_rows_per_message — Максимальное количество строк, записываемых в одно сообщение NATS для построчных форматов. (по умолчанию: 1).
• nats_commit_on_select - Выполнять коммит сообщений при выполнении запроса. Применяется только к JetStream; в core NATS нет подтверждений. По умолчанию: 0.
• nats_handle_error_mode — Как обрабатывать ошибки движка NATS. Возможные значения: default (если не удаётся разобрать сообщение, будет сгенерировано исключение), stream (сообщение об исключении и исходное сообщение будут сохранены в виртуальных столбцах _error и _raw_message).

SSL-соединение:

Для защищённого соединения используйте nats_secure = 1. Проверка сертификата управляется переменной окружения CLICKHOUSE_NATS_TLS_SECURE; если сертификат просрочен, самоподписан, отсутствует или иным образом недействителен, отключите проверку, задав CLICKHOUSE_NATS_TLS_SECURE=0.

Серверный сертификат, подписанный приватным CA, проверяется путём указания в nats_ca_file пути к CA‑сертификату, что предпочтительнее отключения проверки. Если сервер требует клиентские сертификаты, передайте их через nats_client_cert_file и nats_client_key_file. Все три являются настройками уровня оператора: они берутся из named collection, определённой в файле конфигурации сервера. Каждый файл читается при подключении таблицы, поэтому нечитаемый или некорректный файл приводит к ошибке запроса, а не рукопожатия.

Запись в таблицу NATS:

Если таблица читает только из одного subject, любая вставка будет публиковаться в тот же subject. Однако, если таблица читает из нескольких subject, необходимо указать, в какой subject публиковать данные. Поэтому при вставке в таблицу с несколькими subject требуется установка параметра stream_like_engine_insert_queue. Вы можете выбрать один из subject, из которых читает таблица, и публиковать данные туда. Например:

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

Кроме того, вместе с настройками, относящимися к nats, можно указать настройки формата.

Пример:

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

Конфигурацию сервера NATS можно добавить в файл конфигурации ClickHouse. В частности, можно указать пароль для движка NATS:

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

​

Описание

SELECT не особенно полезен для чтения сообщений (кроме отладки), поскольку каждое сообщение можно прочитать только один раз. Гораздо практичнее создавать потоки в реальном времени с помощью materialized views. Для этого:

1. Используйте движок, чтобы создать consumer NATS, и рассматривайте его как поток данных.
2. Создайте таблицу с нужной структурой.
3. Создайте materialized view, которое преобразует данные из движка и помещает их в ранее созданную таблицу.

Когда MATERIALIZED VIEW подключается к движку, оно начинает собирать данные в фоновом режиме. Это позволяет непрерывно получать сообщения из NATS и преобразовывать их в требуемый формат с помощью SELECT. Одна таблица NATS может иметь сколько угодно materialized views; они не читают данные из таблицы напрямую, а получают новые записи (блоками), поэтому вы можете записывать данные в несколько таблиц с разным уровнем детализации (с группировкой — агрегацией — и без неё).

Пример:

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;

Чтобы перестать получать данные из потоков или изменить логику преобразования, отсоедините materialized view:

DETACH TABLE consumer;
ATTACH TABLE consumer;

Если вы хотите изменить целевую таблицу с помощью ALTER, мы рекомендуем отключить materialized view, чтобы избежать расхождений между целевой таблицей и данными из представления.

​

Виртуальные столбцы

• _subject - subject сообщения NATS. Тип данных: String.

Дополнительные виртуальные столбцы, если nats_handle_error_mode='stream':

• _raw_message - Исходное сообщение, которое не удалось успешно разобрать. Тип данных: Nullable(String).
• _error - Сообщение об исключении, возникшем при сбое разбора. Тип данных: Nullable(String).

Примечание: виртуальные столбцы _raw_message и _error заполняются только в случае исключения при разборе; если сообщение успешно разобрано, они всегда имеют значение NULL.

​

Поддержка форматов данных

Движок NATS поддерживает все форматы, доступные в ClickHouse. Количество строк в одном сообщении NATS зависит от того, является ли формат построчным или блочным:

• Для построчных форматов количеством строк в одном сообщении NATS можно управлять с помощью настройки nats_max_rows_per_message.
• Для блочных форматов блок нельзя разделить на более мелкие части, однако количество строк в одном блоке можно регулировать с помощью общей настройки max_block_size.

​

Использование JetStream

Перед использованием движка NATS с NATS JetStream необходимо создать stream в NATS и durable pull consumer. Для этого можно использовать, например, утилиту nats из пакета NATS CLI:

создание 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

создание durable pull consumer

$ 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

После создания stream и durable pull consumer можно создать таблицу с движком NATS. Для этого необходимо инициализировать: nats_stream, nats_consumer_name и 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';

Таблицы JetStream обеспечивают доставку как минимум один раз: сообщение подтверждается только после его вставки в зависимые materialized view, поэтому сообщение, вставка которого завершилась ошибкой или была прервана, остаётся неподтверждённым и доставляется повторно. Core NATS (без JetStream) не поддерживает подтверждения или повторную доставку, поэтому гарантирует доставку не более одного раза, а сообщение, обработка которого была прервана, теряется.

​

Долговечность данных

Этот раздел относится только к JetStream. В Core NATS подтверждений нет, а доставка выполняется по принципу «не более одного раза», как описано выше, поэтому там нет окна, в котором подтвержденное сообщение может быть потеряно.

Таблица JetStream может незаметно потерять уже обработанные строки, если кэш страниц ОС будет сброшен до того, как вставленные данные окажутся записаны на диск. После передачи батча в зависимые materialized view consumer подтверждает эти сообщения, что позволяет stream продвинуться дальше. Однако вставленные строки становятся долговечными только после fsync целевой части, а по умолчанию он не выполняется синхронно (fsync_after_insert = 0). Если кэш страниц теряется после подтверждения, но до fsync целевой части, сообщения уже не будут доставлены повторно, поэтому строки пропадают без каких-либо ошибок, а count() просто оказывается меньше. Обычное завершение процесса эту проблему не выявляет, так как ядро сохраняет кэш страниц и в конечном итоге записывает его на диск. А вот потеря кэша страниц ее проявляет: например, при отключении питания на уровне устройства или некорректном сбросе хоста либо ядра.

Для рекомендуемого способа потребления через materialized view (подтверждение отправляется только после завершения всего конвейера вставки) установка fsync_after_insert = 1 (и fsync_part_directory = 1) для целевых таблиц MergeTree обеспечивает долговечность вставленных частей до отправки подтверждения, что существенно сужает это окно. Настройку необходимо включить для каждой таблицы MergeTree, в которую вставляется батч, включая цели каскадных materialized view; любая такая таблица, оставленная со значением по умолчанию, по-прежнему может потерять свою часть. Асинхронные промежуточные звенья одной лишь этой настройкой долговечности не получают: например, цель Distributed выполняет вставку в фоновом режиме при distributed_foreground_insert = 0, что является значением по умолчанию за пределами ClickHouse Cloud, поэтому ей нужны собственные настройки долговечности или синхронная вставка. Эта мера также не применима к прямому INSERT ... SELECT ... FROM <nats_table> с nats_commit_on_select = 1, где сообщения подтверждаются по достижении конца чтения, а не после того, как пункт назначения записал долговечную часть.