Utiliser Vector avec Kafka et ClickHouse

Vector est un pipeline de données indépendant des fournisseurs, capable de lire des données depuis Kafka et d’envoyer des événements vers ClickHouse. Le guide de prise en main de Vector avec ClickHouse est centré sur le cas d’usage des logs et la lecture d’événements depuis un fichier. Nous utilisons le jeu de données d’exemple GitHub, dont les événements sont stockés dans un topic Kafka. Vector utilise des sources pour récupérer les données selon un modèle push ou pull. Les sinks servent quant à eux de destination pour les événements. Nous utilisons donc la source Kafka et le sink ClickHouse. Notez que, bien que Kafka soit pris en charge comme sink, il n’existe pas de source ClickHouse. Vector n’est donc pas adapté si vous souhaitez transférer des données de ClickHouse vers Kafka. Vector prend également en charge la transformation des données. Cela dépasse le cadre de ce guide. Si vous en avez besoin pour votre jeu de données, consultez la documentation de Vector. Notez que l’implémentation actuelle du sink ClickHouse utilise l’interface HTTP. Le sink ClickHouse ne prend pas en charge l’utilisation d’un schéma JSON pour le moment. Les données doivent être publiées dans Kafka soit au format JSON brut, soit sous forme de Strings.

Licence

Vector est distribué sous la licence MPL-2.0

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

Étapes

Créez le topic Kafka github et insérez le jeu de données GitHub.

cat /opt/data/github/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

Ce jeu de données comprend 200 000 lignes axées sur le dépôt ClickHouse/ClickHouse.

Assurez-vous que la table cible est créée. Ci-dessous, nous utilisons la base de données par défaut.

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

Téléchargez et installez Vector. Créez un fichier de configuration kafka.toml, puis modifiez les valeurs en fonction de vos instances Kafka et ClickHouse.

[sources.github]
type = "kafka"
auto_offset_reset = "smallest"
bootstrap_servers = "<kafka_host>:<kafka_port>"
group_id = "vector"
topics = [ "github" ]
tls.enabled = true
sasl.enabled = true
sasl.mechanism = "PLAIN"
sasl.username = "<username>"
sasl.password = "<password>"
decoding.codec = "json"

[sinks.clickhouse]
type = "clickhouse"
inputs = ["github"]
endpoint = "http://localhost:8123"
database = "default"
table = "github"
skip_unknown_fields = true
auth.strategy = "basic"
auth.user = "username"
auth.password = "password"
buffer.max_events = 10000
batch.timeout_secs = 1

Quelques points importants concernant cette configuration et le comportement de Vector :

Cet exemple a été testé avec Confluent Cloud. Par conséquent, les options de sécurité sasl.* et ssl.enabled peuvent ne pas convenir aux environnements auto-gérés.
Un préfixe de protocole n’est pas nécessaire pour le paramètre de configuration bootstrap_servers, par ex. pkc-2396y.us-east-1.aws.confluent.cloud:9092
Le paramètre source decoding.codec = "json" garantit que le message est transmis au sink ClickHouse sous la forme d’un seul objet JSON. Si vous traitez les messages comme des Strings et utilisez la valeur par défaut bytes, le contenu du message sera ajouté à un champ message. Dans la plupart des cas, cela nécessitera un traitement dans ClickHouse, comme décrit dans le guide de prise en main de Vector.
Vector ajoute plusieurs champs aux messages. Dans notre exemple, nous ignorons ces champs dans le sink ClickHouse via le paramètre de configuration skip_unknown_fields = true. Cela permet d’ignorer les champs qui ne font pas partie du schéma de la table cible. N’hésitez pas à ajuster votre schéma pour inclure ces métachamps, comme offset.
Remarquez comment le sink fait référence à la source des événements via le paramètre inputs.
Notez le comportement du sink ClickHouse tel que décrit ici. Pour un débit optimal, vous pouvez ajuster les paramètres buffer.max_events, batch.timeout_secs et batch.max_bytes. Conformément aux recommandations de ClickHouse, une valeur de 1000 doit être considérée comme le minimum pour le nombre d’événements dans un même batch. Pour les cas d’usage à débit élevé et régulier, vous pouvez augmenter le paramètre buffer.max_events. Des débits plus variables peuvent nécessiter d’ajuster le paramètre batch.timeout_secs
Le paramètre auto_offset_reset = "smallest" force la source Kafka à démarrer au début du topic, garantissant ainsi que nous consommons les messages publiés à l’étape (1). Vous pouvez avoir besoin d’un comportement différent. Voir ici pour plus de détails.

Démarrez Vector

vector --config ./kafka.toml

Par défaut, une vérification d’état est requise avant de commencer les insertions dans ClickHouse. Cela garantit que la connexion peut être établie et que le schéma peut être lu. Faites précéder la commande de VECTOR_LOG=debug pour obtenir des logs supplémentaires, ce qui peut s’avérer utile si vous rencontrez des problèmes.

Confirmez l’insertion des données.

SELECT count() AS count FROM github;

count
200000