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 – NATS 表要订阅/发布的 subject 列表。支持通配符 subject，例如 foo.*.bar 或 baz.>
• nats_format – 消息格式。使用与 SQL FORMAT 函数相同的表示法，例如 JSONEachRow。更多信息，请参见格式部分。

可选参数：

• nats_schema – 如果格式需要 schema 定义，则必须使用此参数。例如，Cap’n Proto 需要提供 schema 文件的 path 以及根对象 schema.capnp:Message 的名称。
• nats_stream – NATS JetStream 中现有 stream 的名称。
• nats_consumer_name – NATS JetStream 中现有持久化拉取消费者的名称。
• nats_num_consumers – 每个表的消费者数量。默认值：1。仅适用于 NATS core：如果单个消费者的吞吐量不足，可指定更多消费者。
• nats_queue_group – NATS 订阅者的 queue group 名称。默认值为表名。
• nats_max_reconnect – 已弃用且不起作用；系统会按 nats_reconnect_wait timeout 永久执行重连。
• nats_reconnect_wait – 每次重连尝试之间的休眠时间 (毫秒) 。默认值：2000。
• nats_server_list - 用于 connection 的 server 列表。可用于连接到 NATS cluster。
• nats_skip_broken_messages - NATS 消息解析器对每个块中与 schema 不兼容消息的容忍数量。默认值：0。如果 nats_skip_broken_messages = N，则该引擎会跳过 N 条无法解析的 NATS 消息 (1 条消息等于 1 行数据) 。
• nats_max_block_size - 为从 NATS flush 数据而通过 poll 收集的行数。默认值：max_insert_block_size。
• nats_flush_interval_ms - flush 从 NATS 读取的数据的 timeout。默认值：stream_flush_interval_ms。
• nats_wait_for_flush_interval - 如果为 true，后台流式周期将在整个 flush 时间间隔内保持开启 (nats_flush_interval_ms，否则为 stream_flush_interval_ms) ，而不是在消费者 queue 耗尽后立即结束；这样可使更多消息累积到单个块中，但会额外增加最多一个 flush 时间间隔的摄取延迟。默认值：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 凭据文件的 path。仅接受来自服务器配置文件中定义的命名集合的值，且该集合的 nats_url 和 nats_server_list 未被查询覆盖，因为服务器会使用自身权限打开该 path。在查询中，请改为通过 nats_credentials 传入文件内容。
• nats_credentials - NATS 凭据内容 (与包含用户 JWT 和 seed 的 .creds 文件中的负载相同) 。由于这是查询唯一可用的写法，它会替换从命名集合继承的 nats_credential_file，而不会与之冲突；除非操作员通过 <nats_credential_file overridable="false"> 锁定了该 path。不能将其赋值为空字符串来移除命名集合携带的凭据。
• nats_ca_file - 包含受信任 CA 证书的文件 path，用于验证 NATS 服务器证书。需要 nats_secure。与 nats_credential_file 一样，仅接受来自服务器配置文件中定义的命名集合的值，且该集合的 nats_url 和 nats_server_list 未被查询覆盖，因为服务器会使用自身权限打开该 path。
• nats_client_cert_file - 提供给 NATS 服务器的客户端证书的 path。需要 nats_secure 和 nats_client_key_file。接受与 nats_ca_file 相同来源的值。
• nats_client_key_file - nats_client_cert_file 私钥的 path。接受与 nats_ca_file 相同来源的值。
• nats_startup_connect_tries - 启动时的连接尝试次数。默认值：5。
• nats_max_rows_per_message — 对于按行组织的格式，一条 NATS 消息中写入的最大行数。默认值：1。
• nats_commit_on_select - 发出查询时提交消息。仅适用于 JetStream；NATS core 不提供确认机制。默认值：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 提供相应文件。这三个都是运维设置： 它们来自服务器配置文件中定义的命名集合。表建立连接时会读取每个文件，因此 文件无法读取或格式错误会导致查询失败，而不是握手失败。

写入 NATS 表：

如果表只从一个 subject 读取，则任何 insert 都会发布到该 subject。 但如果表从多个 subject 读取，就需要指定要发布到哪个 subject。 因此，每当向具有多个 subject 的表中 insert 数据时，都需要设置 stream_like_engine_insert_queue。 你可以从该表读取的 subject 中选择一个，并将数据发布到该 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';

可通过 ClickHouse 配置文件添加 NATS 服务器配置。 更具体地说，可以添加 NATS 引擎的密码：

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

​

说明

