` | MAP | Map(K, V) | ✅ | |
| oneof | STRUCT | Tuple / Variant | ⚠️ | Consulta la sección siguiente sobre cómo traducir oneof al esquema de ClickHouse |
| google.protobuf.DoubleValue | FLOAT64 | Nullable(Float64) | ✅ | |
| google.protobuf.FloatValue | FLOAT32 | Nullable(Float32) | ✅ | |
| google.protobuf.Int64Value | INT64 | Nullable(Int64) | ✅ | |
| google.protobuf.UInt64Value | INT64 | Nullable(UInt64) | ✅ | |
| google.protobuf.UInt32Value | INT64 | Nullable(UInt32) | ✅ | |
| google.protobuf.Int32Value | INT32 | Nullable(Int32) | ✅ | |
| google.protobuf.BoolValue | BOOLEAN | Nullable(Bool) | ✅ | |
| google.protobuf.StringValue | STRING | Nullable(String) | ✅ | |
| google.protobuf.BytesValue | BYTES | Nullable(String) | ✅ | |
| google.protobuf.Timestamp | org.apache.kafka.connect.data.Timestamp | DateTime64(3) | ✅ | |
| google.type.Date | org.apache.kafka.connect.data.Date | Date | ✅ | |
| google.type.TimeOfDay | org.apache.kafka.connect.data.Time | Int32 / Int64 | ✅ | |
| google.protobuf.Duration | STRUCT | Tuple(`seconds` Int64, `nano` Nullable(Int32)) | ✅ | |
| google.protobuf.Any | *N/A* | *N/A* | ❌ | |
| google.protobuf.Empty | *N/A* | *N/A* | ❌ | |
Consulta [Tipos de datos compatibles](#supported-data-types) para ver la asignación entre los tipos de Kafka Connect y los de ClickHouse.
#### Nota sobre la conversión de campos `oneof` en columnas de ClickHouse
El conector no admite convertir uniones de Protobuf (`oneof`) al tipo Variant de ClickHouse. En su lugar, enumera los campos `oneof` como campos anulables individuales en el esquema de tu tabla de ClickHouse.
Por ejemplo:
```protobuf theme={null}
syntax = "proto3";
package com.clickhouse.kafka.connect.proto.test;
message StringIntUnion {
oneof mixed {
string mixed_string = 2;
int32 mixed_int = 3;
}
}
```
se traduce como la siguiente definición de tabla de ClickHouse:
```sql theme={null}
CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
mixed_string Nullable(String),
mixed_int Nullable(Int32)
) ENGINE = ...;
```
#### Esquemas Protobuf no compatibles
Los siguientes esquemas Protobuf no son compatibles con el conector:
* uniones de varios mensajes (**antes de la versión 26.1 de 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;
}
// translates to Nullable(Tuple(label String)) in ClickHouse, which is unsupported
message TypeA {
string label = 1;
}
// translates to Nullable(Tuple(count Int32)) in ClickHouse, which is unsupported
message TypeB {
int32 count = 1;
}
}
```
Desde la versión 26.1 de CH, se admite este esquema cuando `allow_experimental_nullable_tuple_type=1` (consulta [esta página de la documentación](/docs/es/reference/settings/session-settings#allow_experimental_nullable_tuple_type)).
### Compatibilidad con esquemas JSON
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
}
}
```
### Compatibilidad con el convertidor String
El conector admite el convertidor String en diferentes formatos de ClickHouse: [JSON](/docs/es/reference/formats/JSON/JSONEachRow), [CSV](/docs/es/reference/formats/CSV/CSV) y [TSV](/docs/es/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"
}
}
```
### Almacenamiento en búfer interno
El almacenamiento en búfer interno permite que la tarea del sink acumule registros de varias llamadas a `poll()` y los vuelque a ClickHouse en lotes más grandes. Esto puede mejorar el throughput en cargas de trabajo donde cada sondeo produce muchos lotes pequeños por partición.
Comportamiento clave:
* `bufferCount` controla cuántos registros se almacenan en el búfer antes de volcarse.
* `bufferFlushTime` establece un tiempo máximo de espera (en milisegundos) antes de volcar los registros almacenados en el búfer.
* `bufferFlushTime` solo surte efecto cuando `bufferCount > 0`.
* `bufferCount=0` y `bufferFlushTime=0` mantienen el almacenamiento en búfer deshabilitado (comportamiento predeterminado).
* El almacenamiento en búfer no es compatible cuando `exactlyOnce=true`.
Por qué el almacenamiento en búfer es incompatible con el modo exactly-once:
El almacenamiento en búfer cambia los límites de los lotes, lo que rompe la deduplicación de bloques de ClickHouse y la máquina de estados del offset del conector.
Para resolverlo, deshabilita el modo exactly-once con `exactlyOnce=false` en la configuración de tu conector, o deshabilita el almacenamiento en búfer con `bufferCount=0`.
Ejemplo:
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"exactlyOnce": "false",
"bufferCount": "5000",
"bufferFlushTime": "2000"
}
}
```
### Registro
La funcionalidad de registro se proporciona automáticamente mediante Kafka Connect Platform.
El destino y el formato de los registros se pueden configurar mediante el [archivo de configuración](https://docs.confluent.io/platform/current/connect/logging.html#log4j-properties-file) de Kafka Connect.
Si usa Confluent Platform, los registros pueden consultarse ejecutando un comando de la CLI:
```bash theme={null}
confluent local services connect log
```
Para obtener más información, consulta el [tutorial](https://docs.confluent.io/platform/current/connect/logging.html) oficial.
### Monitoreo
ClickHouse Kafka Connect expone métricas de runtime mediante [Java Management Extensions (JMX)](https://www.oracle.com/technical-resources/articles/javase/jmx.html). JMX está habilitado en Kafka Connector de forma predeterminada.
#### Métricas específicas de ClickHouse
El conector expone métricas personalizadas a través del siguiente nombre de MBean:
```java theme={null}
com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}
```
| Nombre de la métrica | Tipo | Descripción |
| ---------------------- | ---- | ------------------------------------------------------------------------------------------------------- |
| `receivedRecords` | long | El número total de registros recibidos. |
| `recordProcessingTime` | long | Tiempo total en nanosegundos empleado en agrupar y convertir los registros en una estructura unificada. |
| `taskProcessingTime` | long | Tiempo total en nanosegundos empleado en procesar e insertar datos en ClickHouse. |
#### Métricas de productor/consumidor de Kafka
El conector expone métricas estándar de productor y consumidor de Kafka que proporcionan información sobre el flujo de datos, el throughput y el rendimiento.
**Métricas a nivel de topic:**
* `records-sent-total`: Número total de registros enviados al topic
* `bytes-sent-total`: Número total de bytes enviados al topic
* `record-send-rate`: Tasa promedio de registros enviados por segundo
* `byte-rate`: Promedio de bytes enviados por segundo
* `compression-rate`: Ratio de compresión alcanzado
**Métricas a nivel de partición:**
* `records-sent-total`: Número total de registros enviados a la partición
* `bytes-sent-total`: Número total de bytes enviados a la partición
* `records-lag`: Lag actual de la partición
* `records-lead`: Lead actual de la partición
* `replica-fetch-lag`: Información de lag de las réplicas
**Métricas de conexión a nivel de nodo:**
* `connection-creation-total`: Número total de conexiones creadas con el nodo de Kafka
* `connection-close-total`: Número total de conexiones cerradas
* `request-total`: Número total de solicitudes enviadas al nodo
* `response-total`: Número total de respuestas recibidas del nodo
* `request-rate`: Tasa promedio de solicitudes por segundo
* `response-rate`: Tasa promedio de respuestas por segundo
Estas métricas ayudan a supervisar:
* **Throughput**: Hacer seguimiento de las tasas de ingestión de datos
* **Lag**: Identificar cuellos de botella y retrasos de procesamiento
* **Compresión**: Medir la eficiencia de la compresión de datos
* **Estado de las conexiones**: Supervisar la conectividad y la estabilidad de la red
#### Métricas de Kafka Connect Framework
El conector se integra con Kafka Connect Framework y expone métricas sobre el ciclo de vida de las tareas y el seguimiento de errores.
**Métricas de estado de las tareas:**
* `task-count`: Número total de tareas en el conector
* `running-task-count`: Número de tareas que se están ejecutando actualmente
* `paused-task-count`: Número de tareas actualmente en pausa
* `failed-task-count`: Número de tareas que han fallado
* `destroyed-task-count`: Número de tareas destruidas
* `unassigned-task-count`: Número de tareas sin asignar
Los valores de estado de las tareas incluyen: `running`, `paused`, `failed`, `destroyed`, `unassigned`
**Métricas de errores:**
* `deadletterqueue-produce-failures`: Número de escrituras fallidas en la DLQ
* `deadletterqueue-produce-requests`: Número total de intentos de escritura en la DLQ
* `last-error-timestamp`: Marca temporal del último error
* `records-skip-total`: Número total de registros omitidos debido a errores
* `records-retry-total`: Número total de registros reintentados
* `errors-total`: Número total de errores detectados
**Métricas de rendimiento:**
* `offset-commit-failures`: Número de commits de offset fallidos
* `offset-commit-avg-time-ms`: Tiempo promedio de los commits de offset
* `offset-commit-max-time-ms`: Tiempo máximo de los commits de offset
* `put-batch-avg-time-ms`: Tiempo promedio para procesar un lote
* `put-batch-max-time-ms`: Tiempo máximo para procesar un lote
* `source-record-poll-total`: Total de registros recuperados
#### Prácticas recomendadas de monitorización
1. **Supervise el consumer lag**: Controle `records-lag` por partición para identificar cuellos de botella en el procesamiento
2. **Controle las tasas de error**: Observe `errors-total` y `records-skip-total` para detectar problemas de calidad de los datos
3. **Observe el estado de las tareas**: Supervise las métricas de estado de las tareas para asegurarse de que se ejecutan correctamente
4. **Mida el rendimiento**: Use `records-send-rate` y `byte-rate` para controlar el rendimiento de la ingestión
5. **Supervise el estado de las conexiones**: Compruebe las métricas de conexión a nivel de nodo para detectar problemas de red
6. **Controle la eficiencia de la compresión**: Use `compression-rate` para optimizar la transferencia de datos
Para ver definiciones detalladas de las métricas JMX y la integración con Prometheus, consulte el archivo de configuración [jmx-export-connector.yml](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/main/jmx-export-connector.yml).
### Limitaciones
* No se admiten eliminaciones.
* El tamaño del lote se hereda de las propiedades del consumer de Kafka.
* Al usar KeeperMap para exactly-once, si se cambia o se restablece el offset, debe eliminar el contenido de KeeperMap para ese topic concreto. (Consulte la guía de solución de problemas a continuación para obtener más detalles)
### Ajuste del rendimiento y optimización del caudal
En esta sección se describen estrategias de ajuste del rendimiento para ClickHouse Kafka Connect Sink. El ajuste del rendimiento es esencial al trabajar con casos de uso de alto caudal o cuando se necesita optimizar el uso de recursos y minimizar la latencia.
#### ¿Cuándo se necesita el ajuste del rendimiento?
El ajuste del rendimiento suele ser necesario en los siguientes escenarios:
* **Cargas de trabajo de alto rendimiento**: al procesar millones de eventos por segundo desde topics de Kafka
* **Consumer lag**: cuando el conector no puede seguir el ritmo de producción de datos, lo que provoca un retraso cada vez mayor
* **Restricciones de recursos**: cuando necesitas optimizar el uso de CPU, memoria o red
* **Múltiples topics**: al consumir simultáneamente de varios topics de gran volumen
* **Mensajes pequeños**: al trabajar con muchos mensajes pequeños que se beneficiarían del procesamiento por lotes en el servidor
El ajuste del rendimiento **NO suele ser necesario** cuando:
* Procesas volúmenes bajos o moderados (\< 10,000 mensajes/segundo)
* El consumer lag es estable y aceptable para tu caso de uso
* La configuración predeterminada del conector ya cumple tus requisitos de rendimiento
* Tu clúster de ClickHouse puede manejar fácilmente la carga entrante
#### Comprender el flujo de datos
Antes de ajustar la configuración, es importante comprender cómo fluyen los datos a través del conector:
1. **Kafka Connect Framework** recupera mensajes de los topics de Kafka en segundo plano
2. **El conector sondea** mensajes del búfer interno del framework
3. **El conector agrupa** los mensajes en lotes según el tamaño del sondeo
4. **ClickHouse recibe** el insert por lotes mediante HTTP/S
5. **ClickHouse procesa** el insert (de forma síncrona o asíncrona)
El rendimiento puede optimizarse en cada una de estas etapas.
#### Ajuste del tamaño del lote en Kafka Connect
El primer nivel de optimización consiste en controlar la cantidad de datos que el conector recibe de Kafka en cada lote.
##### Configuración de `fetch`
Kafka Connect (el framework) recupera mensajes de los topics de Kafka en segundo plano, independientemente del conector:
* **`fetch.min.bytes`**: Cantidad mínima de datos antes de que el framework entregue datos al conector (predeterminado: 1 byte)
* **`fetch.max.bytes`**: Cantidad máxima de datos que se puede recuperar en una sola solicitud (predeterminado: 52428800 / 50 MB)
* **`fetch.max.wait.ms`**: Tiempo máximo de espera antes de devolver datos si no se alcanza `fetch.min.bytes` (predeterminado: 500 ms)
En Confluent Cloud, para ajustar esta configuración es necesario abrir un caso de soporte a través de Confluent Cloud.
##### Configuración de sondeo
El conector sondea mensajes del búfer del framework:
* **`max.poll.records`**: Número máximo de registros devueltos en un único sondeo (predeterminado: 500)
* **`max.partition.fetch.bytes`**: Cantidad máxima de datos por partición (predeterminado: 1048576 / 1 MB)
En Confluent Cloud, para ajustar estos parámetros, es necesario abrir un caso de soporte a través de Confluent Cloud.
##### Configuración recomendada para un alto rendimiento
Para obtener un rendimiento óptimo con ClickHouse, procure usar lotes de mayor tamaño:
```properties theme={null}
# Aumentar el número de registros por sondeo
consumer.override.max.poll.records=5000
# Aumentar el tamaño de fetch por partición (5 MB)
consumer.override.max.partition.fetch.bytes=5242880
# Opcional: Aumentar el tamaño mínimo de fetch para esperar más datos (1 MB)
consumer.override.fetch.min.bytes=1048576
# Opcional: Reducir el tiempo de espera si la latencia es crítica
consumer.override.fetch.max.wait.ms=300
```
Las propiedades anteriores requieren que habilites las anulaciones del cliente en la configuración de tu worker mediante `connector.client.config.override.policy=All`. Consulta la [documentación de Kafka Connect](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) para obtener más información.
**Importante**: Los ajustes de fetch de Kafka Connect representan datos comprimidos, mientras que ClickHouse recibe datos sin comprimir. Equilibra estos ajustes en función de tu ratio de compresión.
**Compromisos**:
* **Lotes más grandes** = Mejor rendimiento de ingestión en ClickHouse, menos partes y menor sobrecarga
* **Lotes más grandes** = Mayor uso de memoria y posible aumento de la latencia de extremo a extremo
* **Lotes demasiado grandes** = Riesgo de timeouts, errores OutOfMemory o de exceder `max.poll.interval.ms`
Más detalles: [documentación de Confluent](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) | [documentación de Kafka](https://kafka.apache.org/documentation/#consumerconfigs)
#### Inserciones asíncronas
Las inserciones asíncronas son una característica muy útil cuando el conector envía lotes relativamente pequeños o cuando desea optimizar aún más la ingestión trasladando a ClickHouse la responsabilidad del procesamiento por lotes.
##### Cuándo usar inserción asíncrona
Considera habilitar la inserción asíncrona cuando:
* **Muchos lotes pequeños**: Tu conector envía lotes pequeños con frecuencia (\< 1000 filas por lote)
* **Alta concurrencia**: Varias tareas del conector escriben en la misma tabla
* **Implementación distribuida**: Ejecutas muchas instancias del conector en distintos hosts
* **Sobrecarga por creación de partes**: Estás experimentando errores de "too many parts"
* **Carga de trabajo mixta**: Combinas la ingestión en tiempo real con cargas de trabajo de consultas
**NO** uses la inserción asíncrona cuando:
* Ya estás enviando lotes grandes (> 10,000 filas por lote) con una frecuencia controlada
* Necesitas visibilidad inmediata de los datos (las consultas deben ver los datos al instante)
* La semántica exactly-once con `wait_for_async_insert=0` entra en conflicto con tus requisitos
* Tu caso de uso puede beneficiarse más de mejoras en la agrupación en lotes del lado del cliente
##### Cómo funcionan las inserciones asíncronas
Con las inserciones asíncronas habilitadas, ClickHouse:
1. Recibe la consulta de inserción del conector
2. Escribe los datos en un búfer en memoria (en lugar de escribirlos inmediatamente en disco)
3. Devuelve una respuesta de éxito al conector (si `wait_for_async_insert=0`)
4. Vuelca el búfer a disco cuando se cumple una de estas condiciones:
* El búfer alcanza `async_insert_max_data_size` (valor predeterminado: 100 MB)
* Han transcurrido `async_insert_busy_timeout_ms` milisegundos desde la primera inserción (valor predeterminado: 1000 ms)
* Se alcanza el número máximo de consultas acumuladas (`async_insert_max_query_number`, valor predeterminado: 100)
Esto reduce significativamente la cantidad de partes creadas y mejora el rendimiento general.
##### Habilitar inserción asíncrona
Añada la configuración de `async insert` al parámetro de configuración `clickhouseSettings`:
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
}
}
```
**Configuraciones clave**:
* **`async_insert=1`**: Habilita las inserciones asíncronas
* **`wait_for_async_insert=1`** (recomendado): El conector espera a que los datos se escriban en el almacenamiento de ClickHouse antes de confirmar la recepción. Proporciona garantías de entrega.
* **`wait_for_async_insert=0`**: El conector confirma la recepción inmediatamente después de almacenarlos en búfer. Ofrece mejor rendimiento, pero los datos pueden perderse si el servidor falla antes de que se escriban en disco.
##### Ajuste del comportamiento de `async insert`
Puede afinar el comportamiento de vaciado de `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 comunes:
* **`async_insert_max_data_size`** (valor predeterminado: 104857600 / 100 MB): Tamaño máximo del búfer antes del vaciado
* **`async_insert_busy_timeout_ms`** (valor predeterminado: 1000): Tiempo máximo (ms) antes del vaciado
* **`async_insert_stale_timeout_ms`** (valor predeterminado: 0): Tiempo (ms) desde la última inserción antes del vaciado
* **`async_insert_max_query_number`** (valor predeterminado: 100): Número máximo de consultas antes del vaciado
**Compromisos**:
* **Beneficios**: Menos partes, mejor rendimiento de las fusiones, menor sobrecarga de CPU, mayor rendimiento con alta concurrencia
* **Consideraciones**: Los datos no se pueden consultar de inmediato, latencia de extremo a extremo ligeramente mayor
* **Riesgos**: Pérdida de datos si el servidor falla y `wait_for_async_insert=0`, posible presión de memoria con búferes grandes
##### Inserciones asíncronas con semántica de exactamente una vez
Cuando se usa `exactlyOnce=true` con inserciones asíncronas:
```json theme={null}
{
"config": {
"exactlyOnce": "true",
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
}
}
```
**Importante**: Usa siempre `wait_for_async_insert=1` con exactly-once para garantizar que las confirmaciones de offsets solo se produzcan después de que los datos se hayan guardado de forma persistente.
Para obtener más información sobre la inserción asíncrona, consulta la [documentación de inserción asíncrona de ClickHouse](/docs/es/concepts/best-practices/selecting-an-insert-strategy#asynchronous-inserts).
#### Paralelismo del conector
Aumente el paralelismo para mejorar el caudal:
##### Tareas por conector
```json theme={null}
"tasks.max": "4"
```
Cada tarea procesa un subconjunto de particiones del topic. Más tareas = más paralelismo, pero:
* El número máximo de tareas efectivas = número de particiones del topic
* Cada tarea mantiene su propia conexión a ClickHouse
* Más tareas = más sobrecarga y posible contención de recursos
**Recomendación**: Empieza con `tasks.max` igual al número de particiones del topic y luego ajústalo en función de la CPU y las métricas de caudal.
##### Ignorar las particiones al agrupar en lotes
De forma predeterminada, el conector agrupa los mensajes por partición. Para obtener un mayor caudal, puede agruparlos entre particiones:
```json theme={null}
"ignorePartitionsWhenBatching": "true"
```
\*\* Advertencia\*\*: Úselo solo cuando `exactlyOnce=false`. Esta configuración puede mejorar el caudal al crear lotes más grandes, pero se pierden las garantías de orden dentro de cada partición.
#### Varios topics de alto caudal
Si el conector está configurado para suscribirse a varios topics, se utiliza `topic2TableMap` para asignar topics a tablas y se está produciendo un cuello de botella en la inserción que provoca consumer lag, considere crear un conector por topic en su lugar.
La razón principal es que, actualmente, los lotes se insertan en cada tabla [en serie](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/578ac07e8be1a920aaa3b26e49183595c3edd04b/src/main/java/com/clickhouse/kafka/connect/sink/ProxySinkTask.java#L95-L100).
**Recomendación**: Para varios topics de alto volumen, implemente una instancia de conector por topic para maximizar el caudal de inserción en paralelo.
#### Consideraciones sobre el motor de tabla de ClickHouse
Elige el motor de tabla de ClickHouse adecuado para tu caso de uso:
* **`MergeTree`**: La mejor opción para la mayoría de los casos de uso; equilibra el rendimiento de las consultas y las inserciones
* **`ReplicatedMergeTree`**: Requerido para alta disponibilidad; añade sobrecarga de replicación
* **`*MergeTree` con un `ORDER BY` adecuado**: Optimiza según tus patrones de consulta
**Configuraciones a tener en cuenta**:
```sql theme={null}
CREATE TABLE my_table (...)
ENGINE = MergeTree()
ORDER BY (timestamp, id)
SETTINGS
-- Aumentar el máximo de hilos de inserción para escritura paralela de partes
max_insert_threads = 4,
-- Permitir inserciones con quórum para mayor confiabilidad (ReplicatedMergeTree)
insert_quorum = 2
```
Para la configuración de inserción a nivel de conector:
```json theme={null}
"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"
```
#### Pool de conexiones y tiempos de espera
El conector mantiene conexiones HTTP con ClickHouse. Ajuste los tiempos de espera para redes de alta latencia:
```json theme={null}
"clickhouseSettings": "socket_timeout=300000,connection_timeout=30000"
```
* **`socket_timeout`** (predeterminado: 30000 ms): Tiempo máximo para las operaciones de lectura
* **`connection_timeout`** (predeterminado: 10000 ms): Tiempo máximo para establecer la conexión
Aumente estos valores si experimenta errores de timeout con lotes grandes.
#### Supervisión y solución de problemas de rendimiento
Supervise estas métricas clave:
1. **Consumer lag**: Use las herramientas de supervisión de Kafka para seguir el retraso por partición
2. **Métricas del conector**: Supervise `receivedRecords`, `recordProcessingTime`, `taskProcessingTime` mediante JMX (consulte [Monitoring](#monitoring))
3. **Métricas de ClickHouse**:
* `system.asynchronous_inserts`: Supervise el uso del búfer de inserción asíncrona
* `system.parts`: Supervise el número de partes para detectar problemas de fusiones
* `system.merges`: Supervise las fusiones activas
* `system.events`: Siga `InsertedRows`, `InsertedBytes`, `FailedInsertQuery`
**Problemas de rendimiento comunes**:
| Síntoma | Posible causa | Solución |
| ------------------------ | ------------------------------------ | ---------------------------------------------------------------------- |
| Consumer lag alto | Lotes demasiado pequeños | Aumente `max.poll.records`, habilite la inserción asíncrona |
| Errores "Too many parts" | Inserts pequeños y frecuentes | Habilite la inserción asíncrona, aumente el tamaño del lote |
| Errores de timeout | Tamaño de lote grande, red lenta | Reduzca el tamaño del lote, aumente `socket_timeout`, compruebe la red |
| Uso elevado de CPU | Demasiadas partes pequeñas | Habilite la inserción asíncrona, aumente la configuración de fusiones |
| Errores OutOfMemory | Tamaño de lote demasiado grande | Reduzca `max.poll.records`, `max.partition.fetch.bytes` |
| Carga desigual de tareas | Distribución desigual de particiones | Reequilibre las particiones o ajuste `tasks.max` |
#### Resumen de prácticas recomendadas
1. **Empieza con los valores predeterminados** y luego mide y ajusta en función del rendimiento real
2. **Prefiere lotes más grandes**: apunta a 10,000-100,000 filas por inserción cuando sea posible
3. **Usa inserción asíncrona** cuando envíes muchos lotes pequeños o haya alta concurrencia
4. **Usa siempre `wait_for_async_insert=1`** con semántica exactly-once
5. **Escala horizontalmente**: aumenta `tasks.max` hasta el número de particiones
6. **Un conector por topic de alto volumen** para obtener el máximo caudal
7. **Supervisa continuamente**: controla el consumer lag, la cantidad de partes y la actividad de fusión
8. **Haz pruebas exhaustivas**: prueba siempre los cambios de configuración con una carga realista antes de la implementación en producción
#### Ejemplo: Configuración de alto caudal
A continuación se muestra un ejemplo completo optimizado para alto caudal:
```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"
}
}
```
La configuración del conector anterior requiere que habilites las anulaciones de cliente en la configuración de tu worker mediante `connector.client.config.override.policy=All`. Consulta la [documentación de Kafka Connect](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) para obtener más información.
**Esta configuración**:
* Procesa hasta 10.000 registros por ciclo de sondeo
* Agrupa en lotes entre particiones para inserciones más grandes
* Usa inserción asíncrona con un búfer de 16 MB
* Ejecuta 8 tareas en paralelo (haz que coincida con tu número de particiones)
* Está optimizada para el caudal por encima del orden estricto
### Solución de problemas
#### "Incongruencia de estado para el topic `[someTopic]` partición `[0]`"
Esto ocurre cuando el offset almacenado en KeeperMap es distinto del offset almacenado en Kafka, normalmente cuando se ha eliminado un topic
o el offset se ha ajustado manualmente.
Para solucionarlo, debes eliminar los valores antiguos almacenados para ese topic + partición:
```sql theme={null}
-- Primero, identifica la base de datos utilizada para almacenar los datos.
SELECT * FROM [database].connect_state
-- Identifica la clave que coincide con el topic y la partición.
ALTER TABLE [database].connect_state DELETE WHERE key = [keyname]
```
Este ajuste puede afectar las garantías exactly-once.
#### "¿Qué errores reintentará el conector?"
Ahora mismo, el objetivo es identificar errores transitorios que puedan reintentarse, entre ellos:
* `ClickHouseException` - Esta es una excepción genérica que puede lanzar ClickHouse.
Suele producirse cuando el server está sobrecargado, y los siguientes códigos de error se consideran especialmente transitorios:
* 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` - Se lanza cuando el socket agota el tiempo de espera.
* `UnknownHostException` - Se lanza cuando no se puede resolver el host.
* `IOException` - Se lanza cuando hay un problema con la red.
#### "Todos mis datos están en blanco o en cero"
Es probable que los campos de sus datos no coincidan con los campos de la tabla; esto es especialmente común con CDC (y el formato Debezium).
Una solución habitual es añadir la transformación flatten a la configuración de su conector:
```properties theme={null}
transforms=flatten
transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
transforms.flatten.delimiter=_
```
Esto transformará tus datos de un JSON anidado a un JSON plano (usando `_` como delimitador). Los campos de la tabla seguirán entonces el formato "field1\_field2\_field3" (es decir, "before\_id", "after\_id", etc.).
#### "Quiero usar las claves de Kafka en ClickHouse"
De forma predeterminada, las claves de Kafka no se almacenan en el campo `value`, pero puedes usar la transformación `KeyToValue` para mover la clave al campo `value` (con el nuevo nombre de campo `_key`):
```properties theme={null}
transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key
```
