Connecteur HTTP Sink Confluent

• Autogéré : Téléchargez le paquet Confluent et installez-le localement. Suivez les instructions d’installation du connecteur décrites 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 les utilisateurs de Confluent Cloud pour l’hébergement de Kafka. Cela nécessite que votre environnement ClickHouse soit accessible depuis Confluent Cloud.

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

• 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

• 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 renforcer la stratégie de nouvelle tentative.

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.

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

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 |