Monitorare le pipeline DLT

Questo articolo descrive l'uso di funzionalità predefinite di monitoraggio e osservabilità per le pipeline DLT. Queste funzionalità supportano attività come:

• Osservare l'avanzamento e lo stato degli aggiornamenti della pipeline. Consulta Quali dettagli della pipeline sono disponibili nell'interfaccia utente?
• Avvisi sugli eventi della pipeline, come il successo o l'insuccesso degli aggiornamenti della pipeline. Vedi Aggiungi le notifiche di posta elettronica agli eventi della pipeline.
• Visualizzazione delle metriche per le origini di streaming come Apache Kafka e Auto Loader (anteprima pubblica). Vedere Visualizzare le metriche di streaming.
• Estrazione di informazioni dettagliate sugli aggiornamenti della pipeline, ad esempio derivazione dei dati, metriche sulla qualità dei dati e utilizzo delle risorse. Vedi Che cos'è il registro eventi DLT?.
• Definizione di azioni personalizzate da eseguire quando si verificano eventi specifici. Vedi Definire il monitoraggio personalizzato delle pipeline DLT con gli hook degli eventi.

Per esaminare le prestazioni delle query e diagnosticarle, vedere Cronologia delle query per le pipeline DLT. Questa funzionalità è disponibile in anteprima pubblica.

Aggiungere notifiche di posta elettronica per gli eventi della pipeline

È possibile configurare uno o più indirizzi di posta elettronica per ricevere notifiche quando si verifica quanto segue:

• Un aggiornamento della pipeline viene completato con successo.
• Un aggiornamento della pipeline non riesce, con un errore recuperabile o non recuperabile. Selezionare questa opzione per ricevere una notifica per tutti gli errori della pipeline.
• Un aggiornamento della pipeline fallisce con un errore non recuperabile (fatale). Selezionare questa opzione per ricevere una notifica solo quando si verifica un errore non riprovabile.
• Un singolo flusso di dati ha esito negativo.

Per configurare le notifiche e-mail quando si crea o modifica una pipeline:

1. Fare clic su Aggiungi notifica.
2. Immettere uno o più indirizzi di posta elettronica per ricevere notifiche.
3. Fare clic sulla casella di controllo per ogni tipo di notifica da inviare agli indirizzi di posta elettronica configurati.
4. Fare clic su Aggiungi notifica.

Quali dettagli della pipeline sono disponibili nell'interfaccia utente?

Il grafico della pipeline viene visualizzato non appena un aggiornamento di una pipeline è stato avviato correttamente. Le frecce rappresentano le dipendenze tra set di dati nella pipeline. Per impostazione predefinita, la pagina dei dettagli della pipeline mostra l'aggiornamento più recente per la tabella, ma è possibile selezionare gli aggiornamenti meno recenti da un menu a discesa.

I dettagli includono l'ID pipeline, il codice sorgente, il costo di calcolo, l'edizione del prodotto e il canale configurato per la pipeline.

Per visualizzare una visualizzazione tabulare dei set di dati, fare clic sulla scheda List. La visualizzazione List consente di visualizzare tutti i set di dati nella pipeline rappresentati come riga in una tabella ed è utile quando il DAG della pipeline è troppo grande per visualizzare nella visualizzazione Graph. È possibile controllare i set di dati visualizzati nella tabella usando più filtri, ad esempio nome del set di dati, tipo e stato. Per tornare alla visualizzazione DAG, fare clic su Graph.

L'utente eseguito come è il proprietario della pipeline e gli aggiornamenti della pipeline vengono eseguiti con le autorizzazioni di questo utente. Per modificare l'utente run as, fare clic su Autorizzazioni e modificare il proprietario della pipeline.

Come è possibile visualizzare i dettagli del set di dati?

Facendo clic su un set di dati nell'elenco del grafico o del set di dati della pipeline vengono visualizzati i dettagli sul set di dati. I dettagli includono lo schema del set di dati, le metriche sulla qualità dei dati e un collegamento al codice sorgente che definisce il set di dati.

Visualizzare la cronologia degli aggiornamenti

Per visualizzare la cronologia e lo stato degli aggiornamenti della pipeline, fare clic sul menu a discesa Cronologia aggiornamenti nella barra superiore.

Selezionare l'aggiornamento nel menu a discesa per visualizzare il grafico, i dettagli e gli eventi per un aggiornamento. Per tornare all'aggiornamento più recente, fare clic su Mostra l'aggiornamento più recente.

Visualizzare le metriche di streaming

È possibile visualizzare le metriche di streaming dalle origini dati supportate da Spark Structured Streaming, come Apache Kafka, Amazon Kinesis, Auto Loader e tabelle Delta, per ogni flusso di streaming nella pipeline DLT. Le metriche vengono visualizzate come grafici nel riquadro destro dell'interfaccia utente e includono secondi di arretrato, byte di arretrato, record di arretrato e file di arretrato. I grafici visualizzano il valore massimo aggregato per minuto e un tooltip mostra i valori massimi quando passi il mouse sul grafico. I dati sono limitati alle ultime 48 ore dall'ora corrente.

