Engine de tabela NATS

Este engine permite integrar o ClickHouse ao NATS para publicar em ou assinar subjects de mensagens e processar novas mensagens à medida que se tornem disponíveis.

Este engine permite integrar o ClickHouse ao NATS.

NATS permite:

• Publicar em ou assinar subjects de mensagens.
• Processar novas mensagens à medida que se tornem disponíveis.

​

Criando uma tabela

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 obrigatórios:

• nats_url – host:port (por exemplo, localhost:4222).
• nats_subjects – Lista de subjects da tabela NATS para assinar/publicar. Suporta subjects curinga como foo.*.bar ou baz.>
• nats_format – Formato da mensagem. Usa a mesma notação da função SQL FORMAT, como JSONEachRow. Para mais informações, consulte a seção Formatos.

Parâmetros opcionais:

• nats_schema – Parâmetro que deve ser usado se o formato exigir uma definição de schema. Por exemplo, Cap’n Proto exige o caminho para o arquivo de schema e o nome do objeto raiz schema.capnp:Message.
• nats_stream – O nome de um stream existente no NATS JetStream.
• nats_consumer_name – O nome de um consumer pull durável existente no NATS JetStream.
• nats_num_consumers – O número de consumers por tabela. Padrão: 1. Especifique mais consumers se a taxa de transferência de um consumer for insuficiente, somente para NATS Core.
• nats_queue_group – Nome do queue group dos assinantes NATS. O padrão é o nome da tabela.
• nats_max_reconnect – Obsoleto e sem efeito; a reconexão é realizada permanentemente com o timeout nats_reconnect_wait.
• nats_reconnect_wait – Tempo de espera, em milissegundos, entre cada tentativa de reconexão. Padrão: 2000.
• nats_server_list - Lista de servidores para conexão. Pode ser especificada para conectar a um cluster NATS.
• nats_skip_broken_messages - Tolerância do parser de mensagens NATS a mensagens incompatíveis com o schema por bloco. Padrão: 0. Se nats_skip_broken_messages = N, o engine ignora N mensagens NATS que não podem ser convertidas (uma mensagem equivale a uma linha de dados).
• nats_max_block_size - Número de linhas coletadas por poll(s) para descarregar dados do NATS. Padrão: max_insert_block_size.
• nats_flush_interval_ms - Timeout para descarregar os dados lidos do NATS. Padrão: stream_flush_interval_ms.
• nats_wait_for_flush_interval - Se true, um ciclo de streaming em segundo plano permanece aberto durante todo o intervalo de descarregamento (nats_flush_interval_ms ou stream_flush_interval_ms, caso contrário), em vez de terminar assim que a fila do consumer é esvaziada, permitindo que mais mensagens se acumulem em um único bloco, ao custo de até um intervalo de descarregamento adicional na latência de ingestão. Padrão: false (comportamento de esvaziar e seguir com baixa latência).
• nats_username - Nome de usuário do NATS. Quando é armazenado em uma coleção nomeada definida no arquivo de configuração do servidor, a consulta não pode sobrescrever nats_url nem nats_server_list da coleção.
• nats_password - Senha do NATS. Quando é armazenada em uma coleção nomeada definida no arquivo de configuração do servidor, a consulta não pode sobrescrever nats_url nem nats_server_list da coleção.
• nats_token - Token de autenticação do NATS. Quando é armazenado em uma coleção nomeada definida no arquivo de configuração do servidor, a consulta não pode sobrescrever nats_url nem nats_server_list da coleção.
• nats_credential_file - Caminho para um arquivo de credentials do NATS. Ele é aceito apenas de uma coleção nomeada definida no arquivo de configuração do servidor cujo nats_url e nats_server_list não são sobrescritos pela consulta, pois o servidor abre o caminho com seus próprios privilégios. Em uma consulta, informe o conteúdo do arquivo em nats_credentials.
• nats_credentials - Conteúdo das credentials do NATS (a mesma carga útil de um arquivo .creds com JWT de usuário e seed). Como é a única forma que uma consulta pode usar, substitui um nats_credential_file herdado de uma coleção nomeada em vez de entrar em conflito com ele, a menos que o operador tenha bloqueado esse caminho com <nats_credential_file overridable="false">. Não pode receber uma string vazia para remover as credentials fornecidas por uma coleção nomeada.
• nats_ca_file - Caminho para um arquivo com os certificados de CA confiáveis usados para verificar o certificado do servidor NATS. Requer nats_secure. Assim como nats_credential_file, é aceito apenas de uma coleção nomeada definida no arquivo de configuração do servidor cujo nats_url e nats_server_list não são sobrescritos pela consulta, pois o servidor abre o caminho com seus próprios privilégios.
• nats_client_cert_file - Caminho para o certificado do cliente apresentado ao servidor NATS. Requer nats_secure e nats_client_key_file. Aceito das mesmas fontes que nats_ca_file.
• nats_client_key_file - Caminho para a chave privada de nats_client_cert_file. Aceito das mesmas fontes que nats_ca_file.
• nats_startup_connect_tries - Número de tentativas de conexão na inicialização. Padrão: 5.
• nats_max_rows_per_message — O número máximo de linhas gravadas em uma mensagem NATS para formatos baseados em linha. (padrão: 1).
• nats_commit_on_select - Faz commit das mensagens quando uma consulta é realizada. Aplica-se somente ao JetStream; o NATS Core não tem confirmações. Padrão: 0.
• nats_handle_error_mode — Como lidar com erros no engine NATS. Valores possíveis: default (a exceção será lançada se não conseguirmos converter uma mensagem), stream (a mensagem de exceção e a mensagem bruta serão salvas nas colunas virtuais _error e _raw_message).

