Partager via

Surveiller les pipelines déclaratifs Lakeflow

Cet article décrit l’utilisation des fonctionnalités intégrées de surveillance et d’observabilité pour les pipelines déclaratifs Lakeflow. Ces fonctionnalités prennent en charge les tâches telles que celles suivantes :

Pour inspecter et diagnostiquer les performances des requêtes, consultez l’historique des requêtes Access pour les pipelines déclaratifs Lakeflow. Cette fonctionnalité est en version préliminaire publique.

Ajouter des notifications par e-mail pour les événements de pipeline

Vous pouvez configurer une ou plusieurs adresses e-mail pour recevoir des notifications lorsque les opérations suivantes se produisent :

  • Une mise à jour du pipeline s'achève avec succès.
  • Une mise à jour de pipeline échoue, soit à cause d'une erreur réessayable, soit à cause d'une erreur non réessayable. Sélectionnez cette option pour recevoir une notification pour toutes les défaillances du pipeline.
  • Une mise à jour de pipeline échoue avec une erreur non réessayable (fatale). Sélectionnez cette option pour recevoir une notification uniquement lorsqu’une erreur non renouvelable se produit.
  • Un flux de données unique échoue.

Pour configurer les notifications par e-mail lorsque vous créez ou modifiez un pipeline :

  1. Cliquez sur Ajouter une notification.
  2. Entrez une ou plusieurs adresses e-mail pour recevoir des notifications.
  3. Cochez la case pour chaque type de notification à envoyer aux adresses e-mail configurées.
  4. Cliquez sur Ajouter une notification.

Affichage des pipelines dans l’interface utilisateur

Recherchez vos pipelines déclaratifs Lakeflow à partir de l’option Travaux &Pipelines dans la barre latérale de l’espace de travail. La page Travaux et pipelines s’ouvre, où vous pouvez afficher des informations sur chaque travail et pipeline auquel vous avez accès. Cliquez sur le nom d’un pipeline pour ouvrir la page des détails du pipeline.

Utilisation de la liste Travaux et pipelines

Pour afficher la liste des pipelines auquel vous avez accès, cliquez sur l’icône Flux de travail.Travaux & Pipelines dans la barre latérale. L’onglet Travaux &pipelines répertorie des informations sur tous les travaux et pipelines disponibles, tels que le créateur, le déclencheur (le cas échéant) et le résultat des cinq dernières exécutions.

Pour modifier les colonnes affichées dans la liste, cliquez sur l’icône Paramètres de colonne et sélectionnez ou désélectionnez les colonnes.

Important

La liste unifiée Travaux et pipelines est en Préversion publique. Vous pouvez désactiver la fonctionnalité et revenir à l’expérience par défaut en désactivant les travaux et les pipelines : gestion unifiée, recherche et filtrage. Pour plus d’informations, consultez Gérer les préversions d’Azure Databricks .

Vous pouvez filtrer les travaux dans la liste Travaux et pipelines , comme illustré dans la capture d’écran suivante.

Affichage de la liste des pipelines avec annotations.

  1. Recherche de texte : la recherche de mots clés est prise en charge pour les champs Nom et ID . Pour rechercher une étiquette créée avec une clé et une valeur, vous pouvez lancer une recherche par clé, valeur ou clé et valeur. Par exemple, pour une étiquette avec la clé department et la valeur finance, vous pouvez rechercher department ou finance pour trouver les travaux correspondants. Pour effectuer une recherche par la clé et la valeur, entrez la clé et la valeur séparées par un signe deux-points (par exemple, department:finance).
  2. Type : filtrez par travaux, pipelines ou tout. Si vous sélectionnez Pipelines , vous pouvez également filtrer par type de pipeline, qui inclut des pipelines ETL et Ingestion.
  3. Propriétaire : afficher uniquement les travaux que vous possédez.
  4. Favoris : afficher les travaux que vous avez marqués comme favoris.
  5. Balises : utilisez des balises. Pour effectuer une recherche par balise, vous pouvez utiliser le menu déroulant des balises pour filtrer jusqu’à cinq balises en même temps ou utiliser directement la recherche de mots clés.
  6. Exécuter en tant que : Filtrer par jusqu’à deux run as valeurs.