Le tabelle nella tua pipeline con metriche di streaming disponibili mostrano l'icona "Grafico DLT" quando si visualizza il DAG della pipeline nell'interfaccia utente in modalità "Grafico". Per visualizzare le metriche di streaming, fare clic sull'icona per visualizzare il grafico delle metriche di streaming nella scheda flussi nel riquadro destro. È anche possibile applicare un filtro per visualizzare solo le tabelle con metriche di streaming facendo clic su List e quindi facendo clic su Has streaming metrics.You can also apply a filter to view only tables with streaming metrics by click List, then click Has streaming metrics.

Ogni origine di streaming supporta solo metriche specifiche. Le metriche non supportate da un'origine di streaming non sono disponibili per la visualizzazione nell'interfaccia utente. La tabella seguente illustra le metriche disponibili per le origini di streaming supportate:

Che cos'è il registro eventi DLT?

Il registro eventi DLT contiene tutte le informazioni correlate a una pipeline, inclusi i log di controllo, i controlli di qualità dei dati, lo stato della pipeline e la derivazione dei dati. È possibile usare il registro eventi per tenere traccia, comprendere e monitorare lo stato delle pipeline di dati.

È possibile visualizzare le voci del registro eventi nell'interfaccia utente DLT, nell'API DLT o eseguendo direttamente una query sul registro eventi. Questa sezione è incentrata sull'esecuzione di query direttamente sul registro eventi.

È anche possibile definire azioni personalizzate da eseguire quando gli eventi vengono registrati, ad esempio l'invio di avvisi, utilizzando gli hook degli eventi.

schema del registro eventi

Nella tabella seguente viene descritto lo schema del registro eventi. Alcuni di questi campi contengono dati JSON che richiedono l'analisi per eseguire alcune query, ad esempio il campo details. Azure Databricks supporta l'operatore : per analizzare i campi JSON. Vedere l'operatore : (segno due punti).

Eseguire una query sul registro eventi

Per impostazione predefinita, DLT scrive il registro eventi in una tabella Delta nascosta nel catalogo predefinito e nello schema configurati per la pipeline. Mentre è nascosta, la tabella può essere comunque interrogata da tutti gli utenti con privilegi sufficientemente elevati. Per impostazione predefinita, solo il proprietario della pipeline può eseguire query sulla tabella del registro eventi.

Per impostazione predefinita, il nome del registro eventi nascosto viene formattato come event_log_{pipeline_id}, dove l'ID della pipeline è l'UUID assegnato dal sistema con i trattini sostituiti da caratteri di sottolineatura.

È possibile interagire con la configurazione JSON per pubblicare il registro eventi. Quando si pubblica un registro eventi, si specifica il nome del registro eventi e facoltativamente è possibile specificare un catalogo e uno schema, come nell'esempio seguente:

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

Il percorso del registro eventi funge anche da percorso dello schema per le query di Auto Loader nella pipeline. Databricks consiglia di creare una vista sulla tabella del log eventi prima di modificare i privilegi, perché alcune impostazioni di calcolo potrebbero consentire agli utenti di accedere ai metadati dello schema se la tabella del log eventi viene condivisa direttamente. La sintassi di esempio seguente crea una vista in una tabella del log eventi e viene usata nelle query del log eventi di esempio incluse in questo articolo.

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

Ogni istanza di un'esecuzione della pipeline viene denominata aggiornamento . Spesso si vogliono estrarre informazioni per l'aggiornamento più recente. Eseguire la query seguente per trovare l'identificatore per l'aggiornamento più recente e salvarlo nella visualizzazione temporanea latest_update. Questa vista viene usata nelle query del log eventi di esempio incluse in questo articolo:

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;

Nel Catalogo Unity, le viste supportano le query di streaming. Nell'esempio seguente viene usato Structured Streaming per eseguire query su una vista definita sopra una tabella del log eventi:

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

Il proprietario della pipeline può pubblicare il registro eventi come tabella Delta pubblica attivando l'opzione Publish event log to metastore nella sezione avanzata della configurazione della pipeline. Facoltativamente, è possibile specificare un nuovo nome di tabella, un catalogo e uno schema per il registro eventi.

Consultare informazioni sulla provenienza dal registro eventi

Gli eventi contenenti informazioni sulla derivazione hanno il tipo di evento flow_definition. L'oggetto details:flow_definition contiene i output_dataset e input_datasets che definiscono ogni relazione nel grafico.

È possibile usare la query seguente per estrarre i set di dati di input e output per visualizzare le informazioni di derivazione:

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

Qualità dei dati di query dal registro eventi

