Surveiller la fraîcheur des données pour les tables de CDC (Change Data Capture)
S'applique à : 
Connecteurs de base de données
Aperçu
Cette fonctionnalité est en préversion publique.
Cette fonctionnalité est déployée progressivement. Si vous ne voyez pas la colonne Récence des données dans votre pipeline, la fonctionnalité n'est pas disponible dans votre workspace.
La fraîcheur des données indique à quel point les données de chaque table de destination sont à jour par rapport à la base de données source. Pour chaque table dans un pipeline de connecteur de base de données, l’interface utilisateur du pipeline indique le temps écoulé depuis la modification la plus récente de la source que le pipeline a ingérée.
La fraîcheur des données est disponible pour les sources SQL Server, PostgreSQL, MySQL et Oracle.
Utilisez le caractère récent des données pour :
- Voyez à quel point chaque table de destination est en retard par rapport à la source, sans interroger le journal des événements.
- Identifier les tables qui sont à la traîne par rapport au reste du pipeline.
- Distinguez les tables qui reçoivent activement des modifications des tables qui sont inactives sur la source.
Comment le caractère récent des données est calculé
La fraîcheur des données est calculée pour chaque table à partir des valeurs suivantes :
Valeur | Définition |
|---|---|
Heure du commit source | Heure à laquelle une transaction a été validée sur la base de données source. |
Heure de commit de la source la plus récente | L’heure la plus récente du commit source parmi les modifications que le pipeline a ingérées pour la table. |
Fraîcheur des données | L'heure actuelle moins l'heure de commit de la dernière source. |
Par exemple, si le dernier changement ingéré pour une table a été validé sur la source il y a 3 minutes, la fraîcheur des données de la table est 3m. Une valeur inférieure signifie que la table de destination est plus proche de l'état actuel de la source.
La fraîcheur des données mesure la actualité de la table de destination. Il diffère des métriques de latence CDC dans le journal des événements de la passerelle, telles que discovery_latency_ms, qui mesurent uniquement le chemin de la base de données source vers le volume de transfert. Voir la latence CDC.
Afficher la fraîcheur des données
- Dans votre Databricks workspace, cliquez sur Jobs & Pipelines dans la barre latérale.
- Cliquez sur le nom de votre pipeline d’ingestion.
- Dans la liste des tables, recherchez la colonne Fraîcheur des données .
La colonne affiche une valeur pour chaque table, en heures, minutes et secondes (par exemple, 3m 23s ou 45h 32m 4s).
Fonctionnement de la récence des données
- Mises à jour en cours : les mises à jour de la fraîcheur des données sont continues. La valeur augmente au fil du temps et diminue lorsque le pipeline ingère de nouveaux changements de la source.
- Mises à jour arrêtées ou terminées : La fraîcheur des données cesse d'être mise à jour lorsque la mise à jour se termine. La colonne continue d'afficher la valeur à la fin de la mise à jour, ce qui vous permet de voir la fraîcheur de chaque table après l'arrêt du pipeline.
- Tables inactives : La fraîcheur des données reflète la dernière modification que le pipeline a reçue de la source. Si une table source n’a pas de nouveaux commits, la fraîcheur de ses données continue d’augmenter, même lorsque le pipeline est sain et entièrement à jour.
Interpréter la fraîcheur des données
Comme les tables sources inactives montrent également une fraîcheur de données croissante, comparez la valeur avec l'activité sur la source avant de la traiter comme un retard.
Modèle | Signification |
|---|---|
Faible et stable sur toutes les tables | Le pipeline est à jour avec la source. |
Augmentation sur une ou plusieurs tables | Les tables sources peuvent être inactives, sans nouveaux commits. Si les tables sources reçoivent des modifications, le pipeline est retardé. Voir Trouver l’origine d’un retard. |
Vide ( | Le pipeline n’a ingéré aucune modification pour la table depuis son démarrage, par exemple pendant l’exécution de l’instantané initial. Voir Limitations. |
Trouver l’origine d’un délai
Pour les pipelines qui utilisent une passerelle d’ingestion, comparez le changement de source le plus récent lu par la passerelle avec l’heure de commit de la source la plus récente ingérée par le pipeline d’ingestion. La passerelle signale event_time:max dans son journal d’événements. Le pipeline d’ingestion signale upstream_time:max dans son journal d’événements. Voir query la récence des données à partir du journal d’événements.
Passerelle | Pipeline d’ingestion | Signification |
|---|---|---|
Avancement | Avancement | Les modifications circulent dans le pipeline. |
Avancement | Ne progresse pas | Le pipeline d’ingestion est en retard par rapport à la passerelle. Vérifiez l’état du pipeline d’ingestion et sa planification. |
Ne progresse pas | Ne progresse pas | La source est inactive ou la passerelle a perdu sa connexion avec la source. Pour diagnostiquer la passerelle, consultez la latence CDC. |
Actualité des données de query à partir du journal des événements
Le log d'événements du pipeline d'ingestion, et non celui de la passerelle, enregistre les métriques sur lesquelles la fraîcheur des données est basée. query le log d'événements pour suivre la fraîcheur des données au fil du temps, comparer des tables ou créer des alertes.
Chaque événement flow_progress pour un flux CDC rapporte les champs suivants sous details:flow_progress:streaming_metrics:
Champ | Type | Description |
|---|---|---|
| Chaîne | Heure de commit de la source la plus récente parmi les modifications dans le batch, sous la forme d’un timestamp ISO 8601 en UTC. |
| Chaîne | Le earliest source commit time parmi les modifications dans le batch, en tant que timestamp ISO 8601 en UTC. |
| Entier | Le délai de bout en bout pour le batch, en millisecondes. Le délai est mesuré entre l'heure de commit source de la modification la plus ancienne dans le batch et l'heure à laquelle le batch a été commit dans la table de destination. |
Dans chaque query, remplacez <ingestion-pipeline-id> par l’ID de votre pipeline d’ingestion. Le journal des événements de la passerelle ne rapporte pas ces champs.
Fraîcheur des données et dernier délai par table
La query suivante retourne l’heure de commit de la source la plus récente, la récence des données, l’heure du dernier batch et le délai de bout en bout du batch le plus récent pour chaque table. Les résultats sont triés de sorte que les tables les moins récentes apparaissent en premier.
La query inclut uniquement les tables de la dernière mise à jour du pipeline. Sans ce filtre, les tables que vous avez supprimées du pipeline restent dans le journal d'événements, et leur fraîcheur des données continue d'augmenter.
WITH latest_update AS (
SELECT origin.update_id AS update_id
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'create_update'
ORDER BY timestamp DESC
LIMIT 1
),
current_flows AS (
SELECT DISTINCT origin.flow_name AS flow_name
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IS NOT NULL
AND origin.update_id = (SELECT update_id FROM latest_update)
)
SELECT
origin.flow_name AS flow_name,
MAX(details:flow_progress:streaming_metrics:upstream_time:max::timestamp) AS latest_source_commit_time,
TIMESTAMPDIFF(
MINUTE,
MAX(details:flow_progress:streaming_metrics:upstream_time:max::timestamp),
current_timestamp()
) AS data_freshness_minutes,
MAX(timestamp) AS last_batch_time,
MAX_BY(details:flow_progress:streaming_metrics:stream_latency_ms::bigint, timestamp) AS latest_lag_ms
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IN (SELECT flow_name FROM current_flows)
AND details:flow_progress:streaming_metrics:upstream_time:max IS NOT NULL
GROUP BY origin.flow_name
ORDER BY data_freshness_minutes DESC
Délai de bout en bout au fil du temps
La query suivante renvoie le retard de bout en bout maximal horaire pour chaque table au cours des dernières 24 heures. Utilisez-la pour savoir quand le retard a augmenté.
SELECT
DATE_TRUNC('HOUR', timestamp) AS ts_hour,
origin.flow_name AS flow_name,
ROUND(MAX(details:flow_progress:streaming_metrics:stream_latency_ms::bigint) / 1000.0 / 60.0, 1) AS max_lag_minutes
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IS NOT NULL
AND details:flow_progress:streaming_metrics:stream_latency_ms IS NOT NULL
AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
GROUP BY DATE_TRUNC('HOUR', timestamp), origin.flow_name
ORDER BY origin.flow_name, ts_hour
Créer une alerte sur la fraîcheur des données
Pour être averti lorsqu'une table n'est plus à jour, créez une alerte Databricks SQL sur une query de fraîcheur des données. Par exemple, la query suivante retourne le nombre de tables dans la dernière mise à jour du pipeline dont la fraîcheur des données est supérieure à 120 minutes. Remplacez 120 par votre threshold et configurez l'alerte pour qu'elle Trigger lorsque stale_tables est supérieur à 0.
WITH latest_update AS (
SELECT origin.update_id AS update_id
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'create_update'
ORDER BY timestamp DESC
LIMIT 1
),
current_flows AS (
SELECT DISTINCT origin.flow_name AS flow_name
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IS NOT NULL
AND origin.update_id = (SELECT update_id FROM latest_update)
),
per_table_freshness AS (
SELECT
origin.flow_name AS flow_name,
TIMESTAMPDIFF(
MINUTE,
MAX(details:flow_progress:streaming_metrics:upstream_time:max::timestamp),
current_timestamp()
) AS data_freshness_minutes
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IN (SELECT flow_name FROM current_flows)
AND details:flow_progress:streaming_metrics:upstream_time:max IS NOT NULL
GROUP BY origin.flow_name
)
SELECT
COUNT_IF(data_freshness_minutes > 120) AS stale_tables
FROM per_table_freshness
Les tables qui n’ont pas encore de valeur de fraîcheur des données, telles que les tables dont l’instantané initial est en cours d’exécution, ne sont pas comptées.
La query compare l’heure du dernier commit source avec l’heure actuelle ; sa valeur ne cesse donc d’augmenter pour les tables sources inactives et pendant que le pipeline n’est pas en cours d’exécution. Contrairement à la valeur dans l’interface utilisateur, elle ne s’arrête pas de changer lorsqu’une mise à jour se termine. Pour éviter les alertes qui se déclenchent lorsque les données sont à jour :
- Définissez le threshold en fonction de la fréquence à laquelle vos tables sources reçoivent des modifications.
- Pour un pipeline qui s'exécute selon une planification, définissez le threshold plus haut que l'intervalle entre les exécutions planifiées.
- Si vos tables reçoivent des modifications à des taux très différents, créez une alerte distincte pour chaque groupe de tables avec des taux similaires et filtrez chaque query sur
flow_name.
Consultez les alertes Databricks SQL et Créer une alerte.
Limites
- Tables sans modifications : si une table n'a subi aucune modification depuis le démarrage du pipeline, la fraîcheur des données est vide. L'UI affiche un tiret (
-), et le journal des événements n'a aucune valeurupstream_timepour la table. Cela inclut les tables dont le snapshot initial est en cours d'exécution. - Sources SQL Server : lorsqu'il n'existe pas de commit time source pour une modification, la passerelle utilise un commit time en cache qu'elle refresh toutes les 120 secondes. Tant que la valeur mise en cache n'est pas refresh, la fraîcheur des données peut être supérieure au délai réel.
- Sources Oracle : pour les tables Oracle avec des colonnes LOB (Large Object), telles que
BLOB,CLOBouNCLOB, la fraîcheur des données peut être négative. Une valeur négative signifie que l’heure de commit de la source enregistrée est postérieure à l’heure actuelle. Cela ne signifie pas que les données sont plus récentes que la source. Si vous voyez des valeurs négatives, contactez l'assistance Databricks.