Pour démarrer un travail ou un pipeline, cliquez sur le bouton de lecture Icône de lecture. Pour arrêter un travail ou un pipeline, cliquez sur le bouton Icône d’arrêt . Pour accéder à d'autres actions, cliquez sur l'icône de menu Kebab. Par exemple, vous pouvez supprimer le travail ou le pipeline, ou accéder aux paramètres d’un pipeline à partir de ce menu.

Quels détails du pipeline sont disponibles dans l’interface utilisateur ?

Le graphique de pipeline s’affiche dès qu’une mise à jour d’un pipeline a démarré avec succès. Les flèches représentent les dépendances entre les jeux de données de votre pipeline. Par défaut, la page des détails du pipeline affiche la mise à jour la plus récente pour la table, mais vous pouvez sélectionner des mises à jour plus anciennes dans un menu déroulant.

Les détails incluent l’ID de pipeline, le code source, le coût de calcul, l’édition du produit et le canal configuré pour le pipeline.

Pour afficher une vue tabulaire des jeux de données, cliquez sur l’onglet Liste. La vue Liste vous permet de voir tous les jeux de données de votre pipeline représentés sous forme de ligne dans une table et est utile lorsque votre DAG de pipeline est trop volumineux pour visualiser dans la vue Graphe. Vous pouvez contrôler les jeux de données affichés dans la table à l’aide de plusieurs filtres tels que le nom, le type et l’état du jeu de données. Pour revenir à la visualisation DAG, cliquez sur Graphe.

L’utilisateur Run as est le propriétaire du pipeline, et les mises à jour du pipeline sont exécutées avec les autorisations de cet utilisateur. Pour modifier l’utilisateur run as, cliquez sur Autorisations et modifiez le propriétaire du pipeline.

Comment pouvez-vous afficher les détails du jeu de données ?

Cliquer sur un jeu de données dans le graphique de pipeline ou la liste des jeux de données affiche des détails sur le jeu de données. Les détails incluent le schéma du jeu de données, les métriques de qualité des données et un lien vers le code source définissant le jeu de données.

Consulter l’historique des mises à jour

Pour afficher l’historique et l’état des mises à jour du pipeline, cliquez sur le menu déroulant Historique des mises à jour dans la barre supérieure.

Sélectionnez la mise à jour dans le menu déroulant pour afficher le graphique, les détails et les événements d’une mise à jour. Pour revenir à la dernière mise à jour, cliquez sur Afficher la dernière mise à jour.

Afficher les métriques de diffusion en continu

Important

L’observabilité de streaming pour les pipelines déclaratifs Lakeflow est disponible en aperçu public.

Vous pouvez afficher les métriques de diffusion en continu à partir des sources de données prises en charge par Spark Structured Streaming, comme Apache Kafka, Amazon Kinesis, Auto Loader et les tables Delta, pour chaque flux de streaming dans vos Pipelines Déclaratifs Lakeflow. Les métriques sont affichées sous forme de graphiques dans le volet droit de l’interface utilisateur des pipelines déclaratifs Lakeflow et incluent des secondes de backlog, des octets de backlog, des enregistrements de backlog et des fichiers de backlog. Les graphiques affichent la valeur maximale agrégée par minute et une info-bulle affiche les valeurs maximales lorsque vous pointez sur le graphique. Les données sont limitées aux 48 dernières heures à partir de l’heure actuelle.

Les tables de votre pipeline avec des mesures de streaming disponibles affichent l'icône Icône de graphique DLT lors de l'affichage du DAG du pipeline dans la vue Graph de l'interface utilisateur. Pour afficher les métriques de streaming, cliquez sur l’icône de graphique DLT pour afficher le graphique de métriques de streaming sous l’onglet Flux dans le volet droit. Vous pouvez également appliquer un filtre pour afficher uniquement les tables avec des mesures de diffusion en continu en cliquant sur Liste, puis sur Avec mesures de diffusion en continu.

