Connecteur HTTP Sink Confluent - ClickHouse Documentation

Le HTTP Sink Connector est indépendant du type de données ; il n’a donc pas besoin de schéma Kafka et prend en charge des types de données spécifiques à ClickHouse, tels que Maps et Arrays. Cette flexibilité supplémentaire s’accompagne d’une légère complexité de configuration. Nous décrivons ci-dessous une installation simple qui récupère les messages d’un seul topic Kafka et insère les lignes dans une table ClickHouse.

Le connecteur HTTP est distribué sous la Confluent Enterprise License.

Étapes de démarrage rapide

Rassemblez vos informations de connexion

Pour vous connecter à ClickHouse via HTTP(S), vous avez besoin des informations suivantes :

Paramètre(s)	Description
`HOST` and `PORT`	En général, le port est 8443 lors de l’utilisation de TLS, ou 8123 sans TLS.
`DATABASE NAME`	Par défaut, une base de données nommée `default` est disponible ; utilisez le nom de la base de données à laquelle vous voulez vous connecter.
`USERNAME` and `PASSWORD`	Par défaut, le nom d’utilisateur est `default`. Utilisez le nom d’utilisateur adapté à votre cas d’usage.

Les informations de votre service ClickHouse Cloud sont disponibles dans la console ClickHouse Cloud. Sélectionnez un service, puis cliquez sur Connect :

Choisissez HTTPS. Les détails de connexion s’affichent dans un exemple de commande curl.

Détails de connexion HTTPS ClickHouse Cloud

Si vous utilisez ClickHouse autogéré, les détails de connexion sont définis par votre administrateur ClickHouse.

Exécuter Kafka Connect et le HTTP Sink Connector

Vous avez deux options :

Autogéré : Téléchargez le paquet Confluent et installez-le localement. Suivez les instructions d’installation du connecteur, comme indiqué ici. Si vous utilisez la méthode d’installation confluent-hub, vos fichiers de configuration locaux seront mis à jour.
Confluent Cloud : Une version entièrement gérée de HTTP Sink est disponible pour celles et ceux qui utilisent Confluent Cloud pour héberger Kafka. Cela nécessite que votre environnement ClickHouse soit accessible depuis Confluent Cloud.

Les exemples suivants utilisent Confluent Cloud.

Créer une table de destination dans ClickHouse

Avant le test de connectivité, commençons par créer une table de test dans ClickHouse Cloud. Cette table recevra les données de Kafka :

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

Configurer HTTP Sink

Créez un topic Kafka et une instance du HTTP Sink Connector :
Configurez le HTTP Sink Connector :

Indiquez le nom du topic que vous avez créé
Authentification
- HTTP Url - URL ClickHouse Cloud avec une requête INSERT spécifiée : <protocol>://<clickhouse_host>:<clickhouse_port>?query=INSERT%20INTO%20<database>.<table>%20FORMAT%20JSONEachRow. Remarque : la requête doit être encodée.
- Endpoint Authentication type - BASIC
- Auth username - nom d’utilisateur ClickHouse
- Auth password - mot de passe ClickHouse

Cette URL HTTP est source d’erreurs. Assurez-vous que l’échappement est précis pour éviter tout problème.

Configuration
- Input Kafka record value formatCela dépend de vos données source, mais dans la plupart des cas, il s’agit de JSON ou d’Avro. Nous partons de l’hypothèse de JSON dans les paramètres suivants.
- Dans la section advanced configurations :
  - HTTP Request Method - Définir sur POST
  - Request Body Format - json
  - Batch batch size - Conformément aux recommandations de ClickHouse, définissez cette valeur sur au moins 1000.
  - Batch json as array - true
  - Retry on HTTP codes - 400-500, mais adaptez selon vos besoins ; par exemple, cela peut changer si vous avez un proxy HTTP devant ClickHouse.
  - Maximum Reties - la valeur par défaut (10) convient, mais n’hésitez pas à l’ajuster pour rendre les tentatives de nouvelle exécution plus robustes.

Tester la connectivité

Créez un message dans un topic configuré par votre HTTP Sink
et vérifiez que le message créé a bien été écrit dans votre instance ClickHouse.

Dépannage

HTTP Sink ne traite pas les messages par lot

D’après la documentation du Sink :

Le connecteur HTTP Sink ne regroupe pas les requêtes pour les messages contenant des valeurs d’en-tête Kafka différentes.

Vérifiez que vos enregistrements Kafka ont la même clé.
Lorsque vous ajoutez des paramètres à l’URL de l’API HTTP, chaque enregistrement peut générer une URL unique. Pour cette raison, le traitement par lot est désactivé lorsque des paramètres d’URL supplémentaires sont utilisés.

