ClickHouse Kafka Connect Sink

​

Licença

​

Requisitos do ambiente

​

Matriz de compatibilidade de versões

​

Principais recursos

• Vem com semântica exactly-once pronta para uso. É baseado em um novo recurso central do ClickHouse chamado KeeperMap (usado como armazenamento de estado pelo conector) e permite uma arquitetura minimalista.
• Suporte a armazenamentos de estado de terceiros: atualmente, o padrão é em memória, mas pode usar o KeeperMap (Redis será adicionado em breve).
• Integração principal: desenvolvida, mantida e suportada pela ClickHouse.
• Testado continuamente no ClickHouse Cloud.
• Inserções de dados com schema declarado e sem schema.
• Suporte a todos os tipos de dados do ClickHouse.

​

Instruções de instalação

​

Obtenha os detalhes da conexão

​

Instruções gerais de instalação

• Baixe um arquivo ZIP contendo o arquivo JAR do conector na página de Releases do repositório ClickHouse Kafka Connect Sink.
• Extraia o conteúdo do arquivo ZIP e copie-o para o local desejado.
• Adicione à configuração plugin.path, no arquivo de propriedades do Connect, o caminho para o diretório do plugin, para que o Confluent Platform possa encontrá-lo.
• Forneça um nome de tópico, o hostname da instância do ClickHouse e a senha na configuração.

connector.class=com.clickhouse.kafka.connect.ClickHouseSinkConnector
tasks.max=1
topics=<topic_name>
ssl=true
jdbcConnectionProperties=?sslmode=STRICT
security.protocol=SSL
hostname=<hostname>
database=<database_name>
password=<password>
ssl.truststore.location=/tmp/kafka.client.truststore.jks
port=8443
value.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
exactlyOnce=true
username=default
schemas.enable=false

• Reinicie a Confluent Platform.
• Se você usa a Confluent Platform, faça login na UI do Confluent Control Center para verificar se o ClickHouse Sink está disponível na lista de conectores.

​

Opções de configuração

• detalhes da conexão: hostname (obrigatório) e porta (opcional)
• credenciais do usuário: senha (obrigatória) e nome de usuário (opcional)
• classe do conector: com.clickhouse.kafka.connect.ClickHouseSinkConnector (obrigatória)
• topics ou topics.regex: os tópicos do Kafka a serem consumidos — os nomes dos tópicos devem corresponder aos nomes das tabelas (obrigatório)
• conversores de chave e valor: defina-os com base no tipo de dados do seu tópico. Obrigatórios se ainda não estiverem definidos na configuração do worker.

​

Tabelas de destino

​

Pré-processamento

​

Tipos de dados suportados

• (1) - JSON é suportado apenas quando as configurações do ClickHouse incluem input_format_binary_read_json_as_string=1. Isso funciona apenas para a família de formatos RowBinary, e a configuração afeta todas as colunas na requisição de insert, portanto todas elas devem ser do tipo string. Nesse caso, o conector converterá STRUCT em uma string JSON.
• (2) - Quando struct tem unions como oneof, o converter deve ser configurado para NÃO adicionar prefixo/sufixo aos nomes de campo. Há a configuração generate.index.for.unions=false para o ProtobufConverter.

​

Receitas de configuração

​

Configuração básica

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "tasks.max": "1",
    "consumer.override.max.poll.records": "5000",
    "consumer.override.max.partition.fetch.bytes": "5242880",
    "database": "default",
    "errors.retry.timeout": "60000",
    "exactlyOnce": "false",
    "hostname": "localhost",
    "port": "8443",
    "ssl": "true",
    "jdbcConnectionProperties": "?ssl=true&sslmode=strict",
    "username": "default",
    "password": "<PASSWORD>",
    "topics": "<TOPIC_NAME>",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "clickhouseSettings": ""
  }
}

​

Configuração básica com vários tópicos

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "topics": "SAMPLE_TOPIC, ANOTHER_TOPIC, YET_ANOTHER_TOPIC",
    ...
  }
}

​

Configuração básica com DLQ

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "<DLQ_TOPIC>",
    "errors.deadletterqueue.context.headers.enable": "true",
  }
}

​

Suporte a schema do Avro

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "<SCHEMA_REGISTRY_HOST>:<PORT>",
    "value.converter.schemas.enable": "true",
  }
}

​

Mapeamento de tipos do Avro

​

Schemas Avro sem suporte

• tipo lógico decimal em fixed

