> ## 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 conector sink JDBC con Kafka Connect y ClickHouse

# JDBC Connector

<Note>
  Este conector solo debe usarse si los datos son simples y constan de tipos de datos primitivos, p. ej., int. Los tipos específicos de ClickHouse, como Map, no son compatibles.
</Note>

Para nuestros ejemplos, utilizamos la distribución de Kafka Connect de Confluent.

A continuación describimos una instalación sencilla, que extrae mensajes de un único topic de Kafka e inserta filas en una tabla de ClickHouse. Recomendamos Confluent Cloud, que ofrece un generoso nivel gratuito para quienes no disponen de un entorno de Kafka.

Tenga en cuenta que el JDBC Connector requiere un esquema (no puede usar JSON o CSV sin formato con el JDBC Connector). Aunque el esquema puede codificarse en cada mensaje, se [recomienda encarecidamente usar Confluent Schema Registry](https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained/#json-schemas) para evitar la sobrecarga asociada. El script de inserción proporcionado infiere automáticamente un esquema a partir de los mensajes y lo inserta en el registro; por lo tanto, este script también puede reutilizarse para otros conjuntos de datos. Se asume que las claves de Kafka son de tipo String. Puede encontrar más detalles sobre los esquemas de Kafka [aquí](https://docs.confluent.io/platform/current/schema-registry/index.html).

<div id="license">
  ### Licencia
</div>

El JDBC Connector se distribuye bajo la [Confluent Community License](https://www.confluent.io/confluent-community-license)

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

<div id="gather-your-connection-details">
  #### Reúne los datos de conexión
</div>

Para conectarse a ClickHouse con HTTP(S), necesita esta información:

| Parámetro(s)              | Descripción                                                                                                                       |
| ------------------------- | --------------------------------------------------------------------------------------------------------------------------------- |
| `HOST` and `PORT`         | Normalmente, el puerto es 8443 cuando se usa TLS o 8123 cuando no se usa TLS.                                                     |
| `DATABASE NAME`           | De forma predeterminada, existe una base de datos llamada `default`; use el nombre de la base de datos a la que desea conectarse. |
| `USERNAME` and `PASSWORD` | De forma predeterminada, el nombre de usuario es `default`. Use el nombre de usuario adecuado para su caso de uso.                |

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

<div className="ch-image-md">
  <Frame>
    <img src="https://mintcdn.com/private-7c7dfe99/CFFsa2agBPbviR4r/images/_snippets/cloud-connect-button.webp?fit=max&auto=format&n=CFFsa2agBPbviR4r&q=85&s=ec0a298a33ca841e947fa5e8bae47362" alt="Botón Connect del servicio de ClickHouse Cloud" width="998" height="932" data-path="images/_snippets/cloud-connect-button.webp" />
  </Frame>
</div>

Elija **HTTPS**. Los detalles de conexión se muestran en un comando `curl` de ejemplo.

<div className="ch-image-md">
  <Frame>
    <img src="https://mintcdn.com/private-7c7dfe99/CFFsa2agBPbviR4r/images/_snippets/connection-details-https.webp?fit=max&auto=format&n=CFFsa2agBPbviR4r&q=85&s=cb0fbd98aa2b5b7ca484c9f53395ee07" alt="Detalles de conexión HTTPS de ClickHouse Cloud" width="1320" height="1184" data-path="images/_snippets/connection-details-https.webp" />
  </Frame>
</div>

Si usa ClickHouse autogestionado, los detalles de conexión los establece su administrador de ClickHouse.

<Steps>
  <Step title="Instalar Kafka Connect y el conector" id="1-install-kafka-connect-and-connector">
    Asumimos que has descargado el paquete de Confluent y lo has instalado localmente. Sigue las instrucciones de instalación para instalar el conector, tal como se documenta [aquí](https://docs.confluent.io/kafka-connect-jdbc/current/#install-the-jdbc-connector).

    Si usas el método de instalación `confluent-hub`, tus archivos de configuración locales se actualizarán.

    Para enviar datos a ClickHouse desde Kafka, usamos el componente sink del conector.
  </Step>

  <Step title="Descarga e instala el driver JDBC" id="2-download-and-install-the-jdbc-driver">
    Descarga e instala el ClickHouse JDBC driver `clickhouse-jdbc-<version>-shaded.jar` desde [aquí](https://github.com/ClickHouse/clickhouse-java/releases). Instálalo en Kafka Connect siguiendo las instrucciones [aquí](https://docs.confluent.io/kafka-connect-jdbc/current/#installing-jdbc-drivers). Es posible que otros drivers funcionen, pero no se han probado.

    <Note>
      Problema común: la documentación sugiere copiar el archivo jar a `share/java/kafka-connect-jdbc/`. Si tienes problemas para que Connect encuentre el driver, copia el driver a `share/confluent-hub-components/confluentinc-kafka-connect-jdbc/lib/`. O modifica `plugin.path` para incluir el driver; consulta más abajo.
    </Note>
  </Step>

  <Step title="Preparar la configuración" id="3-prepare-configuration">
    Sigue [estas instrucciones](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#set-up-a-local-connect-worker-with-cp-install) para configurar una instancia de Connect adecuada para tu tipo de instalación, teniendo en cuenta las diferencias entre un clúster `standalone` y uno distribuido. Si usas Confluent Cloud, la configuración distribuida es la que corresponde.

    Los siguientes parámetros son relevantes para usar el conector JDBC con ClickHouse. Puedes encontrar una lista completa de parámetros [aquí](https://docs.confluent.io/kafka-connect-jdbc/current/sink-connector/index.html):

    * `_connection.url_` - debe tener la forma `jdbc:clickhouse://&lt;clickhouse host>:&lt;clickhouse http port>/&lt;target database>`
    * `connection.user` - un usuario con acceso de escritura a la base de datos de destino
    * `table.name.format`- tabla de ClickHouse en la que insertar datos. Debe existir.
    * `batch.size` - El número de filas que se enviarán en un solo batch. Asegúrate de establecerlo en un valor suficientemente alto. Según las [recomendaciones](/docs/es/reference/statements/insert-into#performance-considerations) de ClickHouse, 1000 debe considerarse el valor mínimo.
    * `tasks.max` - El conector JDBC Sink admite ejecutar una o más tasks. Esto puede usarse para aumentar el rendimiento. Junto con el tamaño del batch, es tu principal mecanismo para mejorar el rendimiento.
    * `value.converter.schemas.enable` - Establécelo en false si usas un registro de esquemas y en true si incrustas los esquemas en los mensajes.
    * `value.converter` - Establécelo según tu tipo de dato; por ejemplo, para JSON, `io.confluent.connect.json.JsonSchemaConverter`.
    * `key.converter` - Establécelo en `org.apache.kafka.connect.storage.StringConverter`. Utilizamos claves String.
    * `pk.mode` - No es relevante para ClickHouse. Establécelo en none.
    * `auto.create` - No es compatible y debe ser false.
    * `auto.evolve` - Recomendamos false para esta configuración, aunque podría ser compatible en el futuro.
    * `insert.mode` - Establécelo en "insert". Actualmente no se admiten otros modos.
    * `key.converter` - Establécelo según los tipos de tus claves.
    * `value.converter` - Establécelo según el tipo de datos de tu topic. Estos datos deben tener un esquema compatible: formatos JSON, Avro o Protobuf.

    Si usas nuestro conjunto de datos de ejemplo para hacer pruebas, asegúrate de establecer lo siguiente:

    * `value.converter.schemas.enable` - Establécelo en false, ya que utilizamos un registro de esquemas. Establécelo en true si incrustas el esquema en cada mensaje.
    * `key.converter` - Establécelo en "org.apache.kafka.connect.storage.StringConverter". Utilizamos claves String.
    * `value.converter` - Establécelo en "io.confluent.connect.json.JsonSchemaConverter".
    * `value.converter.schema.registry.url` - Establece la URL del servidor de esquemas junto con las credenciales del servidor de esquemas mediante el parámetro `value.converter.schema.registry.basic.auth.user.info`.

    Puedes encontrar archivos de configuración de ejemplo para los datos de muestra de GitHub [aquí](https://github.com/ClickHouse/kafka-samples/tree/main/github_events/jdbc_sink), suponiendo que Connect se ejecuta en modo standalone y que Kafka está alojado en Confluent Cloud.
  </Step>

  <Step title="Crear la tabla de ClickHouse" id="4-create-the-clickhouse-table">
    Asegúrate de que la tabla se haya creado, eliminándola si ya existe de ejemplos anteriores. A continuación se muestra un ejemplo compatible con el conjunto de datos reducido de Github. Observa la ausencia de tipos Array o Map, que actualmente no son compatibles:

    ```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,
        state Enum('none' = 0, 'open' = 1, 'closed' = 2),
        assignee LowCardinality(String),
        closed_at DateTime,
        merged_at DateTime,
        merge_commit_sha 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="Iniciar Kafka Connect" id="5-start-kafka-connect">
    Inicia Kafka Connect en modo [standalone](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#standalone-cluster) o [distribuido](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

    ```bash theme={null}
    ./bin/connect-standalone connect.properties.ini github-jdbc-sink.properties.ini
    ```
  </Step>

  <Step title="Añadir datos a Kafka" id="6-add-data-to-kafka">
    Inserta mensajes en Kafka con el [script y la configuración](https://github.com/ClickHouse/kafka-samples/tree/main/producer) proporcionados. Tendrás que modificar `github.config` para incluir tus credenciales de Kafka. El script está configurado actualmente para usarse con Confluent Cloud.

    ```bash theme={null}
    python producer.py -c github.config
    ```

    Este script puede usarse para insertar cualquier archivo ndjson en un topic de Kafka. Intentará inferir automáticamente un esquema. La configuración de ejemplo proporcionada solo insertará 10k mensajes; [modifícala aquí](https://github.com/ClickHouse/clickhouse-docs/tree/main/docs/integrations/data-ingestion/kafka/code/producer/github.config#L25) si es necesario. Esta configuración también elimina del conjunto de datos, durante la inserción en Kafka, cualquier campo Array incompatible.

    Esto es necesario para que el conector JDBC convierta los mensajes en sentencias INSERT. Si usas tus propios datos, asegúrate de insertar un esquema con cada mensaje (estableciendo \_value.converter.schemas.enable \_en true) o de que tu client publique mensajes que hagan referencia a un esquema en el registro de esquemas.

    Kafka Connect debería comenzar a consumir mensajes e insertar filas en ClickHouse. Ten en cuenta que las advertencias sobre "\[JDBC Compliant Mode] Transaction isn't supported." son esperables y pueden ignorarse.

    Una consulta sencilla a la tabla de destino "Github" debería confirmar la inserción de datos.

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

    ```response theme={null}
    | count\(\) |
    | :--- |
    | 10000 |
    ```
  </Step>
</Steps>

<div id="recommended-further-reading">
  ### Lecturas adicionales recomendadas
</div>

* [Parámetros de configuración de Kafka Sink](https://docs.confluent.io/kafka-connect-jdbc/current/sink-connector/sink_config_options.html#sink-config-options)
* [Análisis en profundidad de Kafka Connect – conector JDBC Source](https://www.confluent.io/blog/kafka-connect-deep-dive-jdbc-source-connector)
* [Análisis en profundidad de Kafka Connect JDBC Sink: trabajo con claves primarias](https://rmoff.net/2021/03/12/kafka-connect-jdbc-sink-deep-dive-working-with-primary-keys/)
* [Kafka Connect en acción: JDBC Sink](https://www.youtube.com/watch?v=b-3qN_tlYR4\&t=981s) - para quienes prefieren verlo en lugar de leer.
* [Análisis en profundidad de Kafka Connect – explicación de convertidores y serialización](https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained/#json-schemas)