Chaque source de diffusion en continu prend uniquement en charge des métriques spécifiques. Les métriques non prises en charge par une source de diffusion en continu ne sont pas disponibles pour l’affichage dans l’interface utilisateur. Le tableau suivant présente les métriques disponibles pour les sources de diffusion en continu prises en charge :

Source octets de backlog enregistrements de backlog secondes de backlog fichiers en retard
Kafka ✓ ✓
Cinèse ✓ ✓
Delta ✓ ✓
Chargeur automatique ✓ ✓
Google Pub/Sub ✓ ✓

Qu'est-ce que le journal des événements des pipelines déclaratifs Lakeflow ?

Le journal des événements Pipelines déclaratifs Lakeflow contient toutes les informations relatives à un pipeline, notamment les journaux d’audit, les contrôles de qualité des données, la progression du pipeline et la traçabilité des données. Vous pouvez utiliser le journal des événements pour suivre, comprendre et surveiller l’état de vos pipelines de données.

Vous pouvez afficher les entrées du journal des événements dans l'interface utilisateur des Pipelines Déclaratifs Lakeflow, dans l'API des Pipelines Déclaratifs Lakeflow ou en interrogeant directement le journal des événements. Cette section se concentre sur la façon d’interroger directement le journal des événements.

Vous pouvez également définir des actions personnalisées à exécuter lorsque des événements sont enregistrés, par exemple en envoyant des alertes, avec des hooks d’événements.

Important

Ne supprimez pas le journal des événements ou le catalogue parent ou le schéma où le journal des événements est publié. La suppression du journal des événements peut entraîner l’échec de la mise à jour de votre pipeline pendant les prochaines exécutions.

Schéma du journal des événements

Le tableau suivant décrit le schéma du journal des événements. Certains de ces champs contiennent des données JSON qui nécessitent une analyse pour effectuer des requêtes, comme le champ details. Azure Databricks prend en charge l’opérateur : pour analyser les champs JSON. Consultez : opérateur (signe deux points).

Champ Descriptif
id Identificateur unique de l’enregistrement de journal des événements.
sequence Document JSON contenant les métadonnées d’identification et d’ordre des événements.
origin Document JSON contenant des métadonnées pour l’origine de l’événement, par exemple, le fournisseur de cloud, la région du fournisseur de cloud, user_id, pipeline_id ou pipeline_type pour montrer où le pipeline a été créé, soit DBSQL, soit WORKSPACE.
timestamp Heure à laquelle l’événement a été enregistré.
message Message lisible décrivant l’événement.
level Type d’événement, par exemple INFOWARN, ERROR ou METRICS.
maturity_level Stabilité du schéma d’événement. Les valeurs possibles sont les suivantes :
  • STABLE: le schéma est stable et ne change pas.
  • NULL: le schéma est stable et ne change pas. La valeur peut être NULL si l’enregistrement a été créé avant l’ajout du champ maturity_level (version 2022.37).
  • EVOLVING: le schéma n’est pas stable et peut changer.
  • DEPRECATED: le schéma est déconseillé et le runtime de pipelines déclaratifs Lakeflow peut cesser de produire cet événement à tout moment.
error Détails décrivant l’erreur qui est survenue, le cas échéant.
details Document JSON contenant les détails structurés de l’événement. Il s’agit du champ principal utilisé pour l’analyse des événements.
event_type Type d’événement.

Interroger le journal des événements

Remarque

Cette section décrit le comportement et la syntaxe par défaut de l’utilisation des journaux d’événements pour les pipelines configurés avec le catalogue Unity et le mode de publication par défaut.

Par défaut, Lakeflow Declarative Pipelines écrit le journal des événements dans une table Delta masquée dans le catalogue et le schéma par défaut configurés pour le pipeline. Bien que masquée, la table peut toujours être interrogée par tous les utilisateurs suffisamment privilégiés. Par défaut, seul le propriétaire du pipeline peut interroger la table du journal des événements.

Par défaut, le nom du journal des événements cachés est mis en forme event_log_{pipeline_id}, où l'ID de pipeline est l'UUID assigné par le système, où les tirets sont remplacés par des traits de soulignement.