{"name": "decimal_18_4", "type": "fixed", "size": 8, "logicalType": "decimal", "precision": 18, "scale": 4}

• unions Nullable

{"name": "mixed_union", "type": ["null", "string", "int"], "default": null}

• unions em registros

{
  "name": "record_union",
  "type": [
    {
      "type": "record",
      "name": "TypeA",
      "fields": [
        {
          "name": "label",
          "type": "string"
        }
      ]
    },
    {
      "type": "record",
      "name": "TypeB",
      "fields": [
        {
          "name": "count",
          "type": "int"
        }
      ]
    }
  ]
}

​

Suporte a schema do Protobuf

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "io.confluent.connect.protobuf.ProtobufConverter",
    "value.converter.schema.registry.url": "<SCHEMA_REGISTRY_HOST>:<PORT>",
    "value.converter.schemas.enable": "true",
  }
}

​

Mapeamento de tipos do Protobuf

​

Observação sobre a tradução de campos oneof para colunas do ClickHouse

syntax = "proto3";

package com.clickhouse.kafka.connect.proto.test;

message StringIntUnion {
  oneof mixed {
    string mixed_string = 2;
    int32 mixed_int = 3;
  }
}

CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
    mixed_string Nullable(String),
    mixed_int Nullable(Int32)
) ENGINE = ...;

​

Schemas Protobuf sem suporte

• uniões com várias mensagens (antes da versão 26.1 do CH e, antes da versão 26.9 do CH, a menos que um SETTING esteja habilitado, veja abaixo)

syntax = "proto3";

package com.clickhouse.kafka.connect.proto.test;

message TwoRecords {
  oneof payload {
    TypeA type_a = 2;
    TypeB type_b = 3;
  }

  // translates to Nullable(Tuple(label String)) in ClickHouse, which is unsupported before 26.1
  message TypeA {
    string label = 1;
  }

  // translates to Nullable(Tuple(count Int32)) in ClickHouse, which is unsupported before 26.1
  message TypeB {
    int32 count = 1;
  }
}

​

Suporte a schema JSON

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  }
}

​

Suporte a String

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "customInsertFormat": "true",
    "insertFormat": "CSV"
  }
}

​

Buffering interno

• bufferCount controla quantos registros são mantidos em buffer antes do flush.
• bufferFlushTime define um tempo máximo de espera (em milissegundos) antes de fazer o flush dos registros em buffer.
• bufferFlushTime só tem efeito quando bufferCount > 0.
• bufferCount=0 e bufferFlushTime=0 mantêm o buffering desabilitado (comportamento padrão).
• O buffering não é compatível quando exactlyOnce=true.

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "exactlyOnce": "false",
    "bufferCount": "5000",
    "bufferFlushTime": "2000"
  }
}

​

Logging

confluent local services connect log

​

Monitoramento

​

Métricas específicas do ClickHouse

com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}

​

Métricas de Produtor/Consumidor do Kafka

• records-sent-total: Número total de registros enviados para o tópico
• bytes-sent-total: Total de bytes enviados para o tópico
• record-send-rate: Taxa média de registros enviados por segundo
• byte-rate: Taxa média de bytes enviados por segundo
• compression-rate: Taxa de compressão obtida

• records-sent-total: Total de registros enviados para a partição
• bytes-sent-total: Total de bytes enviados para a partição
• records-lag: Lag atual na partição
• records-lead: Lead atual na partição
• replica-fetch-lag: Informações de lag das réplicas

• connection-creation-total: Total de conexões criadas com o nó do Kafka
• connection-close-total: Total de conexões encerradas
• request-total: Total de solicitações enviadas ao nó
• response-total: Total de respostas recebidas do nó
• request-rate: Taxa média de solicitações por segundo
• response-rate: Taxa média de respostas por segundo

• Taxa de transferência: Acompanhar as taxas de ingestão de dados
• Lag: Identificar gargalos e atrasos no processamento
• Compressão: Medir a eficiência da compressão de dados
• Saúde da conexão: Monitorar a conectividade e a estabilidade da rede

​

Métricas do Kafka Connect Framework

• task-count: Número total de tarefas no conector
• running-task-count: Número de tarefas em execução no momento
• paused-task-count: Número de tarefas pausadas no momento
• failed-task-count: Número de tarefas que falharam
• destroyed-task-count: Número de tarefas destruídas
• unassigned-task-count: Número de tarefas não atribuídas

