Surveiller la progression de la passerelle d'ingestion avec les logs d'événements

S’applique à : connecteurs SaaS connecteurs de base de données

Découvrez comment utiliser les Logs d'événements pour surveiller la progression des passerelles d'ingestion en temps réel. La passerelle émet des métriques par table pour les phases de snapshot et de change data capture (CDC), afin que vous puissiez suivre l'état du pipeline, identifier les tables bloquées et créer des alertes et des tableaux de bord à partir des données d'événements structurées.

Les événements de progression vous permettent de :

• Suivez le nombre de lignes et d'octets ingérés par table sans attendre l'achèvement du pipeline.
• Surveillez la progression de l’instantané et le temps estimé jusqu’à l’achèvement des chargements initiaux importants.
• Mesurez la latence de CDC du commit source au volume Unity Catalog pour chaque table.
• Confirmer que le pipeline est en cours d’exécution, même lorsque aucune modification de données ne se produit.

Fonctionnement des événements de progression

La passerelle émet les types d’événements suivants toutes les cinq minutes pour chaque table de votre pipeline. Les premiers événements apparaissent dans les cinq minutes qui suivent le démarrage du pipeline.

• flow_progress: Émis pour les flux de snapshot et de CDC. Rapporte les lignes upsertées, les lignes supprimées (CDC uniquement) et les octets de sortie depuis l'événement précédent. Pour les flux CDC, rapporte également des métriques de latence.
• operation_progress: Émis pour les flux de snapshot uniquement. Indique le pourcentage d’achèvement du snapshot et le temps estimé restant jusqu’à ce que le snapshot se termine.

Les événements sont émis pour chaque table, y compris les tables sans changement depuis l'événement précédent. Une table qui signale zéro ligne est inactive. Une table qui ne signale aucun événement n'a pas été sélectionnée par le pipeline.

Les deux types d'événements utilisent le niveau de log METRICS. Vous pouvez les query à partir de la table de logs d'événements en utilisant SQL.

Accéder aux événements de progression

Les événements de progression sont stockés dans la table des Logs d'événements. Pour y accéder :

1. Accédez à votre passerelle dans le Databricks Workspace.
2. Cliquez sur l'onglet Event Logs pour afficher les événements dans l'interface utilisateur.
3. Query la table des logs d'événements directement à l'aide de SQL pour une analyse détaillée.

Interroger la table du log des événements

Pour interroger flow_progress événements pour les compteurs de lignes et d'octets :

SQL

SELECT
  timestamp,
  CONCAT(origin.catalog_name, '.', origin.schema_name, '.', origin.dataset_name) AS table_name,
  details:flow_progress:metrics:num_upserted_rows::bigint AS rows_upserted,
  COALESCE(details:flow_progress:metrics:num_deleted_rows::bigint, 0) AS rows_deleted,
  details:flow_progress:metrics:num_output_bytes::bigint AS output_bytes,
  CASE
    WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'snapshot'
    WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'cdc'
    ELSE 'unknown'
  END AS ingestion_phase
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
  AND level = 'METRICS'
  AND origin.pipeline_type = 'INGESTION_GATEWAY'
ORDER BY timestamp DESC

To query operation_progress events for snapshot progress percentage:

SQL

SELECT
  timestamp,
  origin.flow_name AS flow_name,
  details:operation_progress:status::string AS status,
  details:operation_progress:progress_percent::double AS progress_pct
FROM event_log('<pipeline-id>')
WHERE event_type = 'operation_progress'
  AND level = 'METRICS'
  AND origin.pipeline_type = 'INGESTION_GATEWAY'
ORDER BY timestamp DESC

Remplacez <pipeline-id> par l'ID de votre passerelle.

Comprendre la structure d'événement

Les exemples suivants montrent la structure JSON pour chaque type d'événement :

Structure d'événement de progression de flux d'instantané

JSON

{
  "id": "01234567-89ab-cdef-0123-456789abcdef",
  "timestamp": "2025-10-14T13:33:14.175Z",
  "level": "METRICS",
  "event_type": "flow_progress",
  "origin": {
    "pipeline_type": "INGESTION_GATEWAY",
    "pipeline_name": "MyPipeline",
    "dataset_name": "customers",
    "catalog_name": "main",
    "schema_name": "sales",
    "flow_name": "main.sales.customers_snapshot_flow",
    "ingestion_source_type": "SQLSERVER"
  },
  "message": "Completed a streaming update of 'main.sales.customers_snapshot_flow'.",
  "details": {
    "flow_progress": {
      "status": "RUNNING",
      "metrics": {
        "num_upserted_rows": 7512704,
        "num_deleted_rows": null,
        "num_output_bytes": 458752000
      }
    }
  },
  "maturity_level": "STABLE"
}

Structure d'événement de progression du flux CDC

JSON

