Kafka Engine ClickHouse: atomicidad sin pérdidas

Motor Kafka en ClickHouse: Garantizando inserciones atómicas sin pérdida de datos

Las inserciones individuales frecuentes en tablas MergeTree (10,000 INSERT/s) hacen que el rendimiento caiga a 59 filas/s. Para ReplicatedMergeTree, baja a 6 filas/s, y con quorum, solo 2 filas/s. Esto se debe a la creación de una multitud de pequeñas partes de datos, que se fusionan periódicamente según la configuración max_merge_selecting_sleep_ms. El proceso de fusión sobrecarga la CPU y la E/S, reduciendo la velocidad general.

Las tablas buffer parecen una solución: los datos se acumulan en RAM y se vacían en la tabla objetivo en lotes (por tiempo, tamaño o número de filas). Esto minimiza el número de archivos y la carga de fusión. Sin embargo, si ClickHouse se cuelga, los datos no vaciados en memoria se pierden.

Tablas Buffer: Mecanismo y riesgos

Una tabla buffer utiliza el motor Buffer, acumulando datos en memoria hasta que se cumplen las condiciones de vaciado: max_time_to_flush (p. ej., 100 s), min_time_to_flush, max_rows_to_flush (1 millón de filas), max_bytes_to_flush (1 GB). El vaciado ocurre de forma atómica en MergeTree, pero solo después de escribir en disco.

Riesgo clave: Si el servidor se cuelga (kill -9, OOM), el contenido del buffer en RAM se pierde para siempre. No hay replicación ni WAL para buffers, lo que hace que este enfoque sea poco fiable para flujos de alta carga.

Cambiar al Motor Kafka resuelve el problema: los datos persisten en Kafka, y ClickHouse solo confirma el offset después del procesamiento exitoso.

Configurando el pipeline del Motor Kafka

Pila simple: Kafka + ZooKeeper + ClickHouse en Docker. Crea el tema test_topic_json con mensajes JSON:

{"message": "qq"}
{"message": "bb"}

En ClickHouse, el pipeline consta de tres componentes:

• Tabla del Motor Kafka para lectura:

drop table if exists test_kafka_json_parse;
create table test_kafka_json_parse
(
    message String
)
engine = Kafka
SETTINGS kafka_broker_list = 'kafka:29092'
    , kafka_topic_list = 'test_topic_json'
    , kafka_group_name = 'my-group4'
    , kafka_format = 'JSONEachRow'
    , kafka_max_block_size = 2097152;

• Vista materializada como consumidor:

drop view if exists test_kafka_json_parse_view;
create materialized view test_kafka_json_parse_view
TO default.message_json_parse
(
 `message` String
)
AS SELECT message FROM test_kafka_json_parse;

• MergeTree objetivo:

create table message_json_parse
(
    message String
)
engine = MergeTree
order by message;

Demostrando la atomicidad

Después de enviar mensajes, el offset del consumidor para my-group4 es 2 (End=2). Ahora cambia el tipo de columna en message_json_parse a UInt64 — el análisis de la cadena JSON "gg" fallará:

drop table if exists message_json_parse;
create table message_json_parse
(
    message UInt64
)
engine = MergeTree
order by message;

Envía {"message": "gg"}. El offset se queda en 2 (End=3), registro de ClickHouse: Cannot parse string 'gg' as UInt64. El offset se confirma solo después de la escritura exitosa en la vista materializada → MergeTree.

ClickHouse lee el mensaje en RAM pero no confirma el offset en caso de error. Los nuevos mensajes quedan bloqueados hasta que se corrija. Después de corregir el tipo de vuelta a String:

• Se inserta el mensaje "gg".
• El offset se actualiza a 3.
• No hay pérdida de datos incluso después de reiniciar ClickHouse.

Puntos clave

• Atomicidad: El offset de Kafka se confirma después de la inserción en la vista materializada, garantizando semántica exactly-once.
• Recuperación: Después de una falla/error, el procesamiento se reanuda desde el último offset confirmado sin duplicados ni saltos.
• Rendimiento: El procesamiento por lotes desde Kafka minimiza las pequeñas partes en MergeTree.
• Fiabilidad: Kafka almacena los datos de forma persistente; ClickHouse actúa como un consumidor sin estado.
• Limitaciones: Requiere infraestructura Kafka; configura kafka_skip_broken_messages=0 para un manejo estricto.

Recomendaciones de implementación

• Usa kafka_num_consumers >1 para paralelismo.
• Configura max_block_size=2MB+ para lotes.
• Monitorea offsets: SELECT * FROM system.kafka_consumers.
• Para JSON: JSONEachRow con análisis estricto.

— Editorial Team