SELECT 并不特别适合用于读取消息 (调试除外) ，因为每条消息只能读取一次。更实用的做法是使用 materialized views 创建实时处理链路。为此：

1. 使用该引擎创建一个 NATS 消费者，并将其视为数据 stream。
2. 创建一个具有所需结构的表。
3. 创建一个 materialized view，将该引擎中的数据转换后写入前面创建的表中。

当 MATERIALIZED VIEW 连接到该引擎后，就会开始在后台收集数据。这样一来，你就可以持续接收来自 NATS 的消息，并使用 SELECT 将其转换为所需格式。 一个 NATS 表可以拥有任意数量的 materialized view；它们不会直接从该表读取数据，而是接收新的记录 (以块的形式) ，因此你可以写入多个明细粒度不同的表 (带分组聚合和不带分组聚合) 。

示例：

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;

若要停止接收 stream 数据或更改转换逻辑，请分离 materialized view：

DETACH TABLE consumer;
ATTACH TABLE consumer;

如果你想使用 ALTER 修改目标表，我们建议先禁用物化视图，以避免目标表与视图数据之间出现不一致。

​

虚拟列

• _subject - NATS 消息的 subject。数据类型：String。

当 nats_handle_error_mode='stream' 时，会提供以下额外的虚拟列：

• _raw_message - 无法成功解析的原始消息。数据类型：Nullable(String)。
• _error - 解析失败时产生的异常消息。数据类型：Nullable(String)。

注意：_raw_message 和 _error 这两个虚拟列仅会在解析过程中发生异常时填充；如果消息解析成功，它们始终为 NULL。

​

数据格式支持

NATS 引擎 支持 ClickHouse 支持的所有格式。 单条 NATS 消息中的行数取决于格式是按行还是按块：

• 对于按行的格式，可通过设置 nats_max_rows_per_message 来控制单条 NATS 消息中的行数。
• 对于按块的格式，无法将块拆分成更小的部分，但一个块中的行数可通过通用设置 max_block_size 控制。

​

使用 JetStream

在结合 NATS JetStream 使用 NATS 引擎之前，您必须先创建一个 NATS stream 和一个持久化拉取消费者。为此，您可以使用 NATS CLI 包中的 nats 工具，例如：

创建 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

创建持久化拉取消费者

$ 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 和持久化拉取消费者后，我们就可以创建一个使用 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 views 后才会被确认，因此插入失败或中断的消息会保持未确认状态，并被重新投递。Core NATS (不含 JetStream) 没有确认或重放机制，因此仅提供至多一次语义，中断的消息会丢失。

​

数据持久性

本节仅适用于 JetStream。Core NATS 没有确认机制，且如上所述采用至多一次语义，因此不存在已确认消息可能丢失的时间窗口。

如果在插入的数据写入磁盘前 OS page cache 被丢弃，JetStream 表可能会在无提示的情况下丢失已消费的行。批次被推送到依赖的 materialized view 后，消费者会确认这些消息，从而使 stream 越过这些消息继续推进。但是，插入的行只有在目标 parts 被 fsync 后才真正持久化；默认情况下不会同步执行此操作 (fsync_after_insert = 0) 。如果在确认之后、目标 parts 被 fsync 之前 page cache 丢失，消息不会再被重新投递，因此这些行会在没有任何错误的情况下丢失，count() 结果只会变小。普通的进程 kill 不会暴露这个问题，因为 kernel 会保留 page cache，并最终将其写回。page cache 丢失则会暴露该问题，例如设备级断电，以及主机或 kernel 的非正常重置。

对于推荐的 materialized-view 消费路径 (仅在整个插入管道完成后才发送确认) ，在目标 MergeTree 表上设置 fsync_after_insert = 1 (以及 fsync_part_directory = 1) ，可确保插入的 parts 在发送确认前持久化，从而大幅缩小这一时间窗口。必须在批次插入的每个 MergeTree 表上启用该设置，包括级联 materialized-view 的目标表；任何仍使用默认设置的此类表仍可能丢失其 parts。异步中间层不会仅凭此设置获得持久性：例如，当 distributed_foreground_insert = 0 时，Distributed 目标端会在后台插入；这是 ClickHouse Cloud 之外的默认设置，因此它需要自身的持久性设置或同步插入。此缓解措施也不适用于带有 nats_commit_on_select = 1 的直接 INSERT ... SELECT ... FROM <nats_table>；在这种情况下，消息会在读取结束时被确认，而不是在目标端写入持久化 parts 后。