{
  "id": "01234567-89ab-cdef-0123-456789abcdef",
  "timestamp": "2025-10-14T13:33:57.426Z",
  "level": "METRICS",
  "event_type": "flow_progress",
  "origin": {
    "pipeline_type": "INGESTION_GATEWAY",
    "pipeline_name": "MyPipeline",
    "dataset_name": "customers",
    "catalog_name": "main",
    "schema_name": "sales",
    "flow_name": "main.sales.customers_cdc_flow",
    "ingestion_source_type": "SQLSERVER"
  },
  "message": "Completed a streaming update of 'main.sales.customers_cdc_flow'.",
  "details": {
    "flow_progress": {
      "status": "RUNNING",
      "metrics": {
        "num_upserted_rows": 25,
        "num_deleted_rows": 3,
        "num_output_bytes": 18432
      },
      "streaming_metrics": {
        "discovery_latency_ms": 12450,
        "batch_processing_time_ms": 8100,
        "event_time": {
          "max": "2025-10-14T13:33:45.000Z"
        }
      }
    }
  },
  "maturity_level": "STABLE"
}

Structure d'événement de progression d'Opération Snapshot

JSON

{
  "id": "01234567-89ab-cdef-0123-456789abcdef",
  "timestamp": "2025-10-14T13:33:14.175Z",
  "level": "METRICS",
  "event_type": "operation_progress",
  "origin": {
    "pipeline_type": "INGESTION_GATEWAY",
    "pipeline_name": "MyPipeline",
    "dataset_name": "customers",
    "catalog_name": "main",
    "schema_name": "sales",
    "flow_name": "main.sales.customers_snapshot_flow",
    "ingestion_source_type": "SQLSERVER"
  },
  "message": "Snapshot in progress for 'main.sales.customers'.",
  "details": {
    "operation_progress": {
      "type": "CDC_SNAPSHOT",
      "status": "IN_PROGRESS",
      "duration_ms": 3600000,
      "progress_percent": 65.5,
      "estimated_completion_ms": 1885000,
      "cdc_snapshot": {
        "target_table_name": "main.sales.customers",
        "snapshot_timestamp": 1737542400000,
        "snapshot_reason": "NEW_TABLE"
      }
    }
  },
  "maturity_level": "STABLE"
}

Champs d'événement

Le tableau suivant décrit les champs clés des événements en cours :

