Informazioni di riferimento sulla tabella di sistema degli eventi della pipeline

Questo articolo è un riferimento per la tabella di pipeline_events sistema, che registra le voci del registro eventi di Lakeflow pipelines per le pipeline nell'account. Ogni riga è un evento non modificabile dal registro eventi della pipeline, acquisisce transizioni del ciclo di vita, avanzamento del flusso, metriche di qualità dei dati, errori, risorse del cluster e altri dati operativi in tutte le pipeline e aree di lavoro all'interno di un'area.

Requisiti

• Per accedere a questa tabella di sistema, gli utenti devono:
  • 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 degli eventi della pipeline

Un dashboard preassemblato visualizza lo stato della pipeline, la produttività, il backlog, la qualità dei dati e gli errori da questa tabella di sistema.

Per importare e utilizzare la dashboard, vedi Monitorare e interrogare eventi.

Tabelle degli eventi della pipeline disponibili

Informazioni di riferimento dettagliate sullo schema

Schema della tabella degli eventi della pipeline

La tabella degli eventi della pipeline è di sola accodamento. Ogni riga registra un singolo evento generato da un aggiornamento della pipeline al momento dell'emissione e le righe non vengono mai modificate o eliminate sul posto.

I campi popolati in una riga dipendono dal tipo di evento. error, update_ide molti origin.* sottocampi vengono impostati solo sugli eventi in cui si applicano e la struttura del details campo varia anche in base a event_type.

Usare questa tabella per eseguire query sull'attività della pipeline cronologica, creare avvisi in caso di errori della pipeline e correlare il comportamento della pipeline con altre tabelle di sistema Lakeflow.

Percorso tabella: system.lakeflow_pipeline_events_preview.pipeline_events

Chiave primaria: (account_id, pipeline_event_id)

Campi di struct di origine

Campi di struct di errore

Campo dettagli

La details colonna è un VARIANToggetto e i campi che contiene dipendono da event_type. Per i campi disponibili in ogni tipo di evento, vedere Schema del registro eventi della pipeline. Usare la sintassi della variant_get funzione o del punto per leggere i valori annidati. Vedere le query di esempio seguenti per i modelli di accesso tipici.

La event_type chiave avvolge il carico utile. Ad esempio, le metriche di un flow_progress evento sono a $.flow_progress.metrics, non $.metrics. Includi la chiave di tipo evento in ogni percorso.

-- Using variant_get (lets you cast to a specific type)
SELECT
  pipeline_id,
  event_time,
  variant_get(details, '$.flow_progress.status', 'STRING')                       AS flow_status,
  variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT')      AS rows_written,
  variant_get(details, '$.flow_progress.metrics.backlog_bytes', 'BIGINT')        AS backlog_bytes
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR

-- Using dot syntax (returns VARIANT, cast when needed)
SELECT
  pipeline_id,
  event_time,
  details:flow_progress.status::STRING               AS flow_status,
  details:flow_progress.metrics.num_output_rows::BIGINT AS rows_written
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR

Grandi eventi

Quando un evento è troppo grande per essere memorizzato per intero, la tabella memorizza una copia abbreviata del suo details. Un evento accorciato ha un truncation campo in $.truncation, accanto alla chiave di tipo evento. Il suo truncated_fields array elenca ogni campo abbreviato con field_name, ad esempio details. Altri eventi non hanno un truncation campo.

Interrogazioni di esempio