Vous pouvez interagir avec la configuration JSON pour publier le journal des événements. Lorsque vous publiez un journal des événements, vous spécifiez le nom du journal des événements et pouvez éventuellement spécifier un catalogue et un schéma, comme dans l’exemple suivant :

{
  "id": "ec2a0ff4-d2a5-4c8c-bf1d-d9f12f10e749",
  "name": "billing_pipeline",
  "event_log": {
    "catalog": "catalog_name",
    "schema": "schema_name",
    "name": "event_log_table_name"
  }
}

L’emplacement du journal des événements sert également d’emplacement de schéma pour toutes les requêtes du chargeur automatique dans le pipeline. Databricks recommande de créer une vue sur la table du journal des événements avant de modifier les privilèges, car certains paramètres de calcul peuvent permettre aux utilisateurs d’accéder aux métadonnées de schéma si la table du journal des événements est partagée directement. L’exemple de syntaxe suivant crée une vue sur une table du journal des événements et est utilisé dans l’exemple de requêtes de journal des événements incluses dans cet article.

CREATE VIEW event_log_raw
AS SELECT * FROM catalog_name.schema_name.event_log_table_name;

Chaque instance d’une exécution de pipeline est appelée mise à jour. Vous souhaitez souvent extraire des informations pour la mise à jour la plus récente. Exécutez la requête suivante pour rechercher l’identificateur de la dernière mise à jour et l’enregistrer dans la vue temporaire latest_update. Cette vue est utilisée dans les exemples de requêtes de journal des événements compris dans cet article :

CREATE OR REPLACE TEMP VIEW latest_update AS SELECT origin.update_id AS id FROM event_log_raw WHERE event_type = 'create_update' ORDER BY timestamp DESC LIMIT 1;

Dans Unity Catalog, les vues prennent en charge les requêtes en streaming. L’exemple suivant utilise Structured Streaming pour interroger une vue définie en haut d’une table du journal des événements :

df = spark.readStream.table("event_log_raw")

Le propriétaire du pipeline peut publier le journal des événements en tant que table publique Delta en basculant l’option Publish event log to metastore dans la section Avancé de la configuration du pipeline. Vous pouvez éventuellement spécifier un nouveau nom de table, un catalogue et un schéma pour le journal des événements.

Interroger les informations de traçabilité du journal des événements

Les événements contenant des informations de traçabilité ont le type d’événement flow_definition. L’objet details:flow_definition contient output_dataset et input_datasets qui définissent chaque relation dans le graphe.

Vous pouvez utiliser la requête suivante pour extraire les jeux de données d’entrée et de sortie afin de voir les informations de traçabilité :

SELECT
  details:flow_definition.output_dataset as output_dataset,
  details:flow_definition.input_datasets as input_dataset
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'flow_definition'
  AND
  origin.update_id = latest_update.id
output_dataset input_datasets
customers null
sales_orders_raw null
sales_orders_cleaned ["customers", "sales_orders_raw"]
sales_order_in_la ["sales_orders_cleaned"]

Interroger la qualité des données à partir du journal des événements

Si vous définissez des attentes sur les jeux de données de votre pipeline, les métriques du nombre d’enregistrements passés et ayant échoué sont stockées dans l’objet details:flow_progress.data_quality.expectations . La métrique du nombre d’enregistrements supprimés est stockée dans l’objet details:flow_progress.data_quality . Les événements contenant des informations sur la qualité des données ont le type d’événement flow_progress.

Les métriques de qualité des données peuvent ne pas être disponibles pour certains jeux de données. Consultez les limitations attendues.

Les métriques de qualité des données suivantes sont disponibles :

Unité de mesure Descriptif
dropped_records Nombre d’enregistrements qui ont été supprimés parce qu’ils ont échoué une ou plusieurs attentes.
passed_records Nombre d’enregistrements qui ont passé les critères d’attente.
failed_records Nombre d’enregistrements ayant échoué aux critères d'expectation.

L’exemple suivant interroge les métriques de qualité des données relatives à la dernière mise à jour du pipeline :