Conexão SSL:

Para uma conexão segura, use nats_secure = 1. A verificação de certificados é controlada pela variável de ambiente CLICKHOUSE_NATS_TLS_SECURE; se o certificado estiver expirado, for autossinado, estiver ausente ou for inválido de outra forma, desative a verificação definindo CLICKHOUSE_NATS_TLS_SECURE=0.

Um certificado de servidor assinado por uma CA privada é verificado apontando nats_ca_file para o certificado da CA, o que é preferível a desativar a verificação. Quando o servidor exige certificados de cliente, forneça-os com nats_client_cert_file e nats_client_key_file. Todos os três são configurações do operador: eles vêm de uma coleção nomeada definida no arquivo de configuração do servidor. Cada arquivo é lido quando a tabela se conecta; portanto, um arquivo ilegível ou malformado faz a consulta falhar em vez de falhar durante o handshake.

Gravação na tabela NATS:

Se a tabela lê de apenas um subject, qualquer insert publicará nesse mesmo subject. No entanto, se a tabela lê de múltiplos subjects, é preciso especificar em qual subject queremos publicar. Por isso, ao inserir em uma tabela com múltiplos subjects, é necessário definir stream_like_engine_insert_queue. Você pode selecionar um dos subjects dos quais a tabela lê e publicar seus dados nele. Por exemplo:

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

Também é possível adicionar configurações de formato junto com as configurações relacionadas ao NATS.

Exemplo:

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

A configuração do servidor NATS pode ser adicionada usando o arquivo de configuração do ClickHouse. Mais especificamente, você pode adicionar sua senha para o engine NATS:

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

​

Descrição

SELECT não é particularmente útil para ler mensagens (exceto para depuração), porque cada mensagem pode ser lida apenas uma vez. É mais prático criar fluxos em tempo real usando visões materializadas. Para fazer isso:

1. Use o engine para criar um consumer do NATS e tratá-lo como um fluxo de dados.
2. Crie uma tabela com a estrutura desejada.
3. Crie uma visão materializada que converta os dados do engine e os insira em uma tabela criada anteriormente.

Quando a MATERIALIZED VIEW é vinculada ao engine, ela começa a coletar dados em segundo plano. Isso permite que você receba continuamente mensagens do NATS e as converta para o formato necessário usando SELECT. Uma tabela do NATS pode ter quantas visões materializadas você quiser; elas não leem dados diretamente da tabela, mas recebem novos registros (em blocos). Assim, você pode gravar em várias tabelas com diferentes níveis de detalhamento (com agrupamento - agregação e sem).

Exemplo:

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 parar de receber dados dos streams ou alterar a lógica de conversão, desanexe a visão materializada:

DETACH TABLE consumer;
ATTACH TABLE consumer;

Se quiser alterar a tabela de destino com ALTER, recomendamos desabilitar a visão materializada para evitar discrepâncias entre a tabela de destino e os dados da view.