-- Flow throughput for a specific pipeline
SELECT
  origin.flow_name,
  date_trunc('HOUR', event_time) AS hour,
  SUM(variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT')) AS rows_written
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  pipeline_id = '<your-pipeline-id>'
  AND 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_written DESC

-- The latest error for each pipeline that has errored in the last 7 days, with the outermost exception.
-- The exception chain is ordered with the root cause last, so read element -1 for the root cause.
-- On many errors only the first element carries error_class and sql_state.
SELECT
  workspace_id,
  pipeline_id,
  event_time,
  event_type,
  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

-- Data quality: failed expectations by dataset, per update, in the last 1 day
SELECT
  pipeline_id,
  update_id,
  origin.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,
  origin.dataset_name,
  expectation.name
HAVING
  SUM(expectation.failed_records) > 0
ORDER BY
  failed_records DESC

Modelli di join comuni

Creare un join con la tabella per filtrare in base al nome della pipelines pipeline

La pipelines tabella è una dimensione a modifica lenta (SCD2). Prima di creare un join, prendere la versione più recente di ogni pipeline.

WITH latest_pipelines AS (
  SELECT *
  FROM system.lakeflow.pipelines
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id ORDER BY change_time DESC) = 1
)
SELECT
  p.name AS pipeline_name,
  e.event_time,
  e.event_type,
  e.level,
  e.message
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events e
JOIN
  latest_pipelines p
  ON e.workspace_id = p.workspace_id
  AND e.pipeline_id = p.pipeline_id
WHERE
  e.level = 'ERROR'
  AND e.event_time >= current_timestamp() - INTERVAL 24 HOURS
ORDER BY
  e.event_time DESC

Partecipa con pipeline_update_timeline in update_id

WITH updates AS (
  SELECT
    workspace_id,
    pipeline_id,
    update_id,
    MIN(period_start_time) AS update_start,
    MAX(CASE WHEN result_state IS NOT NULL THEN period_end_time END) AS update_end
  FROM
    system.lakeflow.pipeline_update_timeline
  WHERE
    pipeline_id = '<your-pipeline-id>'
  GROUP BY
    workspace_id,
    pipeline_id,
    update_id
)
SELECT
  u.update_start,
  u.update_end,
  e.event_time,
  e.event_type,
  e.level,
  e.message
FROM
  updates u
JOIN
  system.lakeflow_pipeline_events_preview.pipeline_events e
  ON e.workspace_id = u.workspace_id
  AND e.update_id = u.update_id
ORDER BY
  u.update_start DESC,
  e.event_time ASC

Impostazione degli avvisi

È possibile creare avvisi per pipeline_events l'uso degli avvisi SQL di Databricks. Scrivere una query SQL su pipeline_events (facoltativamente unita ad altre tabelle di sistema Lakeflow), pianificarla in un'istanza di SQL Warehouse e configurare una destinazione di notifica (posta elettronica, Slack, webhook, PagerDuty).

Alcuni punti di partenza utili:

Avviso quando non sono arrivati eventi per una pipeline in corso negli ultimi N minuti

Usare questa opzione per rilevare pipeline bloccate o con errori invisibile all'utente.

-- Returns one row per pipeline with a currently-running update that has emitted no event
-- in the last 30 minutes. The alert can trigger when this query returns any rows.
WITH running_pipelines AS (
  SELECT
    workspace_id,
    pipeline_id
  FROM
    system.lakeflow.pipeline_update_timeline
  QUALIFY
    ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id ORDER BY period_end_time DESC) = 1
    AND result_state IS NULL
),
last_event AS (
  SELECT
    workspace_id,
    pipeline_id,
    MAX(event_time) AS last_event_time
  FROM
    system.lakeflow_pipeline_events_preview.pipeline_events
  GROUP BY
    workspace_id,
    pipeline_id
)
SELECT
  r.workspace_id,
  r.pipeline_id,
  le.last_event_time
FROM
  running_pipelines r
  LEFT JOIN last_event le USING (workspace_id, pipeline_id)
WHERE
  le.last_event_time IS NULL
  OR le.last_event_time < current_timestamp() - INTERVAL 30 MINUTES

Avvisare quando il backlog per un flusso specifico è troppo elevato

L'arretrato viene riportato sugli flow_progress eventi come backlog_bytes, e per le fonti dei file anche come backlog_files. Si attiva quando la lettura più recente supera una soglia (ad esempio, 100 MB di lavoro non elaborato). Non tutte le fonti riportano ogni metrica, quindi filtra su quella che la tua fonte popola.