SELECT
  row_expectations.dataset as dataset,
  row_expectations.name as expectation,
  SUM(row_expectations.passed_records) as passing_records,
  SUM(row_expectations.failed_records) as failing_records
FROM
  (
    SELECT
      explode(
        from_json(
          details:flow_progress.data_quality.expectations,
          "array<struct<name: string, dataset: string, passed_records: int, failed_records: int>>"
        )
      ) row_expectations
    FROM
      event_log_raw,
      latest_update
    WHERE
      event_type = 'flow_progress'
      AND origin.update_id = latest_update.id
  )
GROUP BY
  row_expectations.dataset,
  row_expectations.name
dataset expectation passing_records failing_records
sales_orders_cleaned valid_order_number 4083 0

Consulter les événements du chargeur automatique à partir du journal des événements

Les pipelines déclaratifs Lakeflow génèrent des événements lorsque le chargeur automatique traite les fichiers. Pour les événements du chargeur automatique, le event_type est operation_progress et le details:operation_progress:type est soit AUTO_LOADER_LISTING soit AUTO_LOADER_BACKFILL. L'objet details:operation_progress inclut également les champs status, duration_ms, auto_loader_details:source_path et auto_loader_details:num_files_listed.

L’exemple suivant interroge les événements du chargeur automatique pour la dernière mise à jour :

SELECT
  timestamp,
  details:operation_progress.status,
  details:operation_progress.type,
  details:operation_progress:auto_loader_details
FROM
  event_log_raw,
  latest_update
WHERE
  event_type like 'operation_progress'
  AND
  origin.update_id = latest.update_id
  AND
  details:operation_progress.type in ('AUTO_LOADER_LISTING', 'AUTO_LOADER_BACKFILL')

Surveiller le backlog de données en interrogeant le journal des événements

Lakeflow Declarative Pipelines suit la quantité de données présentes dans le backlog de l’objet details:flow_progress.metrics.backlog_bytes . Les événements contenant les métriques de backlog ont le type d’événement flow_progress. L’exemple suivant interroge les métriques du backlog relatives à la dernière mise à jour du pipeline :

SELECT
  timestamp,
  Double(details :flow_progress.metrics.backlog_bytes) as backlog
FROM
  event_log_raw,
  latest_update
WHERE
  event_type ='flow_progress'
  AND
  origin.update_id = latest_update.id

Remarque

Les métriques du backlog peuvent ne pas être disponibles en fonction du type de source de données du pipeline et de la version de Databricks Runtime.

Surveiller les événements de mise à l’échelle automatique améliorée à partir du journal des événements pour les pipelines sans fonctionnalité serverless activée

Pour les pipelines déclaratifs Lakeflow qui n’utilisent pas de calcul serverless, le journal des événements capture les redimensionnements de cluster grâce à la mise à l’échelle automatique améliorée activée dans vos pipelines. Les événements contenant des informations sur la mise à l’échelle automatique améliorée ont le type autoscaled’événement . Les informations de demande de redimensionnement de cluster sont stockées dans l’objet details:autoscale. L'exemple suivant interroge les requêtes de redimensionnement du cluster de mise à l'échelle automatique améliorée pour la dernière mise à jour du pipeline :

SELECT
  timestamp,
  Double(
    case
      when details :autoscale.status = 'RESIZING' then details :autoscale.requested_num_executors
      else null
    end
  ) as starting_num_executors,
  Double(
    case
      when details :autoscale.status = 'SUCCEEDED' then details :autoscale.requested_num_executors
      else null
    end
  ) as succeeded_num_executors,
  Double(
    case
      when details :autoscale.status = 'PARTIALLY_SUCCEEDED' then details :autoscale.requested_num_executors
      else null
    end
  ) as partially_succeeded_num_executors,
  Double(
    case
      when details :autoscale.status = 'FAILED' then details :autoscale.requested_num_executors
      else null
    end
  ) as failed_num_executors
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'autoscale'
  AND
  origin.update_id = latest_update.id

Surveiller l’utilisation des ressources de calcul

Des événements cluster_resources fournissent des métriques sur le nombre d’emplacements de tâches dans le cluster, le nombre d’emplacements de tâches utilisés et le nombre de tâches en attente de planification.

