` | MAP | ✅ | |
| oneof | STRUCT | ⚠️ | Consulte a seção abaixo sobre a tradução de oneof para schema do ClickHouse |
| google.protobuf.DoubleValue | FLOAT64 | ✅ | |
| google.protobuf.FloatValue | FLOAT32 | ✅ | |
| google.protobuf.Int64Value | INT64 | ✅ | |
| google.protobuf.UInt64Value | INT64 | ✅ | |
| google.protobuf.UInt32Value | INT64 | ✅ | |
| google.protobuf.Int32Value | INT32 | ✅ | |
| google.protobuf.BoolValue | BOOLEAN | ✅ | |
| google.protobuf.StringValue | STRING | ✅ | |
| google.protobuf.BytesValue | BYTES | ✅ | |
| google.protobuf.Timestamp | org.apache.kafka.connect.data.Timestamp | ✅ | |
| google.type.Date | org.apache.kafka.connect.data.Date | ✅ | |
| google.type.TimeOfDay | org.apache.kafka.connect.data.Time | ✅ | |
Consulte [Tipos de dados suportados](#supported-data-types) para ver o mapeamento entre os tipos do Kafka Connect e os tipos do ClickHouse.
###### Observação sobre a tradução de campos `oneof` para colunas do ClickHouse
O conector não oferece suporte à tradução de unions (`oneof`) do Protobuf para o tipo Variant do ClickHouse. Em vez disso, liste os campos `oneof` como campos Nullable individuais no schema da sua tabela do ClickHouse.
Por exemplo:
```protobuf theme={null}
syntax = "proto3";
package com.clickhouse.kafka.connect.proto.test;
message StringIntUnion {
oneof mixed {
string mixed_string = 2;
int32 mixed_int = 3;
}
}
```
é traduzido na seguinte definição de tabela do ClickHouse:
```sql theme={null}
CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
mixed_string Nullable(String),
mixed_int Nullable(Int32)
) ENGINE = ...;
```
###### Esquemas Protobuf sem suporte
Os seguintes esquemas Protobuf não são suportados pelo conector:
* uniões com várias mensagens (**antes da versão 26.1 do CH**)
```protobuf theme={null}
syntax = "proto3";
package com.clickhouse.kafka.connect.proto.test;
message TwoRecords {
oneof payload {
TypeA type_a = 2;
TypeB type_b = 3;
}
// traduz para Nullable(Tuple(label String)) no ClickHouse, o que não é suportado
message TypeA {
string label = 1;
}
// traduz para Nullable(Tuple(count Int32)) no ClickHouse, o que não é suportado
message TypeB {
int32 count = 1;
}
}
```
A partir da versão 26.1 do CH, esse schema é suportado quando `allow_experimental_nullable_tuple_type=1` (consulte [esta página da documentação](/pt-BR/reference/settings/session-settings#allow_experimental_nullable_tuple_type)).
##### Suporte a schema JSON
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
}
}
```
##### Suporte a String
O conector oferece suporte ao String Converter em diferentes formatos do ClickHouse: [JSON](/pt-BR/reference/formats/JSON/JSONEachRow), [CSV](/pt-BR/reference/formats/CSV/CSV) e [TSV](/pt-BR/reference/formats/TabSeparated/TabSeparated).
```json theme={null}
{
"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
O buffering interno permite que a tarefa do sink acumule registros de várias chamadas `poll()` e os envie ao ClickHouse em batches maiores. Isso pode melhorar a vazão em workloads em que cada `poll()` produz muitos batches pequenos por partição.
Comportamento principal:
* `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`.
Por que o buffering é incompatível com o modo exactly-once:
O buffering altera os limites dos batches, o que quebra a desduplicação de blocos do ClickHouse e a máquina de estados de offset do conector.
Para resolver isso, desative o modo exactly-once com `exactlyOnce=false` na configuração do conector ou desative o buffering com `bufferCount=0`.
Exemplo:
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"exactlyOnce": "false",
"bufferCount": "5000",
"bufferFlushTime": "2000"
}
}
```
### Logging
O logging é fornecido automaticamente pela plataforma Kafka Connect.
O destino e o formato dos logs podem ser configurados por meio do [arquivo de configuração](https://docs.confluent.io/platform/current/connect/logging.html#log4j-properties-file) do Kafka Connect.
Se estiver usando o Confluent Platform, os logs poderão ser visualizados executando um comando de CLI:
```bash theme={null}
confluent local services connect log
```
Para mais detalhes, confira o [tutorial oficial](https://docs.confluent.io/platform/current/connect/logging.html).
### Monitoramento
O ClickHouse Kafka Connect reporta métricas de runtime por meio do [Java Management Extensions (JMX)](https://www.oracle.com/technical-resources/articles/javase/jmx.html). O JMX vem habilitado no Kafka Connector por padrão.
#### Métricas específicas do ClickHouse
O conector expõe métricas personalizadas por meio do seguinte nome de MBean:
```java theme={null}
com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}
```
| Nome da métrica | Tipo | Descrição |
| ---------------------- | ---- | ------------------------------------------------------------------------------------------------------------- |
| `receivedRecords` | long | Número total de registros recebidos. |
| `recordProcessingTime` | long | Tempo total, em nanossegundos, gasto no agrupamento e na conversão de registros para uma estrutura unificada. |
| `taskProcessingTime` | long | Tempo total, em nanossegundos, gasto no processamento e na inserção de dados no ClickHouse. |
#### Métricas de Produtor/Consumidor do Kafka
O conector expõe métricas padrão de produtor e consumidor do Kafka que fornecem informações sobre o fluxo de dados, a taxa de transferência e o desempenho.
**Métricas no Nível do Tópico:**
* `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
**Métricas no Nível da Partição:**
* `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
**Métricas de Conexão no Nível do Nó:**
* `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
Essas métricas ajudam a monitorar:
* **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
O conector se integra ao framework do Kafka Connect e expõe métricas sobre o ciclo de vida das tarefas e o rastreamento de erros.
**Métricas de status das tarefas:**
* `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
Os valores de status das tarefas incluem: `running`, `paused`, `failed`, `destroyed`, `unassigned`
**Métricas de erro:**
* `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
**Métricas de desempenho:**
* `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
Para ver definições detalhadas das métricas JMX e a integração com o Prometheus, consulte o arquivo de configuração [jmx-export-connector.yml](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/main/jmx-export-connector.yml).
### 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
Esta seção aborda estratégias de ajuste de desempenho para o ClickHouse Kafka Connect Sink. O ajuste de desempenho é essencial ao lidar com casos de uso de alta vazão ou quando é necessário otimizar o uso de recursos e minimizar o lag.
#### Quando o ajuste de desempenho é necessário?
O ajuste de desempenho normalmente é necessário nos seguintes cenários:
* **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
O ajuste de desempenho **normalmente NÃO é necessário** quando:
* 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
Antes de ajustar, é importante entender como os dados fluem pelo conector:
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)
O desempenho pode ser otimizado em cada uma dessas etapas.
#### Ajuste do tamanho do lote no Kafka Connect
O primeiro nível de otimização é controlar a quantidade de dados que o conector recebe por lote do Kafka.
##### Configurações de fetch
O Kafka Connect (o framework) busca mensagens de tópicos do Kafka em segundo plano, independentemente do conector:
* **`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)
***
No Confluent Cloud, para ajustar essas configurações, é necessário abrir um chamado de suporte pelo Confluent Cloud.
##### Configurações de polling
O conector busca mensagens no buffer do framework por polling:
* **`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)
***
No Confluent Cloud, para ajustar essas configurações, é necessário abrir um chamado de suporte pelo Confluent Cloud.
##### Configurações recomendadas para alta vazão
Para obter o melhor desempenho com ClickHouse, procure usar lotes maiores:
```properties theme={null}
# 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
```
As propriedades acima exigem que você habilite overrides de cliente na configuração do worker por meio de `connector.client.config.override.policy=All`. Consulte a [documentação do Kafka Connect](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) para mais informações.
**Importante**: As configurações de fetch do Kafka Connect representam dados compactados, enquanto o ClickHouse recebe dados não compactados. Ajuste essas configurações com base na sua taxa de compressão.
**Trade-offs**:
* **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`
Mais detalhes: [documentação da Confluent](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) | [documentação do Kafka](https://kafka.apache.org/documentation/#consumerconfigs)
#### Inserções assíncronas
As inserções assíncronas são um recurso poderoso quando o conector envia lotes relativamente pequenos ou quando você quer otimizar ainda mais a ingestão, transferindo para o ClickHouse a responsabilidade pelo agrupamento em lotes.
##### Quando usar inserções assíncronas
Considere ativar inserções assíncronas quando:
* **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
**NÃO** use inserções assíncronas quando:
* 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
##### Como as inserções assíncronas funcionam
Com as inserções assíncronas ativadas, o ClickHouse:
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)
Isso reduz significativamente o número de partes criadas e melhora a vazão geral.
##### Habilitando inserções assíncronas
Adicione as configurações de async insert ao parâmetro de configuração `clickhouseSettings`:
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
}
}
```
**Principais configurações**:
* **`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.
##### Ajuste do comportamento de inserção assíncrona
Você pode ajustar o comportamento de flush do async insert:
```json theme={null}
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=104857600,async_insert_busy_timeout_ms=1000"
```
Parâmetros de ajuste comuns:
* **`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
**Trade-offs**:
* **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
##### Inserções assíncronas com semântica de exactly-once
Ao usar `exactlyOnce=true` com inserções assíncronas:
```json theme={null}
{
"config": {
"exactlyOnce": "true",
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
}
}
```
**Importante**: Sempre use `wait_for_async_insert=1` com exactly-once para garantir que os commits de offset ocorram somente depois que os dados forem persistidos.
Para mais informações sobre async inserts, consulte a [documentação de async inserts do ClickHouse](/pt-BR/concepts/best-practices/selecting-an-insert-strategy#asynchronous-inserts).
#### Paralelismo do conector
Aumente o paralelismo para melhorar a vazão:
##### Tarefas por conector
```json theme={null}
"tasks.max": "4"
```
Cada tarefa processa um subconjunto das partições do tópico. Mais tarefas = mais paralelismo, mas:
* 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
**Recomendação**: Comece com `tasks.max` igual ao número de partições do tópico e depois ajuste com base nas métricas de CPU e vazão.
##### Ignorando partições no agrupamento em lotes
Por padrão, o conector agrupa as mensagens em lotes por partição. Para obter maior vazão, você pode agrupar em lotes entre partições:
```json theme={null}
"ignorePartitionsWhenBatching": "true"
```
\*\* Aviso\*\*: Use apenas quando `exactlyOnce=false`. Essa configuração pode melhorar a vazão ao criar batches maiores, mas elimina as garantias de ordenação em cada partição.
#### Múltiplos tópicos de alta vazão
Se o conector estiver configurado para assinar vários tópicos, você estiver usando `topic2TableMap` para mapear tópicos para tabelas e estiver enfrentando um gargalo na inserção, o que resulta em consumer lag, considere criar um conector por tópico.
O principal motivo para isso acontecer é que, atualmente, os lotes são inseridos em cada tabela [em série](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/578ac07e8be1a920aaa3b26e49183595c3edd04b/src/main/java/com/clickhouse/kafka/connect/sink/ProxySinkTask.java#L95-L100).
**Recomendação**: para vários tópicos de alto volume, implante uma instância de conector por tópico para maximizar a vazão de inserção em paralelo.
#### Considerações sobre o engine de tabela do ClickHouse
Escolha o engine de tabela do ClickHouse adequado para seu caso de uso:
* **`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
**Configurações a considerar**:
```sql theme={null}
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
```
Para as configurações de `insert` em nível de conector:
```json theme={null}
"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"
```
#### Pool de conexões e timeouts
O conector mantém conexões HTTP com o ClickHouse. Ajuste os timeouts para redes com alta latência:
```json theme={null}
"clickhouseSettings": "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
Aumente esses valores se ocorrerem erros de timeout com lotes grandes.
#### Monitoramento e solução de problemas de desempenho
Monitore estas métricas principais:
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](#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`
**Problemas comuns de desempenho**:
| Sintoma | Possível causa | Solução |
| ---------------------------- | ---------------------------------- | -------------------------------------------------------------------- |
| Consumer lag alto | Lotes pequenos demais | Aumente `max.poll.records`, habilite inserções assíncronas |
| Erros de "Too many partes" | Inserções pequenas e frequentes | Habilite inserções assíncronas, aumente o tamanho do lote |
| Erros de timeout | Lote grande, rede lenta | Reduza o tamanho do lote, aumente `socket_timeout`, verifique a rede |
| Alto uso de CPU | Muitas partes pequenas | Habilite inserções assíncronas, aumente as configurações de merge |
| Erros de OutOfMemory | Tamanho do lote grande demais | Reduza `max.poll.records`, `max.partition.fetch.bytes` |
| Carga desigual entre tarefas | Distribuição desigual de partições | Rebalanceie as partições ou ajuste `tasks.max` |
#### 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
Aqui está um exemplo completo otimizado para alta vazão:
```json theme={null}
{
"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": "",
"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"
}
}
```
A configuração do conector acima exige que você habilite sobrescritas de cliente na configuração do seu worker por meio de `connector.client.config.override.policy=All`. Consulte a [documentação do Kafka Connect](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) para mais informações.
**Esta configuração**:
* 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]`"
Isso acontece quando o offset armazenado no KeeperMap é diferente do offset armazenado no Kafka, geralmente quando um tópico foi excluído
ou quando o offset foi ajustado manualmente.
Para corrigir isso, você precisará excluir os valores antigos armazenados para esse tópico + partição:
```sql theme={null}
-- 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]
```
Esse ajuste pode ter implicações para a semântica exactly-once.
#### "Quais erros o conector tentará novamente?"
No momento, o foco está em identificar erros transitórios para os quais é possível fazer nova tentativa, incluindo:
* `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"
Provavelmente, os campos nos seus dados não correspondem aos campos da tabela — isso é especialmente comum com CDC (e o formato Debezium).
Uma solução comum é adicionar a transformação `flatten` à configuração do seu conector:
```properties theme={null}
transforms=flatten
transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
transforms.flatten.delimiter=_
```
Isso transformará seus dados de um JSON aninhado em um JSON plano (usando `_` como delimitador). Os campos na tabela passarão então a seguir o formato "campo1\_campo2\_campo3" (ou seja, "before\_id", "after\_id", etc.).
#### "Quero usar minhas chaves do Kafka no ClickHouse"
As chaves do Kafka não são armazenadas no campo `value` por padrão, mas você pode usar a transformação `KeyToValue` para mover a chave para o campo `value` (com o novo nome de campo `_key`):
```properties theme={null}
transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key
```