`). La configuración global se aplica primero y, a continuación, se aplica la configuración a nivel de topic (si existe).
```xml theme={null}
cgrp
3000
logs
4000
smallest
logs
100000
stats
50000
logs
250
stats
400
```
Para obtener una lista de las posibles opciones de configuración, consulte la [referencia de configuración de librdkafka](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). Use el guion bajo (`_`) en lugar del punto en la configuración de ClickHouse. Por ejemplo, `check.crcs=true` se convierte en `true`.
### Soporte para Kerberos
Para trabajar con Kafka con compatibilidad con Kerberos, agregue el elemento secundario `security_protocol` con el valor `sasl_plaintext`. Basta con que el ticket de concesión de tickets de Kerberos se obtenga y el sistema operativo lo almacene en caché.
ClickHouse puede mantener las credenciales de Kerberos mediante un archivo keytab. Considere los elementos secundarios `sasl_kerberos_service_name`, `sasl_kerberos_keytab` y `sasl_kerberos_principal`.
Ejemplo:
```xml theme={null}
SASL_PLAINTEXT
/home/kafkauser/kafkauser.keytab
kafkauser/kafkahost@EXAMPLE.COM
```
## Columnas virtuales
* `_topic` — Topic de Kafka. Tipo de dato: `LowCardinality(String)`.
* `_key` — Clave del mensaje. Tipo de dato: `String`.
* `_offset` — Offset del mensaje. Tipo de dato: `UInt64`.
* `_timestamp` — Marca temporal del mensaje. Tipo de dato: `Nullable(DateTime)`.
* `_timestamp_ms` — Marca temporal del mensaje en milisegundos. Tipo de dato: `Nullable(DateTime64(3))`.
* `_partition` — Partición del topic de Kafka. Tipo de dato: `UInt64`.
* `_headers.name` — Array de claves de los encabezados del mensaje. Tipo de dato: `Array(String)`.
* `_headers.value` — Array de valores de los encabezados del mensaje. Tipo de dato: `Array(String)`.
Columnas virtuales adicionales cuando `kafka_handle_error_mode='stream'`:
* `_raw_message` - Mensaje sin procesar que no pudo analizarse correctamente. Tipo de dato: `String`.
* `_error` - Mensaje de excepción generado durante un error de análisis. Tipo de dato: `String`.
Nota: las columnas virtuales `_raw_message` y `_error` se rellenan solo si se produce una excepción durante el análisis; siempre están vacías cuando el mensaje se analiza correctamente.
## Correspondencia de columnas con los metadatos de mensajes de Kafka
Al producir mensajes con `INSERT INTO`, el motor Kafka siempre usa una columna llamada `_key` (de tipo `String`) como clave del mensaje de Kafka y una columna llamada `_timestamp` (de tipo `DateTime`) como marca temporal del mensaje de Kafka, si esas columnas existen en la tabla. De forma predeterminada, estas columnas también aparecen en el payload del mensaje junto con las demás columnas.
Con `kafka_map_virtual_columns_on_write = 1`, el comportamiento cambia:
* `_key` (tipo `String`) — se asigna a la clave del mensaje de Kafka.
* `_timestamp` (tipo `DateTime`) — se asigna a la marca temporal del mensaje de Kafka.
* `_headers.name` (tipo `Array(String)`) y `_headers.value` (tipo `Array(String)`) — se asignan a los encabezados del mensaje de Kafka. Cada par `(_headers.name[i], _headers.value[i])` se convierte en un encabezado de Kafka. Como `_headers.name` y `_headers.value` comparten el prefijo Nested `_headers`, ClickHouse requiere que ambos arrays tengan el mismo tamaño en cada fila.
Las columnas con estos nombres **se excluyen del payload del mensaje** solo si sus tipos coinciden con los indicados anteriormente; de lo contrario, permanecen en el payload, por lo que los esquemas que reutilizan esos nombres para datos no relacionados siguen funcionando.
Ejemplo:
```sql theme={null}
CREATE TABLE kafka_out
(
event_json String,
`_key` String,
`_timestamp` DateTime,
`_headers.name` Array(String),
`_headers.value` Array(String)
)
ENGINE = Kafka
SETTINGS
kafka_broker_list = 'broker:9092',
kafka_topic_list = 'events',
kafka_group_name = 'events-producer',
kafka_format = 'JSONEachRow',
kafka_map_virtual_columns_on_write = 1;
INSERT INTO kafka_out VALUES
('{"a":1}', 'session-42', now(), ['source', 'trace_id'], ['api', 'abc-123']);
```
El mensaje de Kafka generado contiene el payload `{"event_json":"{\"a\":1}"}`, la clave `session-42`, la marca temporal actual y dos encabezados `source=api` y `trace_id=abc-123`.
## Compatibilidad con formatos de datos
El motor Kafka admite todos los [formatos](/es/reference/formats/index) compatibles con ClickHouse.
El número de filas en un mensaje de Kafka depende de si el formato está basado en filas o en bloques:
* En los formatos basados en filas, el número de filas en un mensaje de Kafka puede controlarse mediante la opción `kafka_max_rows_per_message`.
* En los formatos basados en bloques, no se puede dividir un bloque en partes más pequeñas, pero el número de filas de un bloque puede controlarse mediante la opción general [max\_block\_size](/es/reference/settings/session-settings#max_block_size).
## Motor para almacenar offsets confirmados en ClickHouse Keeper
Si `allow_experimental_kafka_offsets_storage_in_keeper` está habilitado, se pueden especificar dos ajustes adicionales para el motor de tabla de Kafka:
* `kafka_keeper_path` especifica la ruta de la tabla en ClickHouse Keeper
* `kafka_replica_name` especifica el nombre de la réplica en ClickHouse Keeper
Deben especificarse ambos ajustes o ninguno de los dos. Cuando se especifican ambos, se utilizará un nuevo motor Kafka experimental. El nuevo motor no depende de almacenar los offsets confirmados en Kafka, sino que los almacena en ClickHouse Keeper. Sigue intentando confirmar los offsets en Kafka, pero solo depende de ellos cuando se crea la tabla. En cualquier otro caso (si la tabla se reinicia o se recupera después de algún error), se usarán los offsets almacenados en ClickHouse Keeper para seguir consumiendo mensajes. Además del offset confirmado, también almacena cuántos mensajes se consumieron en el último lote, de modo que, si la inserción falla, se volverá a consumir la misma cantidad de mensajes, lo que permite la deduplicación si es necesario.
Ejemplo:
```sql theme={null}
CREATE TABLE experimental_kafka (key UInt64, value UInt64)
ENGINE = Kafka('localhost:19092', 'my-topic', 'my-consumer', 'JSONEachRow')
SETTINGS
kafka_keeper_path = '/clickhouse/{database}/{uuid}',
kafka_replica_name = '{replica}'
SETTINGS allow_experimental_kafka_offsets_storage_in_keeper=1;
```
### Limitaciones conocidas
Como el nuevo motor es experimental, todavía no está listo para producción. La implementación presenta algunas limitaciones conocidas:
* Eliminar y volver a crear rápidamente la tabla, o especificar la misma ruta de ClickHouse Keeper para distintos motores, puede causar problemas. Como práctica recomendada, puede usar `{uuid}` en `kafka_keeper_path` para evitar conflictos entre rutas.
* Para garantizar lecturas repetibles, los mensajes no pueden consumirse desde varias particiones en un solo hilo. Por otro lado, es necesario hacer *poll* a los consumidores de Kafka con regularidad para mantenerlos activos. Como resultado de estos dos requisitos, decidimos permitir la creación de varios consumidores solo si `kafka_thread_per_consumer` está habilitado; de lo contrario, resulta demasiado complicado evitar problemas relacionados con hacer *poll* a los consumidores regularmente.
**Véase también**
* [Columnas virtuales](/es/reference/engines/table-engines/index#table_engines-virtual_columns)
* [background\_message\_broker\_schedule\_pool\_size](/es/reference/settings/server-settings/settings#background_message_broker_schedule_pool_size)
* [system.kafka\_consumers](/es/reference/system-tables/kafka_consumers)