Saltar al contenido principal
Si necesitas ayuda, abre una incidencia en el repositorio o plantea tu pregunta en el Slack público de ClickHouse.
ClickHouse Kafka Connect Sink es el conector de Kafka que envía datos desde un topic de Kafka a una tabla de ClickHouse.

​
Licencia

El Kafka Connector Sink se distribuye bajo la licencia Apache 2.0

​
Requisitos del entorno

El framework Kafka Connect v2.7 o una versión posterior debe estar instalado en el entorno.

​
Matriz de compatibilidad de versiones

​
Funcionalidades principales

  • Incluye semántica exactly-once lista para usar. Se basa en una nueva funcionalidad del núcleo de ClickHouse llamada KeeperMap (utilizada como almacén de estado por el conector) y permite una arquitectura minimalista.
  • Soporte para almacenes de estado de terceros: actualmente usa In-memory de forma predeterminada, pero puede utilizar KeeperMap (Redis se añadirá próximamente).
  • Integración principal: desarrollada, mantenida y respaldada por ClickHouse.
  • Probada continuamente con ClickHouse Cloud.
  • Inserciones de datos con esquema declarado y sin esquema.
  • Soporte para todos los tipos de datos de ClickHouse.

​
Instrucciones de instalación

​
Recopila los datos de conexión

Para conectarse a ClickHouse con HTTP(S), necesita esta información: Los detalles de su servicio de ClickHouse Cloud están disponibles en la consola de ClickHouse Cloud. Seleccione un servicio y haga clic en Connect: Elija HTTPS. Los detalles de conexión se muestran en un comando curl de ejemplo. Si usa ClickHouse autogestionado, los detalles de conexión los establece su administrador de ClickHouse.

​
Instrucciones generales de instalación

El conector se distribuye como un único archivo JAR que contiene todos los archivos de clase necesarios para ejecutar el plugin. Para instalar el plugin, siga estos pasos:
  • Descargue un archivo ZIP que contenga el archivo JAR del conector desde la página de Releases del repositorio de ClickHouse Kafka Connect Sink.
  • Extraiga el contenido del archivo ZIP y cópielo en la ubicación deseada.
  • Agregue la ruta al directorio del plugin en la configuración plugin.path de su archivo de propiedades de Connect para que Confluent Platform pueda encontrar el plugin.
  • Proporcione en la configuración un nombre de topic, el hostname de la instancia de ClickHouse y la contraseña.
  • Reinicie Confluent Platform.
  • Si usa Confluent Platform, inicie sesión en la UI de Confluent Control Center para verificar que ClickHouse Sink esté disponible en la lista de conectores disponibles.

​
Opciones de configuración

Para conectar el ClickHouse Sink al ClickHouse server, debe proporcionar:
  • detalles de conexión: hostname (obligatorio) y puerto (opcional)
  • credenciales de usuario: contraseña (obligatoria) y nombre de usuario (opcional)
  • clase del conector: com.clickhouse.kafka.connect.ClickHouseSinkConnector (obligatoria)
  • topics o topics.regex: los topics de Kafka que se van a consultar; los nombres de los topics deben coincidir con los nombres de las tablas (obligatorio)
  • convertidores de clave y valor: configúrelos en función del tipo de datos de su topic. Es obligatorio si aún no están definidos en la configuración del worker.
La tabla completa de opciones de configuración:

​
Tablas de destino

ClickHouse Connect Sink lee mensajes de los topics de Kafka y los escribe en las tablas correspondientes. ClickHouse Connect Sink escribe datos en tablas ya existentes. Asegúrese de que se haya creado en ClickHouse una tabla de destino con un esquema adecuado antes de empezar a insertar datos en ella. Cada topic requiere su propia tabla de destino en ClickHouse. El nombre de la tabla de destino debe coincidir con el nombre del topic de origen.

​
Preprocesamiento

Si necesitas transformar los mensajes salientes antes de enviarlos a ClickHouse Kafka Connect Sink, usa las transformaciones de Kafka Connect.

​
Tipos de datos compatibles