• deadletterqueue-produce-failures: Número de gravações na DLQ que falharam
• deadletterqueue-produce-requests: Total de tentativas de gravação na DLQ
• last-error-timestamp: Timestamp do último erro
• records-skip-total: Número total de registros ignorados devido a erros
• records-retry-total: Número total de registros que passaram por nova tentativa
• errors-total: Número total de erros encontrados

• offset-commit-failures: Número de commits de offset que falharam
• offset-commit-avg-time-ms: Tempo médio dos commits de offset
• offset-commit-max-time-ms: Tempo máximo dos commits de offset
• put-batch-avg-time-ms: Tempo médio para processar um lote
• put-batch-max-time-ms: Tempo máximo para processar um lote
• source-record-poll-total: Total de registros coletados

​

Boas práticas de monitoramento

1. Monitore o lag do consumidor: Acompanhe records-lag por partição para identificar gargalos de processamento
2. Acompanhe as taxas de erro: Observe errors-total e records-skip-total para detectar problemas de qualidade dos dados
3. Observe a integridade das tarefas: Monitore as métricas de status das tarefas para garantir que estejam em execução corretamente
4. Meça a vazão: Use records-send-rate e byte-rate para acompanhar o desempenho da ingestão
5. Monitore a integridade da conexão: Verifique as métricas de conexão no nível do nó para identificar problemas de rede
6. Acompanhe a eficiência da compressão: Use compression-rate para otimizar a transferência de dados

​

Limitações

• Não há suporte a exclusões.
• O tamanho do batch é herdado das propriedades do consumer do Kafka.
• Ao usar o KeeperMap para exactly-once, se o offset for alterado ou recuado, será necessário excluir o conteúdo do KeeperMap para esse tópico específico. (Consulte o guia de solução de problemas abaixo para mais detalhes)

​

Ajuste de desempenho e otimização da vazão

​

Quando o ajuste de desempenho é necessário?

• Cargas de trabalho de alta vazão: ao processar milhões de eventos por segundo de tópicos do Kafka
• Consumer lag: quando seu conector não consegue acompanhar a taxa de produção de dados, causando um atraso cada vez maior
• Restrições de recursos: quando você precisa otimizar o uso de CPU, memória ou rede
• Múltiplos tópicos: ao consumir simultaneamente vários tópicos de alto volume
• Mensagens pequenas: ao lidar com muitas mensagens pequenas que se beneficiariam do agrupamento em lotes no lado do servidor

• Você está processando volumes baixos a moderados (< 10.000 mensagens/segundo)
• O consumer lag é estável e aceitável para o seu caso de uso
• As configurações padrão do conector já atendem aos seus requisitos de vazão
• Seu cluster ClickHouse consegue lidar facilmente com a carga de entrada

​

Entendendo o fluxo de dados

1. Kafka Connect Framework busca mensagens dos tópicos do Kafka em segundo plano
2. O conector faz polling de mensagens no buffer interno do framework
3. O conector agrupa as mensagens em lotes com base no tamanho do polling
4. O ClickHouse recebe a inserção em lote via HTTP/S
5. O ClickHouse processa a inserção (de forma síncrona ou assíncrona)

​

Ajuste do tamanho do lote no Kafka Connect

• fetch.min.bytes: Quantidade mínima de dados antes de o framework repassar os dados ao conector (padrão: 1 byte)
• fetch.max.bytes: Quantidade máxima de dados a buscar em uma única solicitação (padrão: 52428800 / 50 MB)
• fetch.max.wait.ms: Tempo máximo de espera antes de retornar os dados se fetch.min.bytes não for atingido (padrão: 500 ms)

• max.poll.records: Número máximo de registros retornados em uma única consulta de polling (padrão: 500)
• max.partition.fetch.bytes: Quantidade máxima de dados por partição (padrão: 1048576 / 1 MB)

# Aumentar o número de registros por poll
consumer.override.max.poll.records=5000

# Aumentar o tamanho de fetch da partição (5 MB)
consumer.override.max.partition.fetch.bytes=5242880

# Opcional: Aumentar o tamanho mínimo de fetch para aguardar mais dados (1 MB)
consumer.override.fetch.min.bytes=1048576

# Opcional: Reduzir o tempo de espera se a latência for crítica
consumer.override.fetch.max.wait.ms=300

• Lotes maiores = Melhor desempenho de ingestão no ClickHouse, menos partes, menor sobrecarga
• Lotes maiores = Maior uso de memória, com possível aumento da latência de ponta a ponta
• Lotes grandes demais = Risco de timeouts, erros de OutOfMemory ou de exceder max.poll.interval.ms

