Conector HTTP Sink de Confluent - ClickHouse Documentation

El conector HTTP Sink es independiente del tipo de datos, por lo que no necesita un esquema de Kafka y, además, admite tipos de datos específicos de ClickHouse, como Maps y Arrays. Esta flexibilidad adicional conlleva un ligero aumento de la complejidad de la configuración. 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.

El HTTP Connector se distribuye bajo la Licencia Enterprise de Confluent.

Pasos de inicio rápido

Reúna sus datos de conexión

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:

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

Detalles de conexión HTTPS de ClickHouse Cloud

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

Ejecute Kafka Connect y el conector HTTP Sink

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.

Los siguientes ejemplos utilizan Confluent Cloud.

Crear la tabla de destino en ClickHouse

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()

Configurar HTTP Sink

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

Esta HTTP Url es propensa a errores. Asegúrese de que el escaping sea preciso para evitar problemas.

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.

Probar la conectividad

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.

Solución de problemas

HTTP Sink no agrupa los mensajes en lotes

Según la documentación del sink:

El conector HTTP Sink no agrupa solicitudes de mensajes que contienen valores de cabecera de Kafka diferentes.

Verifique que sus registros de Kafka tengan la misma clave.
Cuando agrega parámetros a la URL de la API HTTP, cada registro puede generar una URL única. Por este motivo, la agrupación en lotes se desactiva cuando se usan parámetros de URL adicionales.

400 solicitud incorrecta

CANNOT_PARSE_QUOTED_STRING

Si HTTP Sink falla y muestra el siguiente mensaje al insertar un objeto JSON en una columna String:

Code: 26. DB::ParsingException: Cannot parse JSON string: expected opening quote: (while reading the value of key key_name): While executing JSONEachRowRowInputFormat: (at row 1). (CANNOT_PARSE_QUOTED_STRING)

Defina la configuración input_format_json_read_objects_as_strings=1 en la URL como una cadena codificada: SETTINGS%20input_format_json_read_objects_as_strings%3D1

Cargar el conjunto de datos de GitHub (opcional)

Tenga en cuenta que este ejemplo conserva los campos Array del conjunto de datos de GitHub. Suponemos que dispone de un topic github vacío para este ejemplo y que usa kcat para insertar mensajes en Kafka.

Preparar la configuración

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.

Crear la tabla de ClickHouse

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)

Agregar datos a Kafka

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\(\) |
| :--- |
| 10000 |

​Pasos de inicio rápido

​Solución de problemas

​HTTP Sink no agrupa los mensajes en lotes

​400 solicitud incorrecta

CANNOT_PARSE_QUOTED_STRING

​Cargar el conjunto de datos de GitHub (opcional)

Pasos de inicio rápido

Solución de problemas

HTTP Sink no agrupa los mensajes en lotes

400 solicitud incorrecta

Cargar el conjunto de datos de GitHub (opcional)