​

Colunas virtuais

• _subject - subject da mensagem no NATS. Tipo de dado: String.

Colunas virtuais adicionais quando nats_handle_error_mode='stream':

• _raw_message - Mensagem bruta que não pôde ser processada com sucesso. Tipo de dado: Nullable(String).
• _error - Mensagem de exceção ocorrida durante uma falha no processamento. Tipo de dado: Nullable(String).

Observação: as colunas virtuais _raw_message e _error são preenchidas apenas em caso de exceção durante o processamento; elas são sempre NULL quando a mensagem é processada com sucesso.

​

Suporte a formatos de dados

O engine NATS oferece suporte a todos os formatos compatíveis com o ClickHouse. O número de linhas em uma mensagem NATS depende de o formato ser baseado em linhas ou em blocos:

• Para formatos baseados em linhas, o número de linhas em uma mensagem NATS pode ser controlado pela configuração nats_max_rows_per_message.
• Para formatos baseados em blocos, não é possível dividir um bloco em partes menores, mas o número de linhas em um bloco pode ser controlado pela configuração geral max_block_size.

​

Usando o JetStream

Antes de usar o engine NATS com o NATS JetStream, você deve criar um stream NATS e um consumer pull durável. Para isso, você pode usar, por exemplo, o utilitário nats do pacote NATS CLI:

criando um 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

criando um consumer pull durável

$ 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

Após criar o stream e o consumer pull durável, podemos criar uma tabela com o engine NATS. Para isso, você precisa definir: nats_stream, nats_consumer_name e 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';

As tabelas JetStream oferecem entrega de pelo menos uma vez: uma mensagem só é confirmada após ser inserida nas visões materializadas dependentes; portanto, uma mensagem cuja inserção falha ou é interrompida permanece sem confirmação e é entregue novamente. O NATS Core (sem JetStream) não oferece confirmação nem reprocessamento; portanto, tem entrega de no máximo uma vez, e uma mensagem interrompida é perdida.

​

Durabilidade dos dados

Esta seção se aplica somente ao JetStream. O NATS Core não tem confirmações e opera no modo de entrega no máximo uma vez, conforme descrito acima. Portanto, não há uma janela em que uma mensagem confirmada possa ser perdida.

Uma tabela JetStream pode perder silenciosamente linhas já consumidas se o cache de páginas do SO for descartado antes que os dados inseridos sejam gravados em disco. Depois que um lote é enviado às visões materializadas dependentes, o consumidor confirma essas mensagens, o que permite que o stream avance além delas. No entanto, as linhas inseridas só se tornam duráveis quando a parte de destino recebe fsync, o que, por padrão, não ocorre de forma síncrona (fsync_after_insert = 0). Se o cache de páginas for perdido após a confirmação, mas antes de a parte de destino receber fsync, as mensagens não serão mais reenviadas; portanto, as linhas serão perdidas sem erro e count() será simplesmente menor. Encerrar um processo normalmente não expõe esse problema, pois o kernel mantém o cache de páginas e acaba gravando-o em disco. Já a perda do cache de páginas o expõe; exemplos incluem perda de energia no nível do dispositivo e uma reinicialização não limpa do host ou do kernel.

No caminho recomendado de consumo por visão materializada (a confirmação só é enviada após a conclusão de todo o pipeline de inserção), definir fsync_after_insert = 1 (e fsync_part_directory = 1) nas tabelas MergeTree de destino torna as partes inseridas duráveis antes do envio da confirmação, reduzindo substancialmente essa janela. A configuração deve ser ativada em todas as tabelas MergeTree nas quais o lote é inserido, incluindo os destinos de visões materializadas em cascata; qualquer tabela desse tipo mantida com a configuração padrão ainda pode perder sua parte. Intermediários assíncronos não ganham durabilidade apenas com essa configuração: por exemplo, um destino Distributed insere em segundo plano quando distributed_foreground_insert = 0, que é o padrão fora do ClickHouse Cloud, portanto, ele precisa de suas próprias configurações de durabilidade ou de inserção síncrona. Essa mitigação também não se aplica a um INSERT ... SELECT ... FROM <nats_table> direto com nats_commit_on_select = 1, em que as mensagens são confirmadas quando a leitura chega ao fim, e não após o destino gravar uma parte durável.