Conector HTTP Sink de Confluent

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

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:

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

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

Tiene dos opciones:

• Autogestionado: Descargue el paquete de Confluent e instálelo localmente. Siga las instrucciones de instalación del conector que se documentan aquí. Si utiliza el método de instalación de confluent-hub, sus archivos de configuración locales se actualizarán.

• Confluent Cloud: Hay disponible una versión totalmente gestionada de HTTP Sink para quienes usan Confluent Cloud para alojar Kafka. Esto requiere que su entorno de ClickHouse sea accesible desde Confluent Cloud.

Antes de la prueba de conectividad, comencemos creando una tabla de prueba en ClickHouse Cloud; esta tabla recibirá los datos de Kafka:

CREATE TABLE default.my_table
(
    `side` String,
    `quantity` Int32,
    `symbol` String,
    `price` Int32,
    `account` String,
    `userid` String
)
ORDER BY tuple()

Cree un topic de Kafka y una instancia de conector HTTP Sink:

Configure el conector HTTP Sink:

• Indique el nombre del topic que creó
• Autenticación
  • HTTP Url - URL de ClickHouse Cloud con una consulta INSERT especificada: <protocol>://<clickhouse_host>:<clickhouse_port>?query=INSERT%20INTO%20<database>.<table>%20FORMAT%20JSONEachRow. Nota: la consulta debe estar codificada.
  • Endpoint Authentication type - BASIC
  • Auth username - nombre de usuario de ClickHouse
  • Auth password - contraseña de ClickHouse

• Configuración
  • Input Kafka record value format Depende de los datos de origen, pero en la mayoría de los casos será JSON o Avro. En la siguiente configuración asumimos JSON.
  • En la sección advanced configurations:
    • HTTP Request Method - Establézcalo en POST
    • Request Body Format - json
    • Batch batch size - Según las recomendaciones de ClickHouse, establézcalo en al menos 1000.
    • Batch json as array - true
    • Retry on HTTP codes - 400-500, pero adáptelo según sea necesario; por ejemplo, esto puede cambiar si tiene un proxy HTTP delante de ClickHouse.
    • Maximum Reties - el valor predeterminado (10) es adecuado, pero no dude en ajustarlo para lograr reintentos más robustos.

Cree un mensaje en un topic configurado por su HTTP Sink

y verifique que el mensaje creado se haya escrito en su instancia de ClickHouse.

Siga estas instrucciones para configurar Connect según su tipo de instalación, teniendo en cuenta las diferencias entre un clúster standalone y uno distribuido. Si usa Confluent Cloud, la configuración distribuida es la adecuada.

El parámetro más importante es http.api.url. La interfaz HTTP de ClickHouse requiere codificar la sentencia INSERT como un parámetro en la URL. Debe incluir el formato (JSONEachRow en este caso) y la base de datos de destino. El formato debe ser coherente con los datos de Kafka, que se convertirán en una cadena en el payload HTTP. Estos parámetros deben codificarse para URL. A continuación se muestra un ejemplo de este formato para el conjunto de datos de GitHub (suponiendo que está ejecutando ClickHouse localmente):

<protocol>://<clickhouse_host>:<clickhouse_port>?query=INSERT%20INTO%20<database>.<table>%20FORMAT%20JSONEachRow

http://localhost:8123?query=INSERT%20INTO%20default.github%20FORMAT%20JSONEachRow

Los siguientes parámetros adicionales son relevantes para usar HTTP Sink con ClickHouse. Puede encontrar una lista completa de parámetros aquí:

• request.method - Establézcalo en POST
• retry.on.status.codes - Establézcalo en 400-500 para reintentar ante cualquier código de error. Ajústelo según los errores esperados en los datos.
• request.body.format - En la mayoría de los casos, será JSON.
• auth.type - Establézcalo en BASIC si utiliza autenticación con ClickHouse. Actualmente no se admiten otros mecanismos de autenticación compatibles con ClickHouse.
• ssl.enabled - establézcalo en true si usa SSL.
• connection.user - nombre de usuario de ClickHouse.
• connection.password - contraseña de ClickHouse.
• batch.max.size - El número de filas que se enviarán en un solo lote. Asegúrese de que este valor sea lo suficientemente grande. Según las recomendaciones de ClickHouse, un valor de 1000 debe considerarse el mínimo.
• tasks.max - El conector HTTP Sink admite ejecutar una o más tareas. Esto puede usarse para aumentar el rendimiento. Junto con el tamaño del lote, este es el principal medio para mejorar el rendimiento.
• key.converter - establézcalo según los tipos de sus claves.
• value.converter - establézcalo según el tipo de datos de su topic. Estos datos no necesitan un esquema. El formato aquí debe ser coherente con el FORMAT especificado en el parámetro http.api.url. La opción más sencilla es usar JSON y el converter org.apache.kafka.connect.json.JsonConverter. También es posible tratar el valor como una cadena mediante el converter org.apache.kafka.connect.storage.StringConverter, aunque esto requerirá que el usuario extraiga un valor en la sentencia insert usando funciones. El formato Avro también es compatible con ClickHouse si se usa el converter io.confluent.connect.avro.AvroConverter.

Puede encontrar aquí una lista completa de la configuración, incluida la forma de configurar un proxy, los reintentos y SSL avanzado.

Puede encontrar aquí archivos de configuración de ejemplo para los datos de muestra de GitHub, suponiendo que Connect se ejecuta en modo standalone y que Kafka está alojado en Confluent Cloud.

Asegúrese de que la tabla se ha creado. A continuación se muestra un ejemplo de un conjunto de datos mínimo de GitHub que utiliza un MergeTree estándar.

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)

Inserte mensajes en Kafka. A continuación, usamos kcat para insertar 10k mensajes.

head -n 10000 github_all_columns.ndjson | kcat -b <host>:<port> -X security.protocol=sasl_ssl -X sasl.mechanisms=PLAIN -X sasl.username=<username>  -X sasl.password=<password> -t github

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

SELECT count() FROM default.github;

| count\