> ## Documentation Index
> Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

> Uso del motor de tabla de Kafka

# Uso del motor de tabla de Kafka

export const Image = ({img, alt, size = "lg"}) => {
  const normalizedSize = ["sm", "md", "lg"].includes(size) ? size : "lg";
  return <div className={`ch-image-${normalizedSize}`}>
      <Frame>
        <img src={img} alt={alt} />
      </Frame>
    </div>;
};

El motor de tabla Kafka se puede utilizar para [**leer** datos de](#kafka-to-clickhouse) y [**escribir** datos en](#clickhouse-to-kafka) Apache Kafka y otros brokers compatibles con la API de Kafka (p. ej., Redpanda, Amazon MSK).

<div id="kafka-to-clickhouse">
  ### De Kafka a ClickHouse
</div>

<Note>
  Si usas ClickHouse Cloud, te recomendamos usar [ClickPipes](/docs/es/integrations/clickpipes/home) en su lugar. ClickPipes admite de forma nativa conexiones de red privadas, el escalado independiente de los recursos de ingestión y del cluster, y una monitorización exhaustiva para transmitir datos de Kafka a ClickHouse.
</Note>

Para usar el motor de tabla Kafka, deberías tener conocimientos generales sobre las [vistas materializadas de ClickHouse](/docs/es/concepts/features/materialized-views/cascading-materialized-views).

<div id="overview">
  #### Resumen
</div>

Inicialmente, nos centramos en el caso de uso más común: utilizar el motor de tabla de Kafka para insertar datos en ClickHouse desde Kafka.

El motor de tabla de Kafka permite a ClickHouse leer directamente de un Kafka topic. Aunque es útil para ver mensajes en un topic, este motor, por diseño, solo permite una única lectura; es decir, cuando se ejecuta una consulta sobre la tabla, consume datos de la cola y avanza el offset del consumer antes de devolver los resultados al cliente. En la práctica, los datos no pueden volver a leerse sin restablecer esos offsets.

Para conservar estos datos a partir de una lectura del motor de tabla, necesitamos un mecanismo para capturarlos e insertarlos en otra tabla. Las vistas materializadas basadas en triggers proporcionan esta funcionalidad de forma nativa. Una vista materializada inicia una lectura del motor de tabla y recibe lotes de documentos. La cláusula TO determina el destino de los datos, normalmente una tabla de la [familia MergeTree](/docs/es/reference/engines/table-engines/mergetree-family/index). Este proceso se visualiza a continuación:

<Image img="https://mintcdn.com/private-7c7dfe99/pIetLsS_hOGHqoPJ/images/integrations/data-ingestion/kafka/kafka_01.webp?fit=max&auto=format&n=pIetLsS_hOGHqoPJ&q=85&s=fd6990e133e6b46eb62f7057956fb5a3" size="lg" alt="Diagrama de arquitectura del motor de tabla de Kafka" style={{width: '80%'}} width="2048" height="852" data-path="images/integrations/data-ingestion/kafka/kafka_01.webp" />

<div id="steps">
  #### Pasos
</div>

<Steps>
  <Step title="Prepare" id="1-prepare">
    Si tiene datos cargados en un topic de destino, puede adaptar lo siguiente para usarlo en su conjunto de datos. Como alternativa, se proporciona un conjunto de datos de muestra de GitHub [aquí](https://datasets-documentation.s3.eu-west-3.amazonaws.com/kafka/github_all_columns.ndjson). Este conjunto de datos se usa en los ejemplos siguientes y utiliza un esquema reducido y un subconjunto de las filas (en concreto, nos limitamos a eventos de GitHub relacionados con el [repositorio de ClickHouse](https://github.com/ClickHouse/ClickHouse)), en comparación con el conjunto de datos completo disponible [aquí](https://ghe.clickhouse.tech/), por brevedad. Aun así, sigue siendo suficiente para que funcionen la mayoría de las consultas [publicadas con el conjunto de datos](https://ghe.clickhouse.tech/).
  </Step>

  <Step title="Configure ClickHouse" id="2-configure-clickhouse">
    Este paso es obligatorio si se está conectando a un Kafka seguro. Estos ajustes no pueden pasarse mediante comandos SQL DDL y deben configurarse en el archivo config.xml de ClickHouse. Suponemos que se está conectando a una instancia protegida con SASL. Este es el método más sencillo para interactuar con Confluent Cloud.

    ```xml theme={null}
    <clickhouse>
       <kafka>
           <sasl_username>username</sasl_username>
           <sasl_password>password</sasl_password>
           <security_protocol>sasl_ssl</security_protocol>
           <sasl_mechanisms>PLAIN</sasl_mechanisms>
       </kafka>
    </clickhouse>
    ```

    Coloque el fragmento anterior en un archivo nuevo dentro de su directorio conf.d/ o combínelo con los archivos de configuración existentes. Para conocer los ajustes que se pueden configurar, consulte [aquí](/docs/es/reference/engines/table-engines/integrations/kafka#configuration).

    También vamos a crear una base de datos llamada `KafkaEngine` para usar en este tutorial:

    ```sql theme={null}
    CREATE DATABASE KafkaEngine;
    ```

    Una vez creada la base de datos, deberá cambiar a ella:

    ```sql theme={null}
    USE KafkaEngine;
    ```
  </Step>

  <Step title="Cree la tabla de destino" id="3-create-the-destination-table">
    Prepare la tabla de destino. En el ejemplo siguiente usamos el esquema reducido de GitHub por brevedad. Tenga en cuenta que, aunque usamos un motor de tabla MergeTree, este ejemplo puede adaptarse fácilmente a cualquier miembro de la [familia MergeTree](/docs/es/reference/engines/table-engines/mergetree-family/index).

    ```sql theme={null}
    CREATE TABLE github
    (
        file_time DateTime,
        event_type Enum('CommitCommentEvent' = 1, 'CreateEvent' = 2, 'DeleteEvent' = 3, 'ForkEvent' = 4, 'GollumEvent' = 5, 'IssueCommentEvent' = 6, 'IssuesEvent' = 7, 'MemberEvent' = 8, 'PublicEvent' = 9, 'PullRequestEvent' = 10, 'PullRequestReviewCommentEvent' = 11, 'PushEvent' = 12, 'ReleaseEvent' = 13, 'SponsorshipEvent' = 14, 'WatchEvent' = 15, 'GistEvent' = 16, 'FollowEvent' = 17, 'DownloadEvent' = 18, 'PullRequestReviewEvent' = 19, 'ForkApplyEvent' = 20, 'Event' = 21, 'TeamAddEvent' = 22),
        actor_login LowCardinality(String),
        repo_name LowCardinality(String),
        created_at DateTime,
        updated_at DateTime,
        action Enum('none' = 0, 'created' = 1, 'added' = 2, 'edited' = 3, 'deleted' = 4, 'opened' = 5, 'closed' = 6, 'reopened' = 7, 'assigned' = 8, 'unassigned' = 9, 'labeled' = 10, 'unlabeled' = 11, 'review_requested' = 12, 'review_request_removed' = 13, 'synchronize' = 14, 'started' = 15, 'published' = 16, 'update' = 17, 'create' = 18, 'fork' = 19, 'merged' = 20),
        comment_id UInt64,
        path String,
        ref LowCardinality(String),
        ref_type Enum('none' = 0, 'branch' = 1, 'tag' = 2, 'repository' = 3, 'unknown' = 4),
        creator_user_login LowCardinality(String),
        number UInt32,
        title String,
        labels Array(LowCardinality(String)),
        state Enum('none' = 0, 'open' = 1, 'closed' = 2),
        assignee LowCardinality(String),
        assignees Array(LowCardinality(String)),
        closed_at DateTime,
        merged_at DateTime,
        merge_commit_sha String,
        requested_reviewers Array(LowCardinality(String)),
        merged_by LowCardinality(String),
        review_comments UInt32,
        member_login LowCardinality(String)
    ) ENGINE = MergeTree ORDER BY (event_type, repo_name, created_at)
    ```
  </Step>

  <Step title="Cree el topic y llénelo" id="4-create-and-populate-the-topic">
    A continuación, crearemos un topic. Hay varias herramientas que podemos usar para hacerlo. Si ejecutamos Kafka localmente en nuestra máquina o dentro de un contenedor Docker, [RPK](https://docs.redpanda.com/current/get-started/rpk-install/) funciona bien. Podemos crear un topic llamado `github` con 5 particiones ejecutando el siguiente comando:

    ```bash theme={null}
    rpk topic create -p 5 github --brokers <host>:<port>
    ```

    Si ejecutamos Kafka en Confluent Cloud, quizá prefiramos usar la [CLI de Confluent](https://docs.confluent.io/platform/current/tutorials/examples/clients/docs/kcat.html#produce-records):

    ```bash theme={null}
    confluent kafka topic create --if-not-exists github
    ```

    Ahora necesitamos poblar este topic con algunos datos, lo que haremos con [kcat](https://github.com/edenhill/kcat). Podemos ejecutar un comando similar al siguiente si estamos ejecutando Kafka localmente con la autenticación deshabilitada:

    ```bash theme={null}
    cat github_all_columns.ndjson |
    kcat -P \
      -b <host>:<port> \
      -t github
    ```

    O bien, lo siguiente si nuestro clúster de Kafka usa SASL para autenticarse:

    ```bash theme={null}
    cat github_all_columns.ndjson |
    kcat -P \
      -b <host>:<port> \
      -t github
      -X security.protocol=sasl_ssl \
      -X sasl.mechanisms=PLAIN \
      -X sasl.username=<username>  \
      -X sasl.password=<password> \
    ```

    El conjunto de datos contiene 200.000 filas, por lo que la ingesta debería completarse en apenas unos segundos. Si desea trabajar con un conjunto de datos más grande, consulte [la sección de conjuntos de datos grandes](https://github.com/ClickHouse/kafka-samples/tree/main/producer#large-datasets) del repositorio de GitHub [ClickHouse/kafka-samples](https://github.com/ClickHouse/kafka-samples).
  </Step>

  <Step title="Cree el motor de tabla de Kafka" id="5-create-the-kafka-table-engine">
    El siguiente ejemplo crea un motor de tabla con el mismo esquema que la tabla MergeTree. Esto no es estrictamente necesario, ya que puede tener un alias o columnas efímeras en la tabla de destino. Sin embargo, la configuración es importante; observe el uso de `JSONEachRow` como tipo de dato para consumir JSON desde un topic de Kafka. Los valores `github` y `clickhouse` representan el nombre del topic y los nombres del grupo de consumidores, respectivamente. Los topics pueden ser, de hecho, una lista de valores.

    ```sql theme={null}
    CREATE TABLE github_queue
    (
        file_time DateTime,
        event_type Enum('CommitCommentEvent' = 1, 'CreateEvent' = 2, 'DeleteEvent' = 3, 'ForkEvent' = 4, 'GollumEvent' = 5, 'IssueCommentEvent' = 6, 'IssuesEvent' = 7, 'MemberEvent' = 8, 'PublicEvent' = 9, 'PullRequestEvent' = 10, 'PullRequestReviewCommentEvent' = 11, 'PushEvent' = 12, 'ReleaseEvent' = 13, 'SponsorshipEvent' = 14, 'WatchEvent' = 15, 'GistEvent' = 16, 'FollowEvent' = 17, 'DownloadEvent' = 18, 'PullRequestReviewEvent' = 19, 'ForkApplyEvent' = 20, 'Event' = 21, 'TeamAddEvent' = 22),
        actor_login LowCardinality(String),
        repo_name LowCardinality(String),
        created_at DateTime,
        updated_at DateTime,
        action Enum('none' = 0, 'created' = 1, 'added' = 2, 'edited' = 3, 'deleted' = 4, 'opened' = 5, 'closed' = 6, 'reopened' = 7, 'assigned' = 8, 'unassigned' = 9, 'labeled' = 10, 'unlabeled' = 11, 'review_requested' = 12, 'review_request_removed' = 13, 'synchronize' = 14, 'started' = 15, 'published' = 16, 'update' = 17, 'create' = 18, 'fork' = 19, 'merged' = 20),
        comment_id UInt64,
        path String,
        ref LowCardinality(String),
        ref_type Enum('none' = 0, 'branch' = 1, 'tag' = 2, 'repository' = 3, 'unknown' = 4),
        creator_user_login LowCardinality(String),
        number UInt32,
        title String,
        labels Array(LowCardinality(String)),
        state Enum('none' = 0, 'open' = 1, 'closed' = 2),
        assignee LowCardinality(String),
        assignees Array(LowCardinality(String)),
        closed_at DateTime,
        merged_at DateTime,
        merge_commit_sha String,
        requested_reviewers Array(LowCardinality(String)),
        merged_by LowCardinality(String),
        review_comments UInt32,
        member_login LowCardinality(String)
    )
       ENGINE = Kafka('kafka_host:9092', 'github', 'clickhouse',
                'JSONEachRow') SETTINGS kafka_thread_per_consumer = 0, kafka_num_consumers = 1;
    ```

    A continuación, analizamos la configuración del motor y el ajuste del rendimiento. En este punto, un `select` sencillo en la tabla `github_queue` debería leer algunas filas. Tenga en cuenta que esto hará avanzar los offsets del consumidor, lo que impedirá volver a leer estas filas sin un [reinicio](#common-operations). Tenga en cuenta el límite y el parámetro obligatorio `stream_like_engine_allow_direct_select.`
  </Step>

  <Step title="Cree la vista materializada" id="6-create-the-materialized-view">
    La vista materializada conectará las dos tablas creadas previamente, leyendo datos del motor de tabla Kafka e insertándolos en la tabla MergeTree de destino. Podemos realizar varias transformaciones de datos. Haremos una lectura e inserción sencillas. El uso de \* supone que los nombres de las columnas son idénticos (distingue entre mayúsculas y minúsculas).

    ```sql theme={null}
    CREATE MATERIALIZED VIEW github_mv TO github AS
    SELECT *
    FROM github_queue;
    ```

    En el momento de su creación, la vista materializada se conecta al motor Kafka y comienza a leer, insertando filas en la tabla de destino. Este proceso continuará indefinidamente y consumirá los mensajes posteriores insertados en Kafka. No dude en volver a ejecutar el script de inserción para insertar más mensajes en Kafka.
  </Step>

  <Step title="Confirme que se hayan insertado filas" id="7-confirm-rows-have-been-inserted">
    Compruebe que hay datos en la tabla de destino:

    ```sql theme={null}
    SELECT count() FROM github;
    ```

    Debería ver 200.000 filas:

    ```response theme={null}
    ┌─count()─┐
    │  200000 │
    └─────────┘
    ```
  </Step>
</Steps>

<div id="common-operations">
  #### Operaciones comunes
</div>

<div id="stopping--restarting-message-consumption">
  ##### Detener & reiniciar el consumo de mensajes
</div>

Para detener el consumo de mensajes, puede desvincular la tabla del motor Kafka:

```sql theme={null}
DETACH TABLE github_queue;
```

Esto no afectará los offsets del grupo de consumidores. Para reanudar el consumo y continuar desde el offset anterior, vuelva a adjuntar la tabla.

```sql theme={null}
ATTACH TABLE github_queue;
```

<div id="adding-kafka-metadata">
  ##### Añadir metadatos de Kafka
</div>

Puede resultar útil hacer un seguimiento de los metadatos de los mensajes originales de Kafka una vez ingeridos en ClickHouse. Por ejemplo, puede que queramos saber cuánto hemos consumido de un topic o una partición concretos. Para ello, el motor de tabla Kafka expone varias [columnas virtuales](/docs/es/reference/engines/table-engines/index#table_engines-virtual_columns). Estas pueden persistirse como columnas en nuestra tabla de destino modificando nuestro esquema y la instrucción SELECT de la vista materializada.

Primero, realizamos la operación de detención descrita anteriormente antes de añadir columnas a nuestra tabla de destino.

```sql theme={null}
DETACH TABLE github_queue;
```

A continuación, añadimos columnas informativas para identificar el topic de origen y la partición de la que procede la fila.

```sql theme={null}
ALTER TABLE github
   ADD COLUMN topic String,
   ADD COLUMN partition UInt64;
```

A continuación, debemos asegurarnos de que las columnas virtuales estén mapeadas según corresponda.
Las columnas virtuales llevan el prefijo `_`.
Puede consultar una lista completa de las columnas virtuales [aquí](/docs/es/reference/engines/table-engines/integrations/kafka#virtual-columns).

Para actualizar nuestra tabla con las columnas virtuales, tendremos que eliminar la vista materializada, volver a adjuntar la tabla con motor Kafka y volver a crear la vista materializada.

```sql theme={null}
DROP VIEW github_mv;
```

```sql theme={null}
ATTACH TABLE github_queue;
```

```sql theme={null}
CREATE MATERIALIZED VIEW github_mv TO github AS
SELECT *, _topic AS topic, _partition as partition
FROM github_queue;
```

Las filas consumidas recientemente deberían incluir los metadatos.

```sql theme={null}
SELECT actor_login, event_type, created_at, topic, partition
FROM github
LIMIT 10;
```

El resultado es así:

| actor\_login  | event\_type        | created\_at         | topic  | partition |
| :------------ | :----------------- | :------------------ | :----- | :-------- |
| IgorMinar     | CommitCommentEvent | 2011-02-12 02:22:00 | github | 0         |
| queeup        | CommitCommentEvent | 2011-02-12 02:23:23 | github | 0         |
| IgorMinar     | CommitCommentEvent | 2011-02-12 02:23:24 | github | 0         |
| IgorMinar     | CommitCommentEvent | 2011-02-12 02:24:50 | github | 0         |
| IgorMinar     | CommitCommentEvent | 2011-02-12 02:25:20 | github | 0         |
| dapi          | CommitCommentEvent | 2011-02-12 06:18:36 | github | 0         |
| sourcerebels  | CommitCommentEvent | 2011-02-12 06:34:10 | github | 0         |
| jamierumbelow | CommitCommentEvent | 2011-02-12 12:21:40 | github | 0         |
| jpn           | CommitCommentEvent | 2011-02-12 12:24:31 | github | 0         |
| Oxonium       | CommitCommentEvent | 2011-02-12 12:31:28 | github | 0         |

<div id="modify-kafka-engine-settings">
  ##### Modificar la configuración del motor Kafka
</div>

Recomendamos eliminar la tabla con motor Kafka y volver a crearla con la nueva configuración. No es necesario modificar la vista materializada durante este proceso; el consumo de mensajes se reanudará una vez recreada la tabla con motor Kafka.

<div id="debugging-issues">
  ##### Depuración de incidencias
</div>

Errores como los problemas de autenticación no se notifican en las respuestas al DDL del motor Kafka. Para diagnosticar estos problemas, recomendamos usar el archivo de registro principal de ClickHouse, clickhouse-server.err.log. También se puede habilitar un registro de trazas más detallado para la biblioteca cliente de Kafka subyacente, [librdkafka](https://github.com/edenhill/librdkafka), mediante la configuración.

```xml theme={null}
<kafka>
   <debug>all</debug>
</kafka>
```

<div id="handling-malformed-messages">
  ##### Manejo de mensajes malformados
</div>

Kafka se utiliza a menudo como un "vertedero" de datos. Esto hace que los topics contengan formatos de mensaje mixtos y nombres de campo incoherentes. Evite esta situación y aproveche funcionalidades de Kafka como Kafka Streams o ksqlDB para garantizar que los mensajes estén bien formados y sean coherentes antes de insertarlos en Kafka. Si estas opciones no son viables, ClickHouse ofrece algunas funciones que pueden ayudar.

* Trate los campos del mensaje como cadenas. Se pueden usar funciones en la sentencia de la vista materializada para realizar la limpieza y la conversión de tipos si es necesario. Esto no debería considerarse una solución para producción, pero puede ser útil para una ingestión puntual.
* Si está consumiendo JSON desde un topic con el format JSONEachRow, use la setting [`input_format_skip_unknown_fields`](/docs/es/reference/settings/formats#input_format_skip_unknown_fields). Al escribir datos, ClickHouse, de forma predeterminada, throws an exception si los datos de entrada contienen columnas que no existen en la tabla de destino. Sin embargo, si esta opción está enabled, esas columnas sobrantes se ignorarán. De nuevo, esto no es una solución apta para producción y podría confundir a otros usuarios.
* Considere la setting `kafka_skip_broken_messages`. Esto requiere que el usuario especifique el nivel de tolerancia por bloque para mensajes malformados, en el contexto de kafka\_max\_block\_size. Si se supera esta tolerancia (medida en número absoluto de mensajes), se restablecerá el comportamiento habitual de excepción y se omitirán otros mensajes.

<div id="delivery-semantics-and-challenges-with-duplicates">
  ##### Semántica de entrega y problemas con los duplicados
</div>

El motor de tabla Kafka tiene semántica de al menos una vez. Es posible que haya duplicados en varias circunstancias poco frecuentes ya conocidas. Por ejemplo, los mensajes pueden leerse de Kafka e insertarse correctamente en ClickHouse. Antes de que pueda confirmarse el nuevo offset, se pierde la conexión con Kafka. En esta situación, es necesario reintentar el bloque. El bloque puede [deduplicarse ](/docs/es/reference/engines/table-engines/mergetree-family/replication)usando una tabla distribuida o ReplicatedMergeTree como tabla de destino. Aunque esto reduce la probabilidad de filas duplicadas, depende de que los bloques sean idénticos. Eventos como un reequilibrio de Kafka pueden invalidar este supuesto y provocar duplicados en casos poco frecuentes.

<div id="quorum-based-inserts">
  ##### Inserciones basadas en quorum
</div>

Puede que necesite [inserciones basadas en quorum](/docs/es/reference/settings/session-settings#insert_quorum) en los casos en los que se requieran mayores garantías de entrega en ClickHouse. Esto no se puede configurar en la vista materializada ni en la tabla de destino. Sin embargo, sí se puede configurar para perfiles de usuario, por ejemplo.

```xml theme={null}
<profiles>
  <default>
    <insert_quorum>2</insert_quorum>
  </default>
</profiles>
```

<div id="clickhouse-to-kafka">
  ### De ClickHouse a Kafka
</div>

Aunque es un caso de uso menos habitual, los datos de ClickHouse también pueden persistirse en Kafka. Por ejemplo, insertaremos filas manualmente en un motor de tabla Kafka. Estos datos los leerá el mismo motor Kafka, cuya vista materializada colocará los datos en una tabla MergeTree. Por último, demostramos cómo aplicar vistas materializadas a las inserciones en Kafka para leer tablas a partir de tablas de origen existentes.

<div id="steps-1">
  #### Pasos
</div>

Nuestro objetivo inicial se ilustra mejor de la siguiente manera:

<Image img="https://mintcdn.com/private-7c7dfe99/pIetLsS_hOGHqoPJ/images/integrations/data-ingestion/kafka/kafka_02.webp?fit=max&auto=format&n=pIetLsS_hOGHqoPJ&q=85&s=9093dc39ca712ae891de358b107bcfad" size="lg" alt="Diagrama del motor de tabla de Kafka con inserciones" width="2048" height="852" data-path="images/integrations/data-ingestion/kafka/kafka_02.webp" />

Suponemos que ya ha creado las tablas y vistas en los pasos de [Kafka a ClickHouse](#kafka-to-clickhouse) y que el topic ya se ha consumido por completo.

<Steps>
  <Step title="Insertar filas directamente" id="1-inserting-rows-directly">
    Primero, confirme el número de filas de la tabla de destino.

    ```sql theme={null}
    SELECT count() FROM github;
    ```

    Debería haber 200.000 filas:

    ```response theme={null}
    ┌─count()─┐
    │  200000 │
    └─────────┘
    ```

    Ahora inserte filas desde la tabla de destino de GitHub de nuevo en el motor de tabla Kafka github\_queue. Observe cómo utilizamos el formato JSONEachRow y aplicamos LIMIT a la consulta SELECT para devolver 100 filas.

    ```sql theme={null}
    INSERT INTO github_queue SELECT * FROM github LIMIT 100 FORMAT JSONEachRow
    ```

    Vuelva a contar las filas en GitHub para confirmar que han aumentado en 100. Como se muestra en el diagrama anterior, las filas se han insertado en Kafka mediante el motor de tabla Kafka antes de volver a ser leídas por el mismo motor e insertadas en la tabla de destino de GitHub por nuestra vista materializada.

    ```sql theme={null}
    SELECT count() FROM github;
    ```

    Deberías ver 100 filas más:

    ```response theme={null}
    ┌─count()─┐
    │  200100 │
    └─────────┘
    ```
  </Step>

  <Step title="Uso de vistas materializadas" id="2-using-materialized-views">
    Podemos utilizar vistas materializadas para enviar mensajes a un motor Kafka (y a un topic) cuando se insertan documentos en una tabla. Cuando se insertan filas en la tabla GitHub, se activa una vista materializada, lo que hace que las filas se vuelvan a insertar en un motor Kafka y en un topic nuevo. De nuevo, se ilustra mejor de la siguiente manera:

    <Image img="https://mintcdn.com/private-7c7dfe99/pIetLsS_hOGHqoPJ/images/integrations/data-ingestion/kafka/kafka_03.webp?fit=max&auto=format&n=pIetLsS_hOGHqoPJ&q=85&s=ead2a8c956700436a1fd15bb803ef49b" size="lg" alt="Diagrama del motor de tabla Kafka con vistas materializadas" width="2048" height="870" data-path="images/integrations/data-ingestion/kafka/kafka_03.webp" />

    Cree un tema de Kafka nuevo `github_out` o uno equivalente. Asegúrese de que un motor de tabla de Kafka `github_out_queue` apunte a este tema.

    ```sql theme={null}
    CREATE TABLE github_out_queue
    (
        file_time DateTime,
        event_type Enum('CommitCommentEvent' = 1, 'CreateEvent' = 2, 'DeleteEvent' = 3, 'ForkEvent' = 4, 'GollumEvent' = 5, 'IssueCommentEvent' = 6, 'IssuesEvent' = 7, 'MemberEvent' = 8, 'PublicEvent' = 9, 'PullRequestEvent' = 10, 'PullRequestReviewCommentEvent' = 11, 'PushEvent' = 12, 'ReleaseEvent' = 13, 'SponsorshipEvent' = 14, 'WatchEvent' = 15, 'GistEvent' = 16, 'FollowEvent' = 17, 'DownloadEvent' = 18, 'PullRequestReviewEvent' = 19, 'ForkApplyEvent' = 20, 'Event' = 21, 'TeamAddEvent' = 22),
        actor_login LowCardinality(String),
        repo_name LowCardinality(String),
        created_at DateTime,
        updated_at DateTime,
        action Enum('none' = 0, 'created' = 1, 'added' = 2, 'edited' = 3, 'deleted' = 4, 'opened' = 5, 'closed' = 6, 'reopened' = 7, 'assigned' = 8, 'unassigned' = 9, 'labeled' = 10, 'unlabeled' = 11, 'review_requested' = 12, 'review_request_removed' = 13, 'synchronize' = 14, 'started' = 15, 'published' = 16, 'update' = 17, 'create' = 18, 'fork' = 19, 'merged' = 20),
        comment_id UInt64,
        path String,
        ref LowCardinality(String),
        ref_type Enum('none' = 0, 'branch' = 1, 'tag' = 2, 'repository' = 3, 'unknown' = 4),
        creator_user_login LowCardinality(String),
        number UInt32,
        title String,
        labels Array(LowCardinality(String)),
        state Enum('none' = 0, 'open' = 1, 'closed' = 2),
        assignee LowCardinality(String),
        assignees Array(LowCardinality(String)),
        closed_at DateTime,
        merged_at DateTime,
        merge_commit_sha String,
        requested_reviewers Array(LowCardinality(String)),
        merged_by LowCardinality(String),
        review_comments UInt32,
        member_login LowCardinality(String)
    )
       ENGINE = Kafka('host:port', 'github_out', 'clickhouse_out',
                'JSONEachRow') SETTINGS kafka_thread_per_consumer = 0, kafka_num_consumers = 1;
    ```

    Ahora cree una nueva vista materializada `github_out_mv` para que apunte a la tabla de GitHub e inserte filas en el engine anterior cuando se active. Como resultado, las nuevas filas añadidas a la tabla de GitHub se enviarán a nuestro nuevo topic de Kafka.

    ```sql theme={null}
    CREATE MATERIALIZED VIEW github_out_mv TO github_out_queue AS
    SELECT file_time, event_type, actor_login, repo_name,
           created_at, updated_at, action, comment_id, path,
           ref, ref_type, creator_user_login, number, title,
           labels, state, assignee, assignees, closed_at, merged_at,
           merge_commit_sha, requested_reviewers, merged_by,
           review_comments, member_login
    FROM github
    FORMAT JsonEachRow;
    ```

    Si insertas datos en el topic original de github, creado como parte de [Kafka to ClickHouse](#kafka-to-clickhouse), los documentos aparecerán mágicamente en el topic "github\_clickhouse". Confírmalo con las herramientas nativas de Kafka. Por ejemplo, a continuación, insertamos 100 filas en el topic github con [kcat](https://github.com/edenhill/kcat) para un topic alojado en Confluent Cloud:

    ```sql theme={null}
    head -n 10 github_all_columns.ndjson |
    kcat -P \
      -b <host>:<port> \
      -t github
      -X security.protocol=sasl_ssl \
      -X sasl.mechanisms=PLAIN \
      -X sasl.username=<username> \
      -X sasl.password=<password>
    ```

    La lectura del topic `github_out` debería confirmar la entrega de los mensajes.

    ```sql theme={null}
    kcat -C \
      -b <host>:<port> \
      -t github_out \
      -X security.protocol=sasl_ssl \
      -X sasl.mechanisms=PLAIN \
      -X sasl.username=<username> \
      -X sasl.password=<password> \
      -e -q |
    wc -l
    ```

    Aunque es un ejemplo elaborado, ilustra el potencial de las vistas materializadas cuando se usan en combinación con el motor Kafka.
  </Step>
</Steps>

<div id="clusters-and-performance">
  ### Clústeres y rendimiento
</div>

<div id="working-with-clickhouse-clusters">
  #### Trabajo con clústeres de ClickHouse
</div>

Mediante grupos de consumidores de Kafka, varias instancias de ClickHouse pueden leer potencialmente del mismo topic. A cada consumidor se le asigna una partición del topic en una relación 1:1. Al escalar el consumo de ClickHouse con el motor de tabla de Kafka, tenga en cuenta que el número total de consumidores dentro de un clúster no puede superar el número de particiones del topic. Por lo tanto, asegúrese de configurar de antemano un particionado adecuado para el topic.

Se pueden configurar varias instancias de ClickHouse para leer de un topic con el mismo ID de grupo de consumidores, especificado durante la creación del motor de tabla de Kafka. De este modo, cada instancia leerá de una o más particiones e insertará segmentos en su tabla de destino local. A su vez, las tablas de destino pueden configurarse para usar ReplicatedMergeTree y gestionar la duplicación de datos. Este enfoque permite escalar las lecturas de Kafka junto con el clúster de ClickHouse, siempre que haya suficientes particiones de Kafka.

<Image img="https://mintcdn.com/private-7c7dfe99/pIetLsS_hOGHqoPJ/images/integrations/data-ingestion/kafka/kafka_04.webp?fit=max&auto=format&n=pIetLsS_hOGHqoPJ&q=85&s=fc14aa743c7b5cda9b7d722e18bc59f0" size="lg" alt="Diagrama del motor de tabla Kafka con clústeres de ClickHouse" width="2048" height="1094" data-path="images/integrations/data-ingestion/kafka/kafka_04.webp" />

<div id="tuning-performance">
  #### Ajuste del rendimiento
</div>

Tenga en cuenta lo siguiente si busca aumentar el rendimiento de una tabla con el motor Kafka:

* El rendimiento variará según el tamaño de los mensajes, el formato y los tipos de las tablas de destino. Un rendimiento de 100k filas/s en un único motor de tabla debe considerarse alcanzable. De forma predeterminada, los mensajes se leen en bloques, controlados por el parámetro kafka\_max\_block\_size. Por defecto, este se establece en [max\_insert\_block\_size](/docs/es/reference/settings/session-settings#max_insert_block_size), cuyo valor predeterminado es 1,048,576. Salvo que los mensajes sean extremadamente grandes, casi siempre conviene aumentarlo. No es raro usar valores entre 500k y 1M. Pruebe y evalúe el efecto sobre el rendimiento.
* El número de consumidores de un motor de tabla puede aumentarse con kafka\_num\_consumers. Sin embargo, de forma predeterminada, las inserciones se serializan en un único hilo, a menos que kafka\_thread\_per\_consumer se cambie desde su valor predeterminado de 1. Establézcalo en 1 para garantizar que los flush se realicen en paralelo. Tenga en cuenta que crear una tabla con el motor Kafka con N consumidores (y kafka\_thread\_per\_consumer=1) es lógicamente equivalente a crear N motores Kafka, cada uno con una vista materializada y kafka\_thread\_per\_consumer=0.
* Aumentar el número de consumidores no sale gratis. Cada consumidor mantiene sus propios búferes e hilos, lo que incrementa la sobrecarga del servidor. Tenga presente esta sobrecarga y, si es posible, escale primero de forma lineal en todo el clúster.
* Si el rendimiento de los mensajes de Kafka es variable y los retrasos son aceptables, considere aumentar stream\_flush\_interval\_ms para asegurarse de que se escriban bloques más grandes.
* [background\_message\_broker\_schedule\_pool\_size](/docs/es/reference/settings/server-settings/settings#background_message_broker_schedule_pool_size) establece el número de hilos que realizan tareas en segundo plano. Estos hilos se usan para el streaming de Kafka. Esta configuración se aplica al iniciar el servidor ClickHouse y no puede cambiarse en una sesión de usuario; su valor predeterminado es 16. Si ve timeouts en los logs, puede ser conveniente aumentar este valor.
* Para la comunicación con Kafka se utiliza la biblioteca librdkafka, que a su vez crea hilos. Por tanto, un gran número de tablas Kafka o de consumidores puede dar lugar a una gran cantidad de cambios de contexto. Distribuya esta carga en todo el clúster, replicando solo las tablas de destino si es posible, o considere usar un motor de tabla para leer de varios topics; se admite una lista de valores. Varias vistas materializadas pueden leer de una sola tabla, cada una filtrando los datos de un topic concreto.

Cualquier cambio en la configuración debe probarse. Recomendamos supervisar el consumer lag de Kafka para asegurarse de que el escalado sea el adecuado.

<div id="additional-settings">
  #### Ajustes adicionales
</div>

Además de los ajustes mencionados anteriormente, los siguientes pueden resultar de interés:

* [Kafka\_max\_wait\_ms](/docs/es/reference/settings/session-settings#kafka_max_wait_ms) - El tiempo de espera, en milisegundos, para leer mensajes de Kafka antes de reintentar. Se establece a nivel de perfil de usuario y su valor predeterminado es 5000.

[Todos los ajustes ](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md)de la librdkafka subyacente también pueden colocarse en los archivos de configuración de ClickHouse dentro de un elemento *kafka*; los nombres de los ajustes deben ser elementos XML con los puntos sustituidos por guiones bajos; p. ej.

```xml theme={null}
<clickhouse>
   <kafka>
       <enable_ssl_certificate_verification>false</enable_ssl_certificate_verification>
   </kafka>
</clickhouse>
```

Estas son configuraciones avanzadas y le sugerimos consultar la documentación de Kafka para obtener una explicación más detallada.