%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 format`Cela 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.
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](https://docs.confluent.io/kafka-connectors/http/current/overview.html#http-sink-connector-for-cp) :
> Le connecteur HTTP Sink ne regroupe pas les requêtes pour les messages contenant des valeurs d'en-tête Kafka différentes.
1. Vérifiez que vos enregistrements Kafka ont la même clé.
2. 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` :
```response theme={null}
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](https://github.com/edenhill/kcat) pour insérer des messages dans Kafka.
Suivez [ces instructions](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#set-up-a-local-connect-worker-with-cp-install) 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](/docs/fr/concepts/features/interfaces/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 :
```response theme={null}
://:?query=INSERT%20INTO%20.%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](https://docs.confluent.io/kafka-connect-http/current/connector_config.html) :
* `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](/docs/fr/reference/statements/insert-into#performance-considerations) 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](/docs/fr/reference/formats/Avro/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](https://docs.confluent.io/kafka-connect-http/current/connector_config.html).
Des fichiers de configuration d’exemple pour les données d’exemple GitHub sont disponibles [ici](https://github.com/ClickHouse/clickhouse-docs/tree/main/docs/integrations/data-ingestion/kafka/code/connectors/http_sink), en supposant que Connect s’exécute en mode autonome et que Kafka est hébergé dans Confluent Cloud.
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.
```sql theme={null}
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)
```
Insérez des messages dans Kafka. Ci-dessous, nous utilisons [kcat](https://github.com/edenhill/kcat) pour y insérer 10k messages.
```bash theme={null}
head -n 10000 github_all_columns.ndjson | kcat -b : -X security.protocol=sasl_ssl -X sasl.mechanisms=PLAIN -X sasl.username= -X sasl.password= -t github
```
Une simple requête de lecture sur la table cible "Github" devrait confirmer l’insertion des données.
```sql theme={null}
SELECT count() FROM default.github;
| count\(\) |
| :--- |
| 10000 |
```