Con un esquema declarado:
  • (1) - JSON solo es compatible cuando la configuración de ClickHouse incluye input_format_binary_read_json_as_string=1. Esto solo funciona con la familia de formatos RowBinary, y la configuración afecta a todas las columnas de la solicitud de insert, por lo que todas deben ser de tipo cadena. En este caso, el conector convertirá STRUCT en una cadena JSON.
  • (2) - Cuando struct tiene unions como oneof, el convertidor debe configurarse para NO añadir prefijos/sufijos a los nombres de los campos. Existe la configuración generate.index.for.unions=false para ProtobufConverter.
Sin un esquema declarado: Un registro se convierte a JSON y se envía a ClickHouse como un valor en formato JSONEachRow.

​
Recetas de configuración

Estas son algunas recetas de configuración habituales para empezar rápidamente.

​
Configuración básica

La configuración más sencilla para empezar: supone que estás ejecutando Kafka Connect en modo distribuido y que tienes un servidor de ClickHouse en ejecución en localhost:8443 con SSL habilitado; los datos están en JSON sin esquema.
La configuración del conector anterior requiere que habilites la sobrescritura de la configuración del cliente en la configuración de tu worker mediante connector.client.config.override.policy=All. Consulta la documentación de Kafka Connect para obtener más información.

​
Configuración básica con varios topics

El conector puede consumir datos de varios topics

​
Configuración básica con DLQ

​
Uso con distintos formatos de datos

Compatibilidad con esquemas Avro
Asignación de tipos de Avro
La asignación de tipos siguiente está definida por io.confluent.connect.avro.AvroConverter, la implementación oficial de serialización/deserialización de Avro en Kafka Connect. Consulta la documentación de Kafka Connect para obtener información avanzada sobre la lógica de conversión. ✅: Compatible ❌: No compatible ️⚠️: Parcialmente compatible Consulta Tipos de datos compatibles para ver la asignación entre los tipos de Kafka Connect y los tipos de ClickHouse.
Esquemas de Avro no compatibles
Los siguientes esquemas de Avro no son compatibles con el conector:
  • tipo lógico decimal de longitud fija
  • uniones nullable
  • uniones de registro
Compatibilidad con esquemas de Protobuf
Tenga en cuenta: si tiene problemas por clases faltantes, no todos los entornos incluyen el convertidor de protobuf y es posible que necesite una versión alternativa del JAR que incluya las dependencias.
Asignación de tipos de Protobuf
La asignación de tipos que aparece a continuación está definida por io.confluent.connect.protobuf.ProtobufConverter, la implementación oficial de serialización/deserialización de Protobuf en Kafka Connect. Consulta la documentación de Kafka Connect para obtener información avanzada sobre la lógica de conversión. ✅: Compatible ❌: No compatible ️⚠️: Compatibilidad parcial Consulta Tipos de datos compatibles 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:
se traduce como la siguiente definición de tabla de ClickHouse:
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)
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).
Compatibilidad con esquemas JSON
Compatibilidad con el convertidor String
El conector admite el convertidor String en diferentes formatos de ClickHouse: JSON, CSV y TSV.

​
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:

​
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 de Kafka Connect. Si usa Confluent Platform, los registros pueden consultarse ejecutando un comando de la CLI:
Para obtener más información, consulta el tutorial oficial.

​
Monitoreo

ClickHouse Kafka Connect expone métricas de runtime mediante Java Management Extensions (JMX). 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:

​
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.

​
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.
Para obtener un rendimiento óptimo con ClickHouse, procure usar lotes de mayor tamaño:
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 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 | documentación de Kafka

​
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:
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:
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:
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.

​
Paralelismo del conector

Aumente el paralelismo para mejorar el caudal:
Tareas por conector
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:
** 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. 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:
Para la configuración de inserción a nivel de conector:

​
Pool de conexiones y tiempos de espera

El conector mantiene conexiones HTTP con ClickHouse. Ajuste los tiempos de espera para redes de alta latencia:
  • 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)
  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:

​
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:
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 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:
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:
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):
Última modificación el 25 de junio de 2026