Champ	Type	Description
event_type	Chaîne	Soit flow_progress (compteurs de lignes et d'octets pour l'instantané et la CDC), soit operation_progress (pourcentage de progression de l'instantané).
level	Chaîne	Toujours METRICS.
timestamp	Chaîne	Timestamp ISO 8601 lorsque l'événement a été généré.
origin.pipeline_type	Chaîne	Toujours INGESTION_GATEWAY.
origin.pipeline_name	Chaîne	Nom de la passerelle.
origin.dataset_name	Chaîne	Nom de la table ingérée.
origin.catalog_name	Chaîne	Nom du catalogue Unity Catalog.
origin.schema_name	Chaîne	Nom du schéma Unity Catalog.
origin.flow_name	Chaîne	Identifiant de flux qui indique la phase d'ingestion. Format : {catalog}.{schema}.{table}_snapshot_flow pour le chargement initial ou {catalog}.{schema}.{table}_cdc_flow pour les modifications incrémentielles.
origin.ingestion_source_type	Chaîne	Type de base de données source (par exemple, SQLSERVER, MYSQL, POSTGRESQL, ORACLE).
details:flow_progress:status	Chaîne	Statut de flux actuel, généralement RUNNING.
details:flow_progress:metrics:num_upserted_rows	Entier	Nombre de lignes insérées ou mises à jour depuis le dernier événement. Métrique Delta. Voir Compteurs de lignes et d'octets.
details:flow_progress:metrics:num_deleted_rows	Entier	Nombre de lignes supprimées depuis le dernier événement. Métrique Delta. null pour les flux d’instantanés.
details:flow_progress:metrics:num_output_bytes	Entier	Nombre d'octets compressés upload sur un volume depuis le dernier événement. Métrique Delta. Non rapporté pour les snapshots de connecteur basés sur une query. Voir Snapshots de connecteur basés sur une query.
details:flow_progress:streaming_metrics:discovery_latency_ms	Entier	Temps en millisecondes entre le changement de source à event_time:max et le moment où cet événement a été émis. Flux CDC uniquement. Peut être null si la source n'expose pas les horodatages de modification.
details:flow_progress:streaming_metrics:batch_processing_time_ms	Entier	Temps en millisecondes que la passerelle a passé à lire et à upload le dernier batch de modifications. N'inclut pas le délai côté source. Flux CDC uniquement. Peut être null jusqu'à ce que le premier batch soit terminé.
details:flow_progress:streaming_metrics:event_time:max	Chaîne	Timestamp ISO 8601 de la modification de source la plus récente que la passerelle a lue pour cette table. Flux CDC seulement.
details:operation_progress:type	Chaîne	Type d'opération. CDC_SNAPSHOT pour les opérations d'instantané.
details:operation_progress:status	Chaîne	Statut actuel des opérations. IN_PROGRESS pendant l'exécution de l'instantané, COMPLETED lorsqu'il est terminé. Les autres valeurs incluent STARTED, CANCELED et FAILED. Les passerelles exécutant un instantané de connecteur basé sur des query signalent également SCHEDULED lorsqu'une table est en file d'attente. Voir Instantanés de connecteur basés sur des query.
details:operation_progress:duration_ms	Entier	Temps total écoulé de l'opération en millisecondes.
details:operation_progress:progress_percent	Double	Pourcentage d'achèvement de l'instantané (0.0 - 100.0). Cumulatif. Voir Progression et ETA de l'instantané.
details:operation_progress:estimated_completion_ms	Entier	Temps estimé restant en millisecondes avant la fin de l'instantané. Voir Progression et ETA de l'instantané.
details:operation_progress:cdc_snapshot:target_table_name	Chaîne	Nom entièrement qualifié de la table faisant l'objet d'une prise d'instantané.
maturity_level	Chaîne	Toujours STABLE.

Champ	Type	Description
event_type	Chaîne	Soit flow_progress (compteurs de lignes et d'octets pour l'instantané et la CDC), soit operation_progress (pourcentage de progression de l'instantané).
level	Chaîne	Toujours METRICS.
timestamp	Chaîne	Timestamp ISO 8601 lorsque l'événement a été généré.
origin.pipeline_type	Chaîne	Toujours INGESTION_GATEWAY.
origin.pipeline_name	Chaîne	Nom de la passerelle.
origin.dataset_name	Chaîne	Nom de la table ingérée.
origin.catalog_name	Chaîne	Nom du catalogue Unity Catalog.
origin.schema_name	Chaîne	Nom du schéma Unity Catalog.
origin.flow_name	Chaîne	Identifiant de flux qui indique la phase d'ingestion. Format : {catalog}.{schema}.{table}_snapshot_flow pour le chargement initial ou {catalog}.{schema}.{table}_cdc_flow pour les modifications incrémentielles.
origin.ingestion_source_type	Chaîne	Type de base de données source (par exemple, SQLSERVER, MYSQL, POSTGRESQL, ORACLE).
details:flow_progress:status	Chaîne	Statut de flux actuel, généralement RUNNING.
details:flow_progress:metrics:num_upserted_rows	Entier	Nombre de lignes insérées ou mises à jour depuis le dernier événement. Métrique Delta. Voir Compteurs de lignes et d'octets.
details:flow_progress:metrics:num_deleted_rows	Entier	Nombre de lignes supprimées depuis le dernier événement. Métrique Delta. null pour les flux d’instantanés.
details:flow_progress:metrics:num_output_bytes	Entier	Nombre d'octets compressés upload sur un volume depuis le dernier événement. Métrique Delta. Non rapporté pour les snapshots de connecteur basés sur une query. Voir Snapshots de connecteur basés sur une query.
details:flow_progress:streaming_metrics:discovery_latency_ms	Entier	Temps en millisecondes entre le changement de source à event_time:max et le moment où cet événement a été émis. Flux CDC uniquement. Peut être null si la source n'expose pas les horodatages de modification.
details:flow_progress:streaming_metrics:batch_processing_time_ms	Entier	Temps en millisecondes que la passerelle a passé à lire et à upload le dernier batch de modifications. N'inclut pas le délai côté source. Flux CDC uniquement. Peut être null jusqu'à ce que le premier batch soit terminé.
details:flow_progress:streaming_metrics:event_time:max	Chaîne	Timestamp ISO 8601 de la modification de source la plus récente que la passerelle a lue pour cette table. Flux CDC seulement.
details:operation_progress:type	Chaîne	Type d'opération. CDC_SNAPSHOT pour les opérations d'instantané.
details:operation_progress:status	Chaîne	Statut actuel des opérations. IN_PROGRESS pendant l'exécution de l'instantané, COMPLETED lorsqu'il est terminé. Les autres valeurs incluent STARTED, CANCELED et FAILED. Les passerelles exécutant un instantané de connecteur basé sur des query signalent également SCHEDULED lorsqu'une table est en file d'attente. Voir Instantanés de connecteur basés sur des query.
details:operation_progress:duration_ms	Entier	Temps total écoulé de l'opération en millisecondes.
details:operation_progress:progress_percent	Double	Pourcentage d'achèvement de l'instantané (0.0 - 100.0). Cumulatif. Voir Progression et ETA de l'instantané.
details:operation_progress:estimated_completion_ms	Entier	Temps estimé restant en millisecondes avant la fin de l'instantané. Voir Progression et ETA de l'instantané.
details:operation_progress:cdc_snapshot:target_table_name	Chaîne	Nom entièrement qualifié de la table faisant l'objet d'une prise d'instantané.
maturity_level	Chaîne	Toujours STABLE.

Interpréter les métriques de progression

Compteurs de lignes et d’octets

S'applique à num_upserted_rows, num_deleted_rows et num_output_bytes.

• Chaque valeur correspond à une variation. Cela représente les changements depuis l’événement précédent, et non un total cumulé. Pour calculer les totaux, additionnez les valeurs de tous les événements. Voir les query de comptage des lignes.
• Les événements sont émis même lorsqu’aucune donnée ne change. Les valeurs zéro servent de signaux de disponibilité qui confirment que le pipeline est en cours d’exécution.
• num_deleted_rows est null pour les flux d'instantanés car les instantanés ne produisent pas de suppressions.
• Les compteurs sont Reset lors du redémarrage, du refresh ou de la reprise du pipeline. Les événements émis avant le redémarrage restent dans le log d'événements.

Progression du snapshot et ETA

S’applique à progress_percent et estimated_completion_ms.

• progress_percent est cumulatif. Il s'accumule de 0.0 à 100.0 pendant la durée de vie d'un instantané.
• progress_percent est calculé comme étant (completed_chunks / total_chunks) × 100, il s'agit donc d'une approximation, et non d'un pourcentage exact au niveau de la ligne. Les tables qui ne sont pas divisées en plusieurs morceaux (généralement des tables plus petites) sautent de 0.0 directement à 100.0. Les grandes tables divisées en plusieurs morceaux sont mises à jour progressivement à mesure que chaque morceau est terminé.
• Un statut COMPLETED rapporte progress_percent = 100.0.
• Un redémarrage de pipeline peut réinitialiser les métriques de progression, y compris progress_percent.
• estimated_completion_ms reflète l'estimation au moment où l'événement a été émis. Elle diminue au fur et à mesure que l'instantané progresse et atteint 0 une fois terminé. Elle peut augmenter brièvement si la progression ralentit, et est null jusqu'à ce que suffisamment de données aient été traitées pour produire une estimation.

Latence CDC

S’applique à event_time:max, discovery_latency_ms et batch_processing_time_ms. Ces métriques sont rapportées pour les flux CDC uniquement.

• Chaque valeur reflète l'état de la métrique au moment où l'événement a été émis.
• Il s’agit des latences de la passerelle d’ingestion. Elles mesurent le chemin de la base de données source à travers la passerelle jusqu’au volume Unity Catalog. Elles n’incluent pas la latence en aval du volume Unity Catalog vers la table de destination, qui est signalée séparément dans le journal des événements du pipeline d’ingestion.
• Pour estimer le décalage côté source, soustrayez batch_processing_time_ms de discovery_latency_ms.
• discovery_latency_ms et batch_processing_time_ms sont rapportés comme 0 si le résultat devait être négatif.

La relation entre ces métriques vous indique où la latence est concentrée sur le chemin de la passerelle :

Modèle	Ce que cela signifie
discovery_latency_ms et batch_processing_time_ms petits	Le CDC est en cours d'exécution et à jour.
discovery_latency_ms élevée, batch_processing_time_ms faible	Décalage côté source. La passerelle lit rapidement. Les commit sont arrivés en retard de la source (délai de réplication, arriéré des Logs source, transactions source de longue durée).
Les deux sont élevés	Délai côté passerelle. Le pipeline d'upload est le goulot d'étranglement. Vérifiez les ressources de compute de la passerelle et le chemin réseau vers le volume Unity Catalog.
discovery_latency_ms portant sur une seule table	Problème de source par table (DDL en cours, contention de verrou, emplacement de réplication bloqué, changement de schéma).
discovery_latency_ms en progression sur chaque table	Problème à l’échelle de la passerelle (Ressources, connectivité du volume, accumulation du journal CDC source affectant toutes les captures).
event_time:max ne progressant pas sur plusieurs émissions	La source est inactive, ou la passerelle a perdu la connectivité à la base de données source. Vérifiez l'activation du CDC source et les logs de la passerelle.

Modèle	Ce que cela signifie
discovery_latency_ms et batch_processing_time_ms petits	Le CDC est en cours d'exécution et à jour.
discovery_latency_ms élevée, batch_processing_time_ms faible	Décalage côté source. La passerelle lit rapidement. Les commit sont arrivés en retard de la source (délai de réplication, arriéré des Logs source, transactions source de longue durée).
Les deux sont élevés	Délai côté passerelle. Le pipeline d'upload est le goulot d'étranglement. Vérifiez les ressources de compute de la passerelle et le chemin réseau vers le volume Unity Catalog.
discovery_latency_ms portant sur une seule table	Problème de source par table (DDL en cours, contention de verrou, emplacement de réplication bloqué, changement de schéma).
discovery_latency_ms en progression sur chaque table	Problème à l’échelle de la passerelle (Ressources, connectivité du volume, accumulation du journal CDC source affectant toutes les captures).
event_time:max ne progressant pas sur plusieurs émissions	La source est inactive, ou la passerelle a perdu la connectivité à la base de données source. Vérifiez l'activation du CDC source et les logs de la passerelle.

Instantanés de connecteur basés sur les query

Lorsque la passerelle exécute l’instantané initial à l’aide de la méthode de connecteur basée sur les requêtes (QBC), elle émet les mêmes métriques d’instantané décrites sur cette page : flow_progress événements pour le nombre de lignes, et operation_progress événements pour le pourcentage de progression et la durée restante estimée (estimated_completion_ms).

Ces métriques proviennent uniquement de la passerelle d’ingestion. Les pipelines de connecteur basés sur des query qui s’exécutent sans passerelle (pipelines qui lisent la source directement via une connexion) ne les émettent pas ; leurs logs d’événements ne contiennent donc aucun événement de progression d’instantané.

Un instantané QBC diffère des instantanés de passerelle (hérités) des manières suivantes :

• Les octets de sortie ne sont pas rapportés. flow_progress événements rapportent le nombre de lignes (num_upserted_rows), mais pas num_output_bytes. Les requêtes qui lisent les octets de sortie, telles que celles des requêtes d'octets de sortie, ne renvoient aucune donnée pour un instantané QBC. Utilisez plutôt la métrique de nombre de lignes pour suivre le volume.
• Les tables signalent un statut SCHEDULED. Chaque table signale SCHEDULED pendant qu’elle est en file d’attente, avant que son instantané ne start. Les instantanés de la passerelle (ancienne version) ne signalent pas de statut de mise en file d’attente.
• Certains champs operation_progress ne sont pas renseignés. Le champ duration_ms et le champ cdc_snapshot.snapshot_reason ne sont pas définis pour un instantané QBC.

Pour surveiller un instantané QBC, utilisez les requêtes de progression d’instantané dans Snapshot progress queries, y compris la requête Snapshot ETA per table pour le temps estimé restant.

Ajustez les requêtes pour la façon dont un instantané QBC signale les tables en file d'attente. Les requêtes de cette section identifient les tables en file d'attente par un pourcentage de progression de 0, mais un instantané QBC signale les tables en file d'attente avec un statut SCHEDULED et aucun pourcentage de progression. Pour les classer ou les compter correctement, modifiez ces conditions afin de vérifier le statut :

• Remplacer une condition progress_pct = 0 par status = 'SCHEDULED'.
• Pour inclure les tables en file d’attente dans la requête Snapshot ETA per table, remplacez son filtre status = 'IN_PROGRESS' par status IN ('IN_PROGRESS', 'SCHEDULED').

Exemples de query

Les requêtes d'exemple suivantes montrent comment surveiller votre passerelle à l'aide du nombre de lignes, des octets de sortie, de la progression et de l'ETA des instantanés, ainsi que des métriques de latence de la CDC. Remplacez <pipeline-id> par votre ID de passerelle dans chaque query.

astuce

Le Notebook Moniteur de progression de la passerelle d'ingestion contient toutes les requêtes de cette section. Importez le Notebook dans le workspace où votre passerelle s'exécute et spécifiez votre ID de passerelle. Exécutez le Notebook pour inspecter la progression de l'instantané, la latence de la CDC et le nombre de lignes. Vous pouvez également ajuster les thresholds de SLA pour correspondre à vos exigences de monitoring.

Queries pour le nombre de lignes

Volume par table (dernières 24 heures)

Total des opérations d'upsert et de suppression pour chaque table au cours des dernières 24 heures, la phase d'ingestion étant classée à partir du suffixe du nom du flux. Utilisez ceci comme titre de tableau de bord pour voir quelles tables ont déplacé le plus de données.

SQL

WITH row_events AS (
  SELECT
    origin.flow_name AS flow_name,
    CASE
      WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'SNAPSHOT'
      WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'CDC'
      ELSE 'OTHER'
    END AS phase,
    details:flow_progress:metrics:num_upserted_rows::bigint AS upserts,
    details:flow_progress:metrics:num_deleted_rows::bigint AS deletes
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'flow_progress'
    AND level = 'METRICS'
    AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
)
SELECT
  flow_name,
  phase,
  SUM(upserts) AS rows_upserted_24h,
  SUM(COALESCE(deletes, 0)) AS rows_deleted_24h,
  SUM(upserts) + SUM(COALESCE(deletes, 0)) AS total_rows_moved_24h
FROM row_events
GROUP BY flow_name, phase
ORDER BY total_rows_moved_24h DESC

Identifier les tables silencieuses ou bloquées

Les tables émettant des événements, mais ne signalant aucune mise à jour ni aucune suppression depuis les 60 dernières minutes, sont candidates au statut « bloqué ». Vérifiez si la source doit être modifiée. Les tables CDC qui sont réellement inactives (par exemple, pendant la nuit) apparaissent également ici.

SQL

WITH recent_window AS (
  SELECT
    origin.flow_name AS flow_name,
    CASE
      WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'SNAPSHOT'
      WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'CDC'
      ELSE 'OTHER'
    END AS phase,
    COUNT(*) AS emissions_in_window,
    SUM(details:flow_progress:metrics:num_upserted_rows::bigint) AS upserts_in_window,
    SUM(COALESCE(details:flow_progress:metrics:num_deleted_rows::bigint, 0)) AS deletes_in_window,
    MAX(timestamp) AS last_event
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'flow_progress'
    AND level = 'METRICS'
    AND timestamp >= current_timestamp() - INTERVAL 60 MINUTES
  GROUP BY origin.flow_name
)
SELECT
  flow_name,
  phase,
  emissions_in_window,
  upserts_in_window,
  deletes_in_window,
  last_event,
  ROUND(TIMESTAMPDIFF(MINUTE, last_event, current_timestamp()), 0) AS minutes_since_last_event
FROM recent_window
WHERE upserts_in_window = 0
  AND deletes_in_window = 0
ORDER BY minutes_since_last_event DESC

Chronologie par table avec totaux cumulés

Historique complet événement par événement pour un seul flux au cours des dernières 24 heures, avec des totaux de lignes cumulés. Remplacez <flow-pattern> par un modèle SQL LIKE (par exemple, '%customers%_cdc_flow').

SQL

SELECT
  origin.flow_name AS flow_name,
  origin.update_id AS update_id,
  timestamp,
  details:flow_progress:metrics:num_upserted_rows::bigint AS upserts_this_period,
  COALESCE(details:flow_progress:metrics:num_deleted_rows::bigint, 0) AS deletes_this_period,
  SUM(details:flow_progress:metrics:num_upserted_rows::bigint)
    OVER (PARTITION BY origin.flow_name, origin.update_id
          ORDER BY timestamp
          ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS cumulative_upserts_this_run,
  SUM(COALESCE(details:flow_progress:metrics:num_deleted_rows::bigint, 0))
    OVER (PARTITION BY origin.flow_name, origin.update_id
          ORDER BY timestamp
          ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS cumulative_deletes_this_run
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
  AND level = 'METRICS'
  AND origin.flow_name LIKE '<flow-pattern>'
  AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
ORDER BY timestamp

Requêtes en octets de sortie

Octets par table (dernières 24 heures)

Total d'octets transférés vers un volume par table au cours des dernières 24 heures, avec des unités MB et GB conviviales. Trier par ordre décroissant pour voir les tables au volume le plus élevé.

SQL

WITH byte_events AS (
  SELECT
    origin.flow_name AS flow_name,
    CASE
      WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'SNAPSHOT'
      WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'CDC'
      ELSE 'OTHER'
    END AS phase,
    details:flow_progress:metrics:num_output_bytes::bigint AS output_bytes
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'flow_progress'
    AND level = 'METRICS'
    AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
)
SELECT
  flow_name,
  phase,
  SUM(output_bytes) AS bytes_24h,
  ROUND(SUM(output_bytes) / 1024.0 / 1024.0, 2) AS mb_24h,
  ROUND(SUM(output_bytes) / 1024.0 / 1024.0 / 1024.0, 3) AS gb_24h
FROM byte_events
GROUP BY flow_name, phase
ORDER BY bytes_24h DESC

Tendance du throughput (Mo par minute)

Série chronologique par minute des octets upload sur tous les flux au cours des dernières 24 heures. Rendu sous forme de graphique linéaire pour repérer les modèles de throughput et les blocages.

SQL

SELECT
  DATE_TRUNC('MINUTE', timestamp) AS ts_minute,
  origin.flow_name AS flow_name,
  CASE
    WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'SNAPSHOT'
    WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'CDC'
    ELSE 'OTHER'
  END AS phase,
  ROUND(SUM(details:flow_progress:metrics:num_output_bytes::bigint) / 1024.0 / 1024.0, 2) AS mb_per_minute
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
  AND level = 'METRICS'
  AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
GROUP BY DATE_TRUNC('MINUTE', timestamp), origin.flow_name
ORDER BY origin.flow_name, ts_minute

Octets moyens par ligne (table large ou détecteur LOB)

Combine num_output_bytes avec les nombres de lignes pour compute la moyenne d'octets par ligne et par table. Les valeurs élevées indiquent généralement des tables LOB ou à schéma large qui influent sur le coût du throughput. Utile pour la planification de la capacité et la révision du schéma.

SQL

WITH joined AS (
  SELECT
    origin.flow_name AS flow_name,
    CASE
      WHEN origin.flow_name LIKE '%_snapshot_flow' THEN 'SNAPSHOT'
      WHEN origin.flow_name LIKE '%_cdc_flow' THEN 'CDC'
      ELSE 'OTHER'
    END AS phase,
    SUM(details:flow_progress:metrics:num_output_bytes::bigint) AS total_bytes,
    SUM(details:flow_progress:metrics:num_upserted_rows::bigint) AS total_upserts,
    SUM(COALESCE(details:flow_progress:metrics:num_deleted_rows::bigint, 0)) AS total_deletes
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'flow_progress'
    AND level = 'METRICS'
    AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
  GROUP BY origin.flow_name
)
SELECT
  flow_name,
  phase,
  total_upserts + total_deletes AS total_rows,
  ROUND(total_bytes / 1024.0 / 1024.0, 2) AS total_mb,
  ROUND(total_bytes / NULLIF(total_upserts + total_deletes, 0), 0) AS avg_bytes_per_row
FROM joined
WHERE total_bytes > 0
ORDER BY avg_bytes_per_row DESC

Instantanés des query de progression

Progression globale de l'instantané

La query suivante renvoie un résumé sur une seule ligne du nombre de tables terminées, en cours ou en file d'attente. Les résultats reflètent le dernier état signalé par table sur toutes les mises à jour de pipeline dans la fenêtre de rétention des logs d'événements. Ainsi, les tables terminées lors d'une mise à jour antérieure continuent de compter pour tables_completed après un refresh ou un redémarrage.

SQL

WITH latest_per_table AS (
  SELECT
    origin.flow_name AS flow_name,
    details:operation_progress:status::string AS status,
    details:operation_progress:progress_percent::double AS progress_pct,
    ROW_NUMBER() OVER (PARTITION BY origin.flow_name ORDER BY timestamp DESC) AS rn
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'operation_progress'
    AND level = 'METRICS'
    AND origin.flow_name IS NOT NULL
    AND origin.pipeline_type = 'INGESTION_GATEWAY'
)
SELECT
  COUNT(*) AS total_tables,
  SUM(CASE WHEN status = 'COMPLETED' THEN 1 ELSE 0 END) AS tables_completed,
  SUM(CASE WHEN status = 'IN_PROGRESS' AND progress_pct > 0 AND progress_pct < 100 THEN 1 ELSE 0 END) AS tables_in_progress,
  SUM(CASE WHEN progress_pct = 0 THEN 1 ELSE 0 END) AS tables_not_started,
  ROUND(AVG(progress_pct), 2) AS overall_progress_pct,
  ROUND(SUM(CASE WHEN status = 'COMPLETED' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 1) AS pct_tables_done
FROM latest_per_table
WHERE rn = 1

Tableau d'état des instantanés par table

La query suivante renvoie chaque table du pipeline avec son dernier statut signalé et son pourcentage de progression. Les résultats incluent les tables complétées lors des mises à jour précédentes, de sorte qu'aucune table n'apparaît comme des lignes manquantes.

SQL

WITH table_status AS (
  SELECT
    origin.flow_name AS flow_name,
    details:operation_progress:status::string AS status,
    details:operation_progress:progress_percent::double AS progress_pct,
    timestamp,
    ROW_NUMBER() OVER (PARTITION BY origin.flow_name ORDER BY timestamp DESC) AS rn
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'operation_progress'
    AND level = 'METRICS'
    AND origin.flow_name IS NOT NULL
    AND origin.pipeline_type = 'INGESTION_GATEWAY'
)
SELECT
  flow_name,
  status,
  ROUND(progress_pct, 2) AS progress_pct,
  timestamp AS last_update,
  CASE
    WHEN status = 'COMPLETED' THEN 'Done'
    WHEN progress_pct = 0 THEN 'Queued'
    ELSE 'Active'
  END AS phase
FROM table_status
WHERE rn = 1
ORDER BY
  CASE status WHEN 'IN_PROGRESS' THEN 0 WHEN 'COMPLETED' THEN 1 ELSE 2 END,
  progress_pct ASC

Nombre de lignes et d’octets chargés lors de la mise à jour actuelle

La requête suivante combine le pourcentage de progression, les lignes mises à jour et les octets upload pour chaque table dans l'exécution instantanée actuelle.

SQL

WITH latest_update AS (
  SELECT origin.update_id AS update_id
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'create_update'
  ORDER BY timestamp DESC LIMIT 1
),
progress AS (
  SELECT
    origin.flow_name AS flow_name,
    details:operation_progress:progress_percent::double AS progress_pct,
    details:operation_progress:status::string AS status,
    ROW_NUMBER() OVER (PARTITION BY origin.flow_name ORDER BY timestamp DESC) AS rn
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'operation_progress'
    AND level = 'METRICS'
    AND origin.flow_name IS NOT NULL
    AND origin.pipeline_type = 'INGESTION_GATEWAY'
    AND origin.update_id = (SELECT update_id FROM latest_update)
),
volume AS (
  SELECT
    origin.flow_name AS flow_name,
    SUM(details:flow_progress:metrics:num_upserted_rows::bigint) AS rows_loaded,
    SUM(details:flow_progress:metrics:num_output_bytes::bigint) AS bytes_loaded
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'flow_progress'
    AND level = 'METRICS'
    AND origin.flow_name IS NOT NULL
    AND origin.pipeline_type = 'INGESTION_GATEWAY'
    AND origin.flow_name LIKE '%_snapshot_flow'
    AND origin.update_id = (SELECT update_id FROM latest_update)
  GROUP BY origin.flow_name
)
SELECT
  p.flow_name,
  p.status,
  ROUND(p.progress_pct, 2) AS progress_pct,
  v.rows_loaded,
  ROUND(v.bytes_loaded / 1024.0 / 1024.0, 2) AS mb_loaded,
  ROUND(v.bytes_loaded / 1024.0 / 1024.0 / 1024.0, 3) AS gb_loaded
FROM progress p
LEFT JOIN volume v ON p.flow_name = v.flow_name
WHERE p.rn = 1
ORDER BY p.progress_pct ASC

Détection de snapshots bloqués

La query suivante renvoie les tables d'instantanés dont le progress_percent n'a pas changé au cours des 30 dernières minutes. Utilisez-le pour identifier les instantanés bloqués mais toujours actifs.

SQL

WITH latest_update AS (
  SELECT origin.update_id AS update_id
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'create_update'
  ORDER BY timestamp DESC LIMIT 1
),
recent AS (
  SELECT
    origin.flow_name AS flow_name,
    details:operation_progress:progress_percent::double AS progress_pct,
    details:operation_progress:status::string AS status,
    timestamp
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'operation_progress'
    AND level = 'METRICS'
    AND origin.flow_name IS NOT NULL
    AND origin.pipeline_type = 'INGESTION_GATEWAY'
    AND origin.update_id = (SELECT update_id FROM latest_update)
    AND timestamp >= current_timestamp() - INTERVAL 30 MINUTES
)
SELECT
  flow_name,
  ROUND(MIN(progress_pct), 2) AS min_pct_30min,
  ROUND(MAX(progress_pct), 2) AS max_pct_30min,
  ROUND(MAX(progress_pct) - MIN(progress_pct), 2) AS pct_change_30min,
  COUNT(*) AS events_in_window,
  MAX(timestamp) AS last_event_ts
FROM recent
WHERE status = 'IN_PROGRESS'
GROUP BY flow_name
HAVING MAX(progress_pct) - MIN(progress_pct) = 0
   AND MAX(progress_pct) < 100
ORDER BY max_pct_30min ASC

ETA de l'instantané par table

La query suivante renvoie le pourcentage de progression actuel et le temps estimé jusqu'à la fin pour chaque table de l'exécution de l'instantané actif. Seules les tables ayant un statut IN_PROGRESS sont renvoyées. Pour inclure les tables terminées et en file d'attente, supprimez le filtre status = 'IN_PROGRESS'.

SQL

WITH latest_update AS (
  SELECT origin.update_id AS update_id
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'create_update'
  ORDER BY timestamp DESC LIMIT 1
),
latest_per_flow AS (
  SELECT
    origin.flow_name AS flow_name,
    details:operation_progress:status::string AS status,
    details:operation_progress:progress_percent::double AS progress_pct,
    details:operation_progress:estimated_completion_ms::bigint AS eta_ms,
    timestamp,
    ROW_NUMBER() OVER (PARTITION BY origin.flow_name ORDER BY timestamp DESC) AS rn
  FROM event_log('<pipeline-id>')
  WHERE event_type = 'operation_progress'
    AND level = 'METRICS'
    AND origin.flow_name IS NOT NULL
    AND origin.pipeline_type = 'INGESTION_GATEWAY'
    AND origin.update_id = (SELECT update_id FROM latest_update)
)
SELECT
  flow_name,
  status,
  ROUND(progress_pct, 2) AS progress_pct,
  eta_ms,
  ROUND(eta_ms / 1000.0 / 60.0, 1) AS eta_minutes,
  ROUND(eta_ms / 1000.0 / 3600.0, 2) AS eta_hours,
  timestamp AS last_update
FROM latest_per_flow
WHERE rn = 1
  AND status = 'IN_PROGRESS'
ORDER BY eta_ms DESC NULLS LAST

Querys de latence CDC

Pour interpréter les valeurs que ces query retournent, consultez Latence de la CDC.

Actualité de la CDC par table

La query suivante renvoie les 30 dernières minutes d'événements de fraîcheur CDC pour chaque table CDC. Avec des événements émis toutes les cinq minutes, cela produit environ six lignes par table. Les résultats sont triés par discovery_latency_ms afin que les tables les plus décalées apparaissent en premier.

SQL

SELECT
  origin.flow_name AS flow_name,
  details:flow_progress:streaming_metrics:event_time:max::string AS latest_source_commit_seen,
  details:flow_progress:streaming_metrics:discovery_latency_ms::bigint AS discovery_latency_ms,
  details:flow_progress:streaming_metrics:batch_processing_time_ms::bigint AS batch_processing_time_ms,
  timestamp AS emission_ts
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
  AND level = 'METRICS'
  AND origin.flow_name IS NOT NULL
  AND origin.pipeline_type = 'INGESTION_GATEWAY'
  AND origin.flow_name LIKE '%_cdc_flow'
  AND details:flow_progress:streaming_metrics IS NOT NULL
  AND timestamp >= current_timestamp() - INTERVAL 30 MINUTES
ORDER BY discovery_latency_ms DESC NULLS LAST, flow_name, emission_ts DESC

Série temporelle de latence CDC

La query suivante renvoie la moyenne horaire de discovery_latency_ms et batch_processing_time_ms pour chaque flux CDC au cours des dernières 24 heures. Utilisez-le pour identifier les moments où l'actualité s'est dégradée. Pour les pipelines avec de nombreux flux CDC, filtrez par nom de flux pour limiter l'ensemble des résultats.

SQL

SELECT
  DATE_TRUNC('HOUR', timestamp) AS ts_hour,
  origin.flow_name AS flow_name,
  ROUND(AVG(details:flow_progress:streaming_metrics:discovery_latency_ms::bigint) / 1000.0, 2) AS avg_discovery_latency_sec,
  ROUND(AVG(details:flow_progress:streaming_metrics:batch_processing_time_ms::bigint) / 1000.0, 2) AS avg_batch_processing_sec
FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
  AND level = 'METRICS'
  AND origin.flow_name IS NOT NULL
  AND origin.pipeline_type = 'INGESTION_GATEWAY'
  AND origin.flow_name LIKE '%_cdc_flow'
  AND details:flow_progress:streaming_metrics 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

Dépannage

Aucun événement de progression n'apparaît

Si vous ne voyez pas les événements de progression dans les Logs d'événements :

1. Patientez au moins cinq minutes après le start du pipeline.
2. Vérifiez que le pipeline est en cours d'exécution et en ingestion.
3. Vérifiez que vos filtres de query sur level = 'METRICS' et sur le type d’événement correct (flow_progress ou operation_progress).
4. Si des événements manquent toujours ou s’ils arrivent à un intervalle différent de cinq minutes, contactez le support Databricks.

Les métriques affichent zéro après le redémarrage du pipeline

Un redémarrage, un refresh ou une reprise de pipeline peut Reset les métriques de progression. La passerelle start immédiatement à rapporter de nouvelles valeurs, et les événements émis avant le redémarrage restent dans le Logs des événements. Voir Interpréter les métriques de progression.

Métriques manquantes pour certaines tables

Si certaines tables n'affichent pas d'événements de progression :

1. Confirmez que la table est incluse dans la configuration de la passerelle et qu’elle n’est pas exclue par un filtre.
2. Pour la phase CDC, assurez-vous que la table source a la CDC ou le suivi des modifications activé.
3. Notez que les événements operation_progress sont émis uniquement pour les flux d'instantanés. Les flux CDC n'émettent pas d'événements operation_progress car la CDC n'a pas de notion d'achèvement.

Les champs ETA de l'instantané ou latence CDC sont null

Certains champs sont null jusqu'à ce que la passerelle dispose de suffisamment d'informations pour les signaler. Pour savoir quand chaque champ est null, voir Champs d'événement et Interpréter les métriques de progression.

Ressources supplémentaires

• Log des événements du pipeline
• Schéma du log des événements du pipeline