Monitorare e interrogare gli eventi

Ogni pipeline emette eventi che catturano log di audit, controlli di qualità dei dati, progressi e lineage dei dati. Puoi interrogare questi eventi da due fonti:

• La pipeline_events tabella di sistema contiene eventi per tutte le pipeline tra gli spazi di lavoro di una regione ed è il modo consigliato per interrogare gli eventi. È in Beta.
• Il registro eventi per pipeline è una tabella Delta che contiene eventi per una singola pipeline.

Interroghi gli eventi con SQL standard. Per aiutarti a iniziare, questa pagina fornisce anche un dashboard di esempio e query comuni per la tabella di sistema.

Requisiti

Per accedere a questa tabella di sistema, gli utenti devono o:

• Essere sia un amministratore del metastore che un amministratore dell'account oppure
• Disporre delle autorizzazioni USE e SELECT per gli schemi di sistema. Vedere Concedere l'accesso alle tabelle di sistema.

Dashboard di esempio

Questo esempio di dashboard legge la pipeline_events tabella di sistema per tracciare gli aggiornamenti della pipeline, la velocità di flusso, il backlog, la qualità dei dati e gli errori in ogni pipeline di una regione. Filtra ogni pagina per pipeline, tabella, tag e intervallo temporale.

Importare il dashboard

1. Scarica il file JSON della dashboard.
2. Importare il dashboard nell'area di lavoro. Per le istruzioni, vedi Importa un file dashboard.

Monitoraggio delle query

Le seguenti query dalla dashboard mostrano casi d'uso comuni del monitoraggio delle pipeline.

Errore più recente per pipeline

Questa query restituisce l'errore più recente per ogni pipeline che ha commesso errori negli ultimi 7 giorni, con l'eccezione più esterna.

SELECT
  workspace_id,
  pipeline_id,
  event_time,
  message,
  error.exceptions[0].error_class AS exception_error_class,
  error.exceptions[0].sql_state   AS exception_sql_state
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  level = 'ERROR'
  AND event_time >= current_timestamp() - INTERVAL 7 DAYS
QUALIFY
  ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id ORDER BY event_time DESC) = 1
ORDER BY
  event_time DESC

Tasso di errore orario

Questa query conta gli errori per ogni pipeline, ora per ora, in modo da poter individuare eventuali picchi.

SELECT
  pipeline_id,
  date_trunc('HOUR', event_time) AS hour,
  count(*) AS num_errors
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  level = 'ERROR'
  AND event_time >= current_timestamp() - INTERVAL 7 DAYS
GROUP BY
  pipeline_id,
  date_trunc('HOUR', event_time)
ORDER BY
  hour DESC

Righe cambiate per flusso

Questa query somma, per ogni ora, le righe aggiunte, aggiornate o inserite ed eliminate da un flusso. Ogni metrica è un conteggio per micro-batch, quindi la somma nell'intervallo fornisce il throughput.

SELECT
  origin.flow_name,
  date_trunc('HOUR', event_time) AS hour,
  SUM(
    ifnull(variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT'), 0)
    + ifnull(variant_get(details, '$.flow_progress.metrics.num_upserted_rows', 'BIGINT'), 0)
    + ifnull(variant_get(details, '$.flow_progress.metrics.num_deleted_rows', 'BIGINT'), 0)
  ) AS rows_changed
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 7 DAYS
GROUP BY
  origin.flow_name,
  date_trunc('HOUR', event_time)
ORDER BY
  hour DESC,
  rows_changed DESC

Arretrato per flusso

Questa query restituisce la lettura più recente del backlog per ogni flusso. Un COMPLETED flusso viene recuperato, quindi il suo arretrato viene riportato come 0.

WITH latest_flow_progress AS (
  SELECT
    workspace_id,
    pipeline_id,
    origin.flow_name,
    event_time,
    CASE
      WHEN variant_get(details, '$.flow_progress.status', 'STRING') = 'COMPLETED' THEN 0
      ELSE variant_get(details, '$.flow_progress.metrics.backlog_bytes', 'BIGINT')
    END AS backlog_bytes
  FROM
    system.lakeflow_pipeline_events_preview.pipeline_events
  WHERE
    event_type = 'flow_progress'
    AND event_time >= current_timestamp() - INTERVAL 1 HOUR
    AND (
      variant_get(details, '$.flow_progress.metrics.backlog_bytes', 'BIGINT') IS NOT NULL
      OR variant_get(details, '$.flow_progress.status', 'STRING') = 'COMPLETED'
    )
  QUALIFY
    ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id, origin.flow_name ORDER BY event_time DESC) = 1
)
SELECT *
FROM latest_flow_progress
WHERE backlog_bytes > 0
ORDER BY backlog_bytes DESC

Aspettative non soddisfatte del dataset

Questa query restituisce le aspettative associate ai record non riusciti per dataset, per aggiornamento, nell'ultimo giorno. Si basa sull'aspettativa stessa dataset, che è popolata anche quando quella dell'evento origin.dataset_name non lo è.

SELECT
  pipeline_id,
  update_id,
  coalesce(expectation.dataset, origin.dataset_name) AS dataset_name,
  expectation.name AS expectation_name,
  SUM(expectation.failed_records) AS failed_records
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
  LATERAL VIEW explode(variant_get(details, '$.flow_progress.data_quality.expectations', 'ARRAY<STRUCT<name:STRING,dataset:STRING,passed_records:BIGINT,failed_records:BIGINT>>')) AS expectation
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 DAY
GROUP BY
  pipeline_id,
  update_id,
  coalesce(expectation.dataset, origin.dataset_name),
  expectation.name
HAVING
  SUM(expectation.failed_records) > 0
ORDER BY
  failed_records DESC