> ## 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.
> Этот движок позволяет интегрировать ClickHouse с NATS для публикации сообщений в subject, подписки на них и обработки новых сообщений по мере их поступления.
# Движок таблицы NATS
Этот движок позволяет интегрировать ClickHouse с [NATS](https://nats.io/).
`NATS` позволяет:
* Публиковать сообщения в subject и подписываться на них.
* Обрабатывать новые сообщения по мере их поступления.
## Создание таблицы
```sql theme={null}
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']
```
Обязательные параметры:
* `nats_url` – host:port (например, `localhost:4222`)..
* `nats_subjects` – Список `subject` для таблицы NATS, на которые нужно подписываться или в которые нужно публиковать сообщения. Поддерживаются `subject` с подстановочными символами, например `foo.*.bar` или `baz.>`
* `nats_format` – Формат сообщений. Используется та же нотация, что и в SQL-функции `FORMAT`, например `JSONEachRow`. Подробнее см. в разделе [Форматы](/ru/reference/formats/index).
Необязательные параметры:
* `nats_schema` – Параметр, который необходимо использовать, если формат требует определения схемы. Например, [Cap'n Proto](https://capnproto.org/) требует путь к файлу схемы и имя корневого объекта `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](/ru/reference/settings/session-settings#max_insert_block_size).
* `nats_flush_interval_ms` - Тайм-аут для сброса данных, прочитанных из NATS. По умолчанию: [stream\_flush\_interval\_ms](/ru/reference/settings/session-settings#stream_flush_interval_ms).
* `nats_username` - Имя пользователя NATS.
* `nats_password` - Пароль NATS.
* `nats_token` - Токен аутентификации NATS.
* `nats_credential_file` - Путь к файлу учётных данных NATS.
* `nats_startup_connect_tries` - Количество попыток подключения при запуске. По умолчанию: `5`.
* `nats_max_rows_per_message` — Максимальное количество строк, записываемых в одно сообщение NATS для построчных форматов. (по умолчанию: `1`).
* `nats_handle_error_mode` — Как обрабатывать ошибки движка NATS. Возможные значения: default (если не удаётся разобрать сообщение, будет сгенерировано исключение), stream (сообщение об исключении и исходное сообщение будут сохранены в виртуальных столбцах `_error` и `_raw_message`).
SSL-подключение:
Для безопасного подключения используйте `nats_secure = 1`.
Проверка сертификата управляется переменной окружения `CLICKHOUSE_NATS_TLS_SECURE`;
Если срок действия сертификата истёк, сертификат самоподписанный, отсутствует или иным образом недействителен, отключите проверку, установив `CLICKHOUSE_NATS_TLS_SECURE=0`.
Запись в таблицу NATS:
Если таблица читает только из одного subject, любая вставка будет публиковаться в тот же subject.
Однако если таблица читает из нескольких subject, нужно указать, в какой subject следует публиковать данные.
Поэтому при вставке в таблицу с несколькими subject необходимо задать `stream_like_engine_insert_queue`.
Вы можете выбрать один из subject, из которых читает таблица, и опубликовать туда свои данные. Например:
```sql theme={null}
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.
Пример:
```sql theme={null}
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:
```xml theme={null}
click
house
clickhouse
```
## Описание
`SELECT` не особенно полезен для чтения сообщений (кроме отладки), поскольку каждое сообщение можно прочитать только один раз. Гораздо практичнее создавать потоки в реальном времени с помощью [materialized views](/ru/reference/statements/create/view). Для этого:
1. Используйте движок, чтобы создать consumer NATS, и рассматривайте его как поток данных.
2. Создайте таблицу с нужной структурой.
3. Создайте materialized view, которое преобразует данные из движка и помещает их в ранее созданную таблицу.
Когда `MATERIALIZED VIEW` подключается к движку, оно начинает собирать данные в фоновом режиме. Это позволяет непрерывно получать сообщения из NATS и преобразовывать их в требуемый формат с помощью `SELECT`.
Одна таблица NATS может иметь сколько угодно materialized views; они не читают данные из таблицы напрямую, а получают новые записи (блоками), поэтому вы можете записывать данные в несколько таблиц с разным уровнем детализации (с группировкой — агрегацией — и без неё).
Пример:
```sql theme={null}
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:
```sql theme={null}
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 поддерживает все [форматы](/ru/reference/formats/index), доступные в ClickHouse.
Количество строк в одном сообщении NATS зависит от того, является ли формат построчным или блочным:
* Для построчных форматов количеством строк в одном сообщении NATS можно управлять с помощью настройки `nats_max_rows_per_message`.
* Для блочных форматов блок нельзя разделить на более мелкие части, однако количество строк в одном блоке можно регулировать с помощью общей настройки [max\_block\_size](/ru/reference/settings/session-settings#max_block_size).
## Использование JetStream
Перед использованием движка NATS с NATS JetStream необходимо создать stream в NATS и durable pull consumer. Для этого можно использовать, например, утилиту `nats` из пакета [NATS CLI](https://github.com/nats-io/natscli):
```bash theme={null}
$ 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
```
```bash theme={null}
$ 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:
```SQL theme={null}
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';
```