Conector HTTP Sink de Confluent

• Autogestionado: Descargue el paquete de Confluent e instálelo localmente. Siga las instrucciones de instalación del conector, como se documenta aquí. Si utiliza el método de instalación confluent-hub, se actualizarán sus archivos de configuración locales.
• Confluent Cloud: Hay una versión totalmente gestionada de HTTP Sink disponible para quienes usan Confluent Cloud para alojar Kafka. Esto requiere que su entorno de ClickHouse sea accesible desde Confluent Cloud.

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

• 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 como 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 hacer los reintentos más robustos.

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

1. Verifique que sus registros de Kafka tengan la misma clave.
2. 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.

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)

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

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

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

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)

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

SELECT count() FROM default.github;

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