-- Returns the most recent backlog reading per flow for a given pipeline.
-- The alert can trigger when backlog_bytes exceeds the threshold for any flow.
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
    pipeline_id = '<your-pipeline-id>'
    AND 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 > 100000000

Per vedere quale sorgente è indietro, $.flow_progress.metrics.source_metrics è una serie di letture per fonte, ciascuna con source_name accanto a backlog_bytesquella di quella fonte , backlog_records o backlog_files.

Avviso per le cadute di qualità dei dati in una pipeline

Ogni flow_progress evento segnala il numero di righe eliminate dalle EXPECT … DROP aspettative. Sommare questi dati per ogni dataset durante una finestra di aggiornamento e avvisare quando il totale supera una soglia.

-- Returns datasets where more than 100 rows were dropped by expectations, per update, in the last hour.
SELECT
  pipeline_id,
  update_id,
  origin.dataset_name,
  SUM(variant_get(details, '$.flow_progress.data_quality.dropped_records', 'BIGINT')) AS dropped_records
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR
GROUP BY
  pipeline_id,
  update_id,
  origin.dataset_name
HAVING
  SUM(variant_get(details, '$.flow_progress.data_quality.dropped_records', 'BIGINT')) > 100

Avvisa quando un flusso elabora troppe righe

flow_progress gli eventi riportano metrics.num_output_rows come conteggio per micro-lotto, quindi sommando gli eventi in una finestra si ottengono le righe scritte sopra quella finestra. Crea un avviso quando la produttività scende sotto un minimo previsto. Ad esempio, un flusso che normalmente scrive migliaia di righe all'ora ma produce quasi zero può indicare una sorgente mal configurata.

Questa query riporta solo i flussi che hanno emesso un flow_progress evento con un conteggio di righe nella finestra. Un flusso completamente bloccato non emette eventi, quindi abbina questo allarme all'allarme di eventi mancanti sopra.

-- Returns active flows (latest status not terminal) that wrote fewer than 100 rows in the last hour.
-- The alert can trigger when this query returns any rows.
WITH flow_state AS (
  SELECT
    pipeline_id,
    origin.flow_name,
    event_time,
    variant_get(details, '$.flow_progress.status', 'STRING') AS status,
    variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT') AS num_output_rows,
    ROW_NUMBER() OVER (PARTITION BY pipeline_id, origin.flow_name ORDER BY event_time DESC) AS rn
  FROM
    system.lakeflow_pipeline_events_preview.pipeline_events
  WHERE
    pipeline_id = '<your-pipeline-id>'
    AND event_type = 'flow_progress'
    AND event_time >= current_timestamp() - INTERVAL 1 HOUR
),
active_flows AS (
  SELECT
    pipeline_id,
    flow_name
  FROM
    flow_state
  WHERE
    rn = 1
    AND status NOT IN ('COMPLETED', 'FAILED', 'SKIPPED', 'STOPPED', 'EXCLUDED')
)
SELECT
  f.pipeline_id,
  f.flow_name,
  SUM(f.num_output_rows) AS rows_written_last_hour
FROM
  flow_state f
  JOIN active_flows a USING (pipeline_id, flow_name)
WHERE
  f.num_output_rows IS NOT NULL
GROUP BY
  f.pipeline_id,
  f.flow_name
HAVING
  SUM(f.num_output_rows) < 100

Suggerimenti per gli avvisi di produzione

• Filtrare in base pipeline_id a (e workspace_id se si mantengono gli avvisi per area di lavoro) in modo che ogni avviso sia destinato a un ambito specifico anziché all'intero account.
• Scegliere una frequenza di valutazione corrispondente alla sensibilità dell'avviso. Usare un intervallo breve (ad esempio, ogni 5 minuti) per i segnali di errore rapido e un intervallo più lungo (ad esempio, ogni ora) per le tendenze di backlog e qualità dei dati. Una condizione "trigger quando la query restituisce più di 0 righe" funziona per la maggior parte dei casi.

Valori di riferimento

Valori dei livelli

Valori dei livelli di maturità

Valori del tipo di evento

Il event_type campo è un'enumerazione. Set completo di valori: