Modèle Dataflow Pub/Sub vers ClickHouse

Le modèle Pub/Sub vers ClickHouse est un pipeline de streaming qui lit des messages encodés en JSON à partir d’un abonnement Pub/Sub et les écrit dans une table ClickHouse. Les messages qui ne peuvent pas être analysés ou qui ne correspondent pas au schéma cible sont acheminés vers une destination dead-letter : une table ClickHouse, un topic Pub/Sub, ou les deux.

• L’abonnement Pub/Sub source doit exister.
• Les messages publiés dans l’abonnement doivent être au format JSON valide.
• La table ClickHouse cible doit exister, et ses noms de colonnes doivent correspondre aux noms des champs de la charge utile JSON.
• L’hôte ClickHouse doit être accessible depuis les machines worker de Dataflow.
• Au moins une destination dead-letter (clickHouseDeadLetterTable ou deadLetterTopic) doit être fournie. Si les deux sont fournies, les messages en échec sont acheminés vers les deux destinations simultanément.
• Lorsque clickHouseDeadLetterTable est défini, la table dead-letter doit déjà exister dans ClickHouse avec le schéma indiqué dans Gestion des dead-letter.
• Lorsque deadLetterTopic est défini, le topic Pub/Sub doit déjà exister.

Les messages Pub/Sub doivent être des objets JSON dont les noms des champs de premier niveau correspondent exactement aux noms des colonnes de la table ClickHouse cible.

Pour faire correspondre les messages entrants à la table cible, le pipeline effectue les opérations suivantes au démarrage :

1. Récupère le schéma de la table ClickHouse cible.
2. Construit un schéma Beam Row à partir de ce schéma ClickHouse.
3. Pour chaque message Pub/Sub entrant, analyse la charge utile JSON et assemble une ligne en lisant les champs nommés dans le schéma ClickHouse.

Les valeurs JSON sont converties dans le type de colonne ClickHouse correspondant :

Comme le pipeline traite les données en continu, les lignes entrantes sont accumulées dans des fenêtres avant d’être écrites dans ClickHouse. La stratégie de fenêtrage est déterminée à partir des paramètres que vous fournissez :

Le réglage de ces valeurs vous permet d’arbitrer entre la latence et l’efficacité des insertions. Des fenêtres plus petites réduisent la latence de bout en bout ; des fenêtres plus grandes produisent des lots INSERT moins nombreux, mais plus volumineux.

Les messages qui échouent lors de l'analyse JSON, de la correspondance de schéma ou de la coercition de type sont acheminés vers la ou les destinations dead-letter configurées. Au moins l'un de clickHouseDeadLetterTable ou deadLetterTopic doit être fourni ; si les deux sont définis, les messages en échec sont envoyés aux deux.

Lorsque clickHouseDeadLetterTable est défini, la table dead-letter doit déjà exister avec ce schéma fixe :

Définition minimale pour un déploiement sur un seul nœud :

CREATE TABLE my_table_dead_letter (
    raw_message   String,
    error_message String,
    stack_trace   String,
    failed_at     DateTime
) ENGINE = MergeTree()
ORDER BY failed_at;

Lorsque deadLetterTopic est défini, chaque message en échec est republié dans le topic avec :

• Payload : les octets du message d’origine.
• Attribut errorMessage : le message d’exception capturé au moment de l’échec.
• Attribut failedAt : l’horodatage de traitement indiquant le moment où la ligne a échoué.

Cela permet de rejouer facilement les messages en échec une fois le problème sous-jacent de schéma ou de producteur résolu.

Le modèle Pub/Sub vers ClickHouse est disponible depuis la Google Cloud Console.

Connectez-vous à votre Google Cloud Console et recherchez Dataflow.

1. Cliquez sur le bouton CREATE JOB FROM TEMPLATE.

   

2. Une fois le formulaire du modèle ouvert, saisissez un nom de job et sélectionnez la région souhaitée.

3. Dans le champ Dataflow Template, saisissez ClickHouse ou Pub/Sub, puis sélectionnez le modèle Pub/Sub vers ClickHouse.

4. Une fois le modèle sélectionné, le formulaire se déplie. Renseignez :

   • L'abonnement d'entrée Pub/Sub, au format projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_NAME>.
   • L'URL du point de terminaison ClickHouse — pour ClickHouse Cloud, utilisez https://<HOST>:8443.
   • La base de données ClickHouse, la table cible, le nom d'utilisateur et le mot de passe.
   • Au moins une destination dead-letter : une table ClickHouse ou un topic Pub/Sub (ou les deux).

5. Vous pouvez éventuellement personnaliser les paramètres de traitement par lots (windowSeconds, batchRowCount) et de réglage de ClickHouseIO, comme indiqué dans la section Template parameters.

Accédez à l’onglet Dataflow Jobs dans votre Google Cloud Console pour suivre l’état de la tâche. Vous y trouverez ses détails, notamment sa progression et les éventuelles erreurs :

Le modèle émet également les métriques personnalisées suivantes dans l’espace de noms PubSubToClickHouse, consultables depuis la page de la tâche Dataflow :

Cette erreur se produit lorsque ClickHouse manque de mémoire lors du traitement de lots volumineux de données. Pour résoudre ce problème :

• Augmentez les ressources de l'instance : passez votre ClickHouse server à une instance plus grande avec davantage de mémoire afin de prendre en charge la charge de traitement des données.
• Réduisez la taille des lots : réduisez batchRowCount (et/ou maxInsertBlockSize) dans la configuration de votre job Dataflow afin d'envoyer à ClickHouse des blocs de données plus petits, ce qui réduit la consommation de mémoire par lot.

Les causes les plus fréquentes sont les suivantes :

• Les noms des champs JSON ne correspondent pas exactement aux noms des colonnes ClickHouse (la correspondance est sensible à la casse).
• La valeur JSON ne peut pas être convertie vers le type de la colonne (par exemple, une chaîne non ISO-8601 dans une colonne DateTime).
• Le schéma de la table cible a changé depuis le démarrage du pipeline — le schéma n’est récupéré qu’une seule fois au démarrage. Redémarrez le job après avoir appliqué les modifications du schéma.

Inspectez les colonnes error_message et stack_trace de la table dead-letter ClickHouse (ou l’attribut errorMessage des messages dead-letter Pub/Sub) afin d’identifier la cause du problème.

• Vérifiez que l’abonnement reçoit bien des messages — consultez la métrique messages-received sur la page du job Dataflow.
• En mode temporel (windowSeconds uniquement), les lignes ne sont écrites qu’aux limites de la fenêtre. Réduisez windowSeconds pour vérifier que des flushes ont bien lieu.
• Vérifiez l’accessibilité réseau entre les workers Dataflow et le point de terminaison ClickHouse (pare-feu, VPC peering ou Private Service Connect).

Le code source du modèle est disponible dans :

• GoogleCloudPlatform/DataflowTemplates — le dépôt d’origine de Google Cloud Platform.
• ClickHouse/DataflowTemplates — le fork de ClickHouse.