Lorsque la mise à l’échelle automatique améliorée est activée, cluster_resources les événements contiennent également des métriques pour l’algorithme de mise à l’échelle automatique, notamment latest_requested_num_executors, et optimal_num_executors. Les événements montrent également l’état de l’algorithme sous la forme d’états différents tels que CLUSTER_AT_DESIRED_SIZE, SCALE_UP_IN_PROGRESS_WAITING_FOR_EXECUTORS et BLOCKED_FROM_SCALING_DOWN_BY_CONFIGURATION. Ces informations peuvent être consultées conjointement avec les événements de mise à l’échelle automatique pour fournir une image globale de la mise à l’échelle automatique améliorée.

L’exemple suivant interroge l’historique de taille de la file d’attente des tâches pour la dernière mise à jour du pipeline :

SELECT
  timestamp,
  Double(details :cluster_resources.avg_num_queued_tasks) as queue_size
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'cluster_resources'
  AND
  origin.update_id = latest_update.id

L’exemple suivant interroge l’historique d’utilisation de la dernière mise à jour du pipeline :

SELECT
  timestamp,
  Double(details :cluster_resources.avg_task_slot_utilization) as utilization
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'cluster_resources'
  AND
  origin.update_id = latest_update.id

L’exemple suivant interroge l’historique du nombre d’exécuteurs, accompagné de métriques disponibles uniquement pour les pipelines de mise à l’échelle automatique améliorés, y compris le nombre d’exécuteurs demandés par l’algorithme dans la dernière requête, le nombre optimal d’exécuteurs recommandés par l’algorithme en fonction des métriques les plus récentes et l’état de l’algorithme de mise à l’échelle automatique :

SELECT
  timestamp,
  Double(details :cluster_resources.num_executors) as current_executors,
  Double(details :cluster_resources.latest_requested_num_executors) as latest_requested_num_executors,
  Double(details :cluster_resources.optimal_num_executors) as optimal_num_executors,
  details :cluster_resources.state as autoscaling_state
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'cluster_resources'
  AND
  origin.update_id = latest_update.id

Auditez les pipelines déclaratifs Lakeflow

Vous pouvez utiliser les enregistrements du journal des événements des pipelines déclaratifs Lakeflow et d’autres journaux d’audit Azure Databricks pour obtenir une image complète de la façon dont les données sont mises à jour dans les pipelines déclaratifs Lakeflow.

Les pipelines déclaratifs Lakeflow utilisent les informations d’identification du propriétaire du pipeline pour exécuter les mises à jour. Vous pouvez modifier les informations d’identification utilisées en mettant à jour le propriétaire du pipeline. Lakeflow Declarative Pipelines enregistre l’utilisateur pour les actions sur le pipeline, notamment la création du pipeline, les modifications apportées à la configuration et le déclenchement des mises à jour.

Consultez Événements Unity Catalog pour obtenir des informations de référence sur les événements d’audit Unity Catalog.

Interroger les actions utilisateur dans le journal des événements

Vous pouvez utiliser le journal des événements pour auditer des événements, comme les actions utilisateur. Les événements contenant des informations sur les actions utilisateur ont le type d’événement user_action.

Les informations relatives à chaque action sont stockées dans l’objet user_action dans le champ details. Utilisez la requête suivante pour créer un journal d’audit des événements utilisateur. Pour créer la event_log_raw vue utilisée dans cette requête, consultez Interroger le journal des événements.

SELECT timestamp, details:user_action:action, details:user_action:user_name FROM event_log_raw WHERE event_type = 'user_action'
timestamp action user_name
2021-05-20T19:36:03.517+0000 START user@company.com
2021-05-20T19:35:59.913+0000 CREATE user@company.com
2021-05-27T00:35:51.971+0000 START user@company.com

Informations sur le runtime

Vous pouvez consulter des informations sur l'environnement d'exécution pour la mise à jour du pipeline, par exemple, la version de Databricks Runtime pour la mise à jour.

SELECT details:create_update:runtime_version:dbr_version FROM event_log_raw WHERE event_type = 'create_update'
dbr_version
11.0