Se si definiscono le aspettative sui set di dati nella pipeline, le metriche relative alla qualità dei dati vengono archiviate nell'oggetto details:flow_progress.data_quality.expectations. Gli eventi contenenti informazioni sulla qualità dei dati hanno il tipo di evento flow_progress. L'esempio seguente interroga le metriche relative alla qualità dei dati per l'ultimo aggiornamento della 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

Eseguire query sugli eventi di Auto Loader dal registro eventi

DLT genera eventi quando il caricatore automatico elabora i file. Per gli eventi del caricatore automatico, il event_type è operation_progress e il details:operation_progress:type è AUTO_LOADER_LISTING o AUTO_LOADER_BACKFILL. L'oggetto details:operation_progress include anche campi status, duration_ms, auto_loader_details:source_pathe auto_loader_details:num_files_listed.

L'esempio seguente esegue una query sugli eventi di Auto Loader per l'ultimo aggiornamento.

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')

Monitorare il backlog dei dati effettuando una query sul registro eventi

DLT tiene traccia della quantità di dati presenti nel backlog nell'oggetto details:flow_progress.metrics.backlog_bytes. Gli eventi contenenti metriche backlog hanno il tipo di evento flow_progress. Nell'esempio seguente si interrogano le metriche del backlog per l'ultimo aggiornamento della 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

Monitorare gli eventi di scalabilità automatica avanzati dal registro eventi per le pipeline senza abilitare serverless

Per le pipeline DLT che non usano il calcolo serverless, il log eventi acquisisce il ridimensionamento del cluster quando la scalabilità automatica avanzata è abilitata nelle pipeline. Gli eventi contenenti informazioni sulla scalabilità automatica avanzata hanno il tipo di evento autoscale. Le informazioni sulla richiesta di ridimensionamento del cluster vengono archiviate nell'oggetto details:autoscale. Il seguente esempio esegue una query sulle richieste avanzate di ridimensionamento automatico del cluster per l'ultimo aggiornamento della 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

Monitorare l'utilizzo delle risorse di calcolo

cluster_resources gli eventi forniscono metriche sul numero di slot di attività nel cluster, sull'utilizzo di questi slot di attività e sul numero di attività in attesa di essere pianificate.

Quando la scalabilità automatica avanzata è abilitata, gli eventi cluster_resources contengono anche metriche per l'algoritmo di scalabilità automatica, tra cui latest_requested_num_executorse optimal_num_executors. Gli eventi mostrano anche lo stato dell'algoritmo come stati diversi, ad esempio CLUSTER_AT_DESIRED_SIZE, SCALE_UP_IN_PROGRESS_WAITING_FOR_EXECUTORSe BLOCKED_FROM_SCALING_DOWN_BY_CONFIGURATION. Queste informazioni possono essere visualizzate insieme agli eventi di scalabilità automatica per fornire un quadro generale della scalabilità automatica migliorata.

Nell'esempio seguente viene eseguita una query sulla cronologia delle dimensioni della coda di attività per l'ultimo aggiornamento della 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'esempio seguente esegue una query sulla cronologia di utilizzo per l'ultimo aggiornamento della 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'esempio seguente esegue una query sulla cronologia del conteggio dell'executor, accompagnata da metriche disponibili solo per le pipeline di scalabilità automatica avanzate, incluso il numero di executor richiesti dall'algoritmo nella richiesta più recente, il numero ottimale di executor consigliati dall'algoritmo in base alle metriche più recenti e lo stato dell'algoritmo di scalabilità automatica:

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

verificare le pipeline DLT

È possibile usare i record del registro eventi DLT e altri log di controllo di Azure Databricks per ottenere un quadro completo del modo in cui i dati vengono aggiornati in DLT.

DLT usa le credenziali del proprietario della pipeline per eseguire gli aggiornamenti. È possibile modificare le credenziali usate aggiornando il proprietario della pipeline. DLT registra l'utente per le azioni nella pipeline, tra cui la creazione della pipeline, le modifiche alla configurazione e l'attivazione degli aggiornamenti.

Per informazioni di riferimento sugli eventi di controllo di Unity Catalog, vedere eventi del catalogo Unity.

Eseguire query sulle azioni utente nel registro eventi

È possibile usare il registro eventi per controllare gli eventi, ad esempio le azioni dell'utente. Gli eventi contenenti informazioni sulle azioni utente hanno il tipo di evento user_action.

Le informazioni sull'azione vengono archiviate nell'oggetto user_action nel campo details. Usare la query seguente per costruire un log di controllo degli eventi utente. Per creare la vista event_log_raw usata in questa query, vedere Eseguire una query sul registro eventi.

SELECT timestamp, details:user_action:action, details:user_action:user_name FROM event_log_raw WHERE event_type = 'user_action'

informazioni sul runtime di

È possibile visualizzare le informazioni di runtime per un aggiornamento della pipeline, ad esempio la versione di Databricks Runtime per l'aggiornamento:

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