​

Inserções assíncronas

• Muitos lotes pequenos: Seu conector envia pequenos lotes com frequência (< 1000 linhas por lote)
• Alta concorrência: Várias tarefas do conector estão gravando na mesma tabela
• Implantação distribuída: Você executa muitas instâncias do conector em hosts diferentes
• Sobrecarga na criação de partes: Você está enfrentando erros de “too many partes”
• Carga de trabalho mista: Combinação de ingestão em tempo real com cargas de trabalho de consulta

• Você já estiver enviando lotes grandes (> 10.000 linhas por lote) com frequência controlada
• Você precisar de visibilidade imediata dos dados (as consultas precisam ver os dados instantaneamente)
• A semântica exactly-once com wait_for_async_insert=0 entrar em conflito com seus requisitos
• Seu caso de uso puder se beneficiar, em vez disso, de melhorias no batching no lado do cliente

1. Recebe a consulta INSERT do conector
2. Grava os dados em um buffer na memória (em vez de gravá-los imediatamente no disco)
3. Retorna sucesso ao conector (se wait_for_async_insert=0)
4. Grava o buffer no disco quando uma destas condições é atendida:
   • O buffer atinge async_insert_max_data_size (padrão: 100 MB)
   • async_insert_busy_timeout_ms milissegundos se passaram desde a primeira inserção (padrão: 1000 ms)
   • Número máximo de consultas acumuladas (async_insert_max_query_number, padrão: 100)

{
  "name": "clickhouse-connect",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    ...
    "clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
  }
}

• async_insert=1: Habilita inserções assíncronas
• wait_for_async_insert=1 (recomendado): O conector espera os dados serem gravados no armazenamento do ClickHouse antes de confirmar o recebimento. Isso oferece garantias de entrega.
• wait_for_async_insert=0: O conector confirma o recebimento imediatamente após armazenar os dados em buffer. Melhor desempenho, mas os dados podem ser perdidos se o servidor falhar antes da gravação.

"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=104857600,async_insert_busy_timeout_ms=1000"

• async_insert_max_data_size (padrão: 104857600 / 100 MB): Tamanho máximo do buffer antes da gravação
• async_insert_busy_timeout_ms (padrão: 1000): Tempo máximo (ms) antes da gravação
• async_insert_stale_timeout_ms (padrão: 0): Tempo (ms) desde o último insert antes da gravaação
• async_insert_max_query_number (padrão: 100): Número máximo de consultas antes da gravação

• Benefícios: Menos partes, melhor desempenho de merge, menor sobrecarga de CPU, maior throughput em cenários de alta concorrência
• Considerações: Os dados não ficam imediatamente disponíveis para consulta, latência de ponta a ponta um pouco maior
• Riscos: Perda de dados em caso de falha do servidor se wait_for_async_insert=0, possível pressão de memória com buffers grandes

{
  "config": {
    "exactlyOnce": "true",
    "clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
  }
}

​

Paralelismo do conector

"tasks.max": "4"

• O número máximo efetivo de tarefas = número de partições do tópico
• Cada tarefa mantém sua própria conexão com o ClickHouse
• Mais tarefas = maior sobrecarga e possível contenção de recursos

"ignorePartitionsWhenBatching": "true"

​

Múltiplos tópicos de alta vazão

​

Considerações sobre o engine de tabela do ClickHouse

• MergeTree: Melhor para a maioria dos casos de uso; equilibra o desempenho de consulta e inserção
• ReplicatedMergeTree: Necessário para alta disponibilidade; adiciona sobrecarga de replicação
• *MergeTree com ORDER BY adequado: Otimize para seus padrões de consulta

CREATE TABLE my_table (...)
ENGINE = MergeTree()
ORDER BY (timestamp, id)
SETTINGS 
    -- Aumentar o número máximo de insert threads para escrita paralela de partes
    max_insert_threads = 4,
    -- Permitir inserts com quorum para maior confiabilidade (ReplicatedMergeTree)
    insert_quorum = 2

"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"

​

Pool de conexões e timeouts

"jdbcConnectionProperties": "?socket_timeout=300000,connection_timeout=30000"

• socket_timeout (padrão: 30000 ms): Tempo máximo para operações de leitura
• connection_timeout (padrão: 10000 ms): Tempo máximo para estabelecer a conexão

​

Buffer de rede do cliente