400 Requête incorrecte

CANNOT_PARSE_QUOTED_STRING

Si HTTP Sink échoue avec le message suivant lors de l’insertion d’un objet JSON dans une colonne 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)

Définissez le paramètre input_format_json_read_objects_as_strings=1 dans l’URL sous la forme d’une chaîne encodée SETTINGS%20input_format_json_read_objects_as_strings%3D1

Charger le jeu de données GitHub (facultatif)

Notez que cet exemple conserve les champs Array du jeu de données GitHub. Nous partons du principe que vous disposez d’un topic github vide dans les exemples et que vous utilisez kcat pour insérer des messages dans Kafka.

Préparer la configuration

Suivez ces instructions pour configurer Connect en fonction de votre type d’installation, en tenant compte des différences entre un cluster en mode autonome et un cluster distribué. Si vous utilisez Confluent Cloud, c’est la configuration distribuée qui s’applique.Le paramètre le plus important est http.api.url. L’interface HTTP de ClickHouse exige que vous encodiez l’instruction INSERT dans l’URL sous forme de paramètre. Celui-ci doit inclure le format (JSONEachRow dans ce cas) et la base de données cible. Le format doit être cohérent avec les données Kafka, qui seront converties en chaîne dans le payload HTTP. Ces paramètres doivent être encodés pour l’URL. Un exemple de ce format pour le jeu de données GitHub (en supposant que vous exécutez ClickHouse localement) est présenté ci-dessous :

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

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

Les paramètres supplémentaires suivants sont pertinents pour l’utilisation de HTTP Sink avec ClickHouse. Une liste complète des paramètres est disponible ici :

request.method - Définir sur POST
retry.on.status.codes - Définir sur 400-500 pour réessayer sur tous les codes d’erreur. À ajuster en fonction des erreurs attendues dans les données.
request.body.format - Dans la plupart des cas, ce sera JSON.
auth.type - Définir sur BASIC si vous utilisez l’authentification avec ClickHouse. Les autres mécanismes d’authentification compatibles avec ClickHouse ne sont pas pris en charge actuellement.
ssl.enabled - définir sur true si vous utilisez SSL.
connection.user - nom d’utilisateur pour ClickHouse.
connection.password - mot de passe pour ClickHouse.
batch.max.size - Le nombre de lignes à envoyer dans un seul lot. Assurez-vous que cette valeur est définie sur un nombre suffisamment élevé. Conformément aux recommandations de ClickHouse, 1000 doit être considéré comme une valeur minimale.
tasks.max - Le connecteur HTTP Sink permet d’exécuter une ou plusieurs tâches. Cela peut être utilisé pour améliorer les performances. Avec la taille des lots, c’est votre principal levier d’optimisation des performances.
key.converter - définir en fonction des types de vos clés.
value.converter - définir en fonction du type de données de votre topic. Ces données n’ont pas besoin de schéma. Le format indiqué ici doit être cohérent avec le FORMAT spécifié dans le paramètre http.api.url. Le plus simple est d’utiliser JSON et le convertisseur org.apache.kafka.connect.json.JsonConverter. Il est également possible de traiter la valeur comme une chaîne via le convertisseur org.apache.kafka.connect.storage.StringConverter, bien que cela oblige l’utilisateur à extraire une valeur dans l’instruction insert à l’aide de fonctions. Le format Avro est également pris en charge dans ClickHouse si vous utilisez le convertisseur io.confluent.connect.avro.AvroConverter.

Une liste complète des paramètres, y compris la configuration d’un proxy, des nouvelles tentatives et des options SSL avancées, est disponible ici.Des fichiers de configuration d’exemple pour les données d’exemple GitHub sont disponibles ici, en supposant que Connect s’exécute en mode autonome et que Kafka est hébergé dans Confluent Cloud.

Créer la table ClickHouse

Assurez-vous que la table a bien été créée. Un exemple de jeu de données GitHub minimal reposant sur un MergeTree standard est présenté ci-dessous.

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)

Ajouter des données à Kafka

Insérez des messages dans Kafka. Ci-dessous, nous utilisons kcat pour y insérer 10k messages.

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

Une simple requête de lecture sur la table cible “Github” devrait confirmer l’insertion des données.

SELECT count() FROM default.github;

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

​Étapes de démarrage rapide

​Dépannage

​HTTP Sink ne traite pas les messages par lot

​400 Requête incorrecte

CANNOT_PARSE_QUOTED_STRING

​Charger le jeu de données GitHub (facultatif)

Étapes de démarrage rapide

Dépannage

HTTP Sink ne traite pas les messages par lot

400 Requête incorrecte

Charger le jeu de données GitHub (facultatif)