JDBC 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í.

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.

Descarga e instala el ClickHouse JDBC driver clickhouse-jdbc-<version>-shaded.jar desde aquí. Instálalo en Kafka Connect siguiendo las instrucciones aquí. Es posible que otros drivers funcionen, pero no se han probado.

Sigue estas instrucciones 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í:

• _connection.url_ - debe tener la forma jdbc:clickhouse://<clickhouse host>:<clickhouse http port>/<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 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í, suponiendo que Connect se ejecuta en modo standalone y que Kafka está alojado en Confluent Cloud.

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:

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)

Inicia Kafka Connect en modo standalone o distribuido.

./bin/connect-standalone connect.properties.ini github-jdbc-sink.properties.ini

Inserta mensajes en Kafka con el script y la configuración proporcionados. Tendrás que modificar github.config para incluir tus credenciales de Kafka. El script está configurado actualmente para usarse con Confluent Cloud.

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

SELECT count() FROM default.github;

| count\(\) |
| :--- |
| 10000 |