"jdbcConnectionProperties": "?client_network_buffer_size=1000000"

"jdbcConnectionProperties": "?ssl=true&client_network_buffer_size=1000000"

​

Monitoramento e solução de problemas de desempenho

1. Consumer lag: Use ferramentas de monitoramento do Kafka para acompanhar o lag por partição
2. Métricas do conector: Monitore receivedRecords, recordProcessingTime, taskProcessingTime via JMX (consulte Monitoring)
3. Métricas do ClickHouse:
   • system.asynchronous_inserts: Monitore o uso do buffer de async insert
   • system.parts: Monitore o número de partes para detectar problemas de merge
   • system.merges: Monitore merges ativos
   • system.events: Acompanhe InsertedRows, InsertedBytes, FailedInsertQuery

​

Resumo das boas práticas

1. Comece com os padrões, depois meça e ajuste com base no desempenho real
2. Prefira lotes maiores: Busque 10.000-100.000 linhas por insert, quando possível
3. Use inserções assíncronas ao enviar muitos lotes pequenos ou em cenários de alta concorrência
4. Sempre use wait_for_async_insert=1 com semântica de exactly-once
5. Escale horizontalmente: Aumente tasks.max até o número de partições
6. Um conector por tópico de alto volume para obter vazão máxima
7. Monitore continuamente: Acompanhe o consumer lag, a contagem de partes e a atividade de merge
8. Teste minuciosamente: Sempre teste alterações de configuração sob carga realista antes da implantação em produção

​

Exemplo: Configuração de alta vazão

{
  "name": "clickhouse-high-throughput",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "tasks.max": "8",
    
    "topics": "high_volume_topic",
    "hostname": "my-clickhouse-host.cloud",
    "port": "8443",
    "database": "default",
    "username": "default",
    "password": "<PASSWORD>",
    "ssl": "true",
    
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    
    "exactlyOnce": "false",
    "ignorePartitionsWhenBatching": "true",
    
    "consumer.override.max.poll.records": "10000",
    "consumer.override.max.partition.fetch.bytes": "5242880",
    "consumer.override.fetch.min.bytes": "1048576",
    "consumer.override.fetch.max.wait.ms": "500",
    
    "clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=16777216,async_insert_busy_timeout_ms=1000,socket_timeout=300000"
  }
}

• Processa até 10.000 registros por poll
• Agrupa partições em lotes para inserts maiores
• Usa inserções assíncronas com buffer de 16 MB
• Executa 8 tarefas em paralelo (corresponda à sua quantidade de partições)
• Otimizada para vazão em vez de ordenação estrita

​

Solução de problemas

​

“Inconsistência de estado para o tópico [someTopic] partição [0]”

-- Primeiro, identifique o banco de dados usado para armazenar os dados.
SELECT * FROM [database].connect_state

-- Identifique a chave que corresponde ao tópico e à partição.
ALTER TABLE [database].connect_state DELETE WHERE key = [keyname]

​

“Quais erros o conector tentará novamente?”

• ClickHouseException - Esta é uma exceção genérica que pode ser lançada pelo ClickHouse. Em geral, ela é lançada quando o servidor está sobrecarregado, e os seguintes códigos de erro são considerados especialmente transitórios:
  • 3 - UNEXPECTED_END_OF_FILE
  • 107 - FILE_DOESNT_EXIST
  • 159 - TIMEOUT_EXCEEDED
  • 164 - READONLY
  • 202 - TOO_MANY_SIMULTANEOUS_QUERIES
  • 203 - NO_FREE_CONNECTION
  • 209 - SOCKET_TIMEOUT
  • 210 - NETWORK_ERROR
  • 241 - MEMORY_LIMIT_EXCEEDED
  • 242 - TABLE_IS_READ_ONLY
  • 252 - TOO_MANY_PARTS
  • 285 - TOO_FEW_LIVE_REPLICAS
  • 319 - UNKNOWN_STATUS_OF_INSERT
  • 425 - SYSTEM_ERROR
  • 999 - KEEPER_EXCEPTION
• SocketTimeoutException - Esta é lançada quando ocorre timeout no socket.
• UnknownHostException - Esta é lançada quando não é possível resolver o host.
• IOException - Esta é lançada quando há um problema de rede.

​

“Todos os meus dados estão em branco/zerados”

transforms=flatten
transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
transforms.flatten.delimiter=_

​

“Quero usar minhas chaves do Kafka no ClickHouse”

transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key