Cette rubrique explique comment utiliser le collecteur de données non conformes dans les jobs d'ingestion de données Flink CDC.
Présentation
Lors de la synchronisation des données en temps réel, il peut arriver que les données provenant d'une source ne puissent pas être analysées en raison de problèmes tels que des formats incorrects, des erreurs d'encodage ou des incompatibilités de schéma. Ce type de données impossibles à traiter est appelé « données non conformes ».
À partir de la version 11.5 de VVR, l'ingestion de données prend en charge la collecte des données non conformes pour la source de données Kafka. Les options de configuration du connecteur vous permettent d'ignorer les erreurs, d'enregistrer les détails et de poursuivre l'exécution du job.
Lorsque le connecteur rencontre des données impossibles à analyser, le système capture automatiquement les données brutes et les informations sur l'exception, puis les écrit dans un collecteur spécifié. Cette fonctionnalité vous permet de :
Tolérer une petite quantité de données non conformes pour éviter l'interruption de l'ensemble du pipeline.
Enregistrer le contexte complet pour faciliter le dépannage et l'analyse ultérieurs.
Définir un seuil afin d'éviter un nombre excessif d'erreurs.
Cas d'utilisation typiques
|
Cas d'utilisation |
Objectif |
|
Pipelines de collecte de journaux (Pour les sources de données non structurées telles que les journaux d'applications) |
La qualité des données est incohérente. Ignorez quelques enregistrements erronés pour garantir la poursuite de l'exécution du processus principal. |
|
Synchronisation des tables métier critiques (Pour les systèmes essentiels tels que les commandes ou les modifications de compte) |
Nécessite une cohérence élevée. Les erreurs déclenchent des alertes immédiates pour une intervention manuelle. |
|
Phase d'exploration et d'analyse des données |
Traitez rapidement l'ensemble du jeu de données afin de comprendre sa structure globale avant de résoudre les problèmes liés aux données non conformes. |
Limites et considérations
Avant d'utiliser cette fonctionnalité, prenez connaissance de ses limites et des risques potentiels :
Connecteurs pris en charge : cette fonctionnalité est actuellement disponible uniquement pour la source de données Kafka. La prise en charge d'autres sources sera ajoutée dans les versions futures.
Types de collecteurs pris en charge : seul le type
loggerest actuellement pris en charge ; il écrit les données non conformes dans un fichier journal.
Cette fonctionnalité convient aux phases de débogage et de début de production. Si de grandes quantités de données non conformes persistent, nous vous recommandons de mettre en œuvre des mesures de gouvernance des données sur le système en amont.
Syntaxe
Activer le collecteur de données non conformes
Le collecteur de données non conformes est défini dans le module de pipeline. La syntaxe est la suivante :
pipeline:
dirty-data.collector:
name: Logger Dirty Data Collector
type: logger|
Paramètre |
Description |
|
|
Nom du collecteur. Nous vous recommandons d'utiliser un nom explicite, tel que |
|
|
Type de collecteur. Valeurs possibles :
|
Si vous ne définissez pas cette option, le système n'enregistre pas les données non conformes, même si la tolérance aux erreurs est activée.
Configurer la tolérance aux erreurs dans la source de données
La simple configuration du collecteur de données non conformes ne permet pas d'ignorer les erreurs d'analyse. Vous devez utiliser cette fonctionnalité conjointement avec la politique de tolérance aux erreurs de Kafka. Pour plus de détails, consultez la documentation du connecteur Kafka. L'exemple suivant présente une configuration typique :
source:
type: kafka
# Skip the first 100 parsing errors; if the count exceeds 100, the job fails.
ingestion.ignore-errors: true
ingestion.error-tolerance.max-count: 100|
Paramètre |
Valeur par défaut |
Description |
|
|
|
Indique s'il faut ignorer les erreurs d'analyse. Si la valeur est définie sur |
|
|
|
Nombre maximal d'enregistrements de données non conformes à tolérer. Lorsque |
Collecteur de données non conformes Logger
Le collecteur de données non conformes Logger stocke les données non conformes dans un fichier journal distinct. Pour consulter les journaux des données non conformes, suivez les étapes ci-dessous :
Accédez à la page Deployments et cliquez sur l'onglet Logs.
Cliquez sur Logs, sélectionnez le sous-onglet Running Task Managers, puis sélectionnez le nœud TaskManager correspondant à l'opérateur concerné.
Cliquez sur Log List, puis cliquez sur le fichier journal nommé
yaml-dirty-data.outdans la liste pour afficher ou enregistrer les enregistrements de données non conformes.
Chaque enregistrement de données non conformes inclut les métadonnées suivantes :
L'horodatage du traitement des données non conformes
L'opérateur et l'index de sous-tâche qui ont émis l'enregistrement de données non conformes
Le contenu des données brutes
Les informations sur l'exception ayant provoqué l'échec du traitement
Format des enregistrements de données non conformes
Chaque enregistrement contient les métadonnées suivantes :
text[2025-04-05 10:23:45] [Operator: SourceKafka -> Subtask: 2]
Raw Data: {"id": "abc", "ts": "invalid-timestamp"}
Exception: java.time.format.DateTimeParseException: Text 'invalid-timestamp' could not be parsed at index 0
---|
Champ |
Description |
|
Horodatage |
Moment de la capture des données non conformes. |
|
Opérateur et sous-tâche |
Opérateur spécifique et index de sa sous-tâche parallèle où l'erreur s'est produite. |
|
Données brutes |
Données brutes non analysées (au format Base64 ou chaîne de caractères). |
|
Exception |
Type d'exception et résumé de la trace de la pile pour l'échec d'analyse. |
FAQ
Les données non conformes affectent-elles les checkpoints ?
Non. Les données non conformes sont interceptées avant de pouvoir être incluses dans une mise à jour d'état, elles ne provoquent donc pas l'échec des checkpoints.
Collecte des données non conformes vs sortie latérale
Collecteur de données non conformes : gère les données qui ne peuvent pas être désérialisées ou analysées.
Sortie latérale : gère les données qui peuvent être analysées mais ne répondent pas aux exigences de la logique métier.