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 목록입니다. foo.*.bar 또는 baz.> 같은 와일드카드 subject를 지원합니다.
• nats_format – 메시지 포맷입니다. JSONEachRow와 같이 SQL FORMAT 함수와 동일한 표기법을 사용합니다. 자세한 내용은 포맷 섹션을 참조하십시오.

• nats_schema – 포맷에 스키마 정의가 필요한 경우 반드시 사용해야 하는 매개변수입니다. 예를 들어 Cap’n Proto는 스키마 파일의 경로와 루트 schema.capnp:Message 객체 이름이 필요합니다.
• nats_stream – NATS JetStream에 있는 기존 스트림의 이름입니다.
• nats_consumer_name – NATS JetStream에 있는 기존 durable pull consumer의 이름입니다.
• nats_num_consumers – 테이블당 consumer 수입니다. 기본값: 1. NATS core에서만, 단일 consumer의 처리량이 부족한 경우 consumer를 더 지정하십시오.
• nats_queue_group – NATS subscriber의 큐 그룹 이름입니다. 기본값은 테이블 이름입니다.
• nats_max_reconnect – 더 이상 권장되지 않으며 아무런 효과가 없습니다. 재연결은 nats_reconnect_wait 타임아웃에 따라 계속 수행됩니다.
• nats_reconnect_wait – 각 재연결 시도 사이에 대기할 시간(밀리초)입니다. 기본값: 2000.
• nats_server_list - 연결을 위한 서버 목록입니다. NATS 클러스터에 연결할 때 지정할 수 있습니다.
• nats_skip_broken_messages - 블록당 스키마와 호환되지 않는 메시지에 대한 NATS 메시지 파서 허용치입니다. 기본값: 0. nats_skip_broken_messages = N이면 엔진은 파싱할 수 없는 NATS 메시지 N개를 건너뜁니다(메시지 하나는 데이터 한 행과 같습니다).
• nats_max_block_size - NATS에서 데이터를 플러시하기 위해 폴링으로 수집하는 행 수입니다. 기본값: max_insert_block_size.
• nats_flush_interval_ms - NATS에서 읽은 데이터를 플러시하는 타임아웃입니다. 기본값: stream_flush_interval_ms.
• nats_wait_for_flush_interval - true이면 consumer 큐가 비워지는 즉시 종료되는 대신 백그라운드 스트리밍 사이클이 전체 플러시 인터벌(nats_flush_interval_ms, 또는 그 외에는 stream_flush_interval_ms) 동안 열려 있어, 최대 한 플러시 인터벌의 추가 수집 지연 시간을 감수하고 더 많은 메시지를 하나의 블록에 누적할 수 있습니다. 기본값: false(낮은 지연 시간의 drain-and-go 동작).
• 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 자격 증명 내용(사용자 JWT 및 seed가 포함된 .creds 파일과 동일한 페이로드)입니다. 쿼리에서 사용할 수 있는 유일한 표기이므로, <nats_credential_file overridable="false">로 운영자가 해당 경로를 잠그지 않은 한 명명된 컬렉션에서 상속된 nats_credential_file을 대체하며 충돌하지 않습니다. 명명된 컬렉션이 보유한 자격 증명을 제거하기 위해 빈 문자열을 할당할 수는 없습니다.
• nats_ca_file - NATS 서버 인증서를 검증하는 데 사용하는 신뢰할 수 있는 CA 인증서가 포함된 파일의 경로입니다. 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의 private key 경로입니다. 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에 저장됨).

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

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>
    <user>click</user>
    <password>house</password>
    <token>clickhouse</token>
</nats>

​

설명

1. 엔진을 사용해 NATS consumer를 생성하고 이를 데이터 스트림으로 간주합니다.
2. 원하는 구조로 테이블을 생성합니다.
3. 엔진의 데이터를 변환해 앞서 생성한 테이블에 저장하는 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;

DETACH TABLE consumer;
ATTACH TABLE consumer;

​

가상 컬럼

• _subject - NATS 메시지 subject입니다. 데이터 타입: String.

• _raw_message - 성공적으로 파싱되지 않은 원시 메시지입니다. 데이터 타입: Nullable(String).
• _error - 파싱 실패 중 발생한 예외 메시지입니다. 데이터 타입: Nullable(String).

​

데이터 포맷 지원

• 행 기반 포맷에서는 하나의 NATS 메시지에 포함되는 행 수를 nats_max_rows_per_message 설정으로 제어할 수 있습니다.
• 블록 기반 포맷에서는 블록을 더 작은 부분으로 나눌 수는 없지만, 하나의 블록에 포함되는 행 수는 일반 설정인 max_block_size로 제어할 수 있습니다.

​

JetStream 사용

$ 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

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

​

데이터 내구성