Referenz zur Pipelineereignisse-Systemtabelle

Dieser Artikel ist ein Verweis auf die Systemtabelle, in der pipeline_eventsLakeflow-Pipelines-Ereignisprotokolleinträge für Pipelines in Ihrem Konto aufgezeichnet werden. Jede Zeile ist ein unveränderliches Ereignis aus dem Pipelineereignisprotokoll, das Erfassen von Lebenszyklusübergängen, der Ablauffortschritt, Datenqualitätsmetriken, Fehler, Clusterressourcen und andere Betriebsdaten in allen Pipelines und Arbeitsbereichen innerhalb einer Region.

Requirements

• Um auf diese Systemtabelle zuzugreifen, müssen Benutzer eine der folgenden Aktionen ausführen:
  • Seien Sie sowohl ein Metastore-Administrator als auch ein Kontoadministrator, oder ...
  • Verfügen Sie über USE- und SELECT-Berechtigungen für die Systemschemata. Siehe Gewähren des Zugriffs auf Systemtabellen.

Pipeline-Ereignis-Dashboard

Ein vorgefertigtes Dashboard visualisiert die Gesundheit der Pipeline, den Durchsatz, den Rückstau, die Datenqualität und Fehler aus dieser Systemtabelle.

Um das Dashboard zu importieren und zu verwenden, siehe Überwachen und Abfrageereignisse.

Verfügbare Pipeline-Ereignistabellen

Detaillierte Schemareferenz

Tabellenschema für Pipelineereignisse

Die Pipelineereignissetabelle ist nur anzufügen. Jede Zeile zeichnet ein einzelnes Ereignis auf, das von einer Pipelineaktualisierung zum Zeitpunkt der Ausgabe ausgegeben wird, und Zeilen werden nie geändert oder gelöscht.

Welche Felder in einer Zeile aufgefüllt werden, hängt vom Ereignistyp ab. error, update_idund viele origin.* Unterfelder werden nur für Ereignisse festgelegt, bei denen sie angewendet werden, und die Struktur des details Felds variiert ebenfalls nach event_type.

Verwenden Sie diese Tabelle, um historische Pipelineaktivitäten abzufragen, Um Warnungen zu Pipelinefehlern zu erstellen und das Pipelineverhalten mit anderen Lakeflow-Systemtabellen zu korrelieren.

Tabellenpfad: system.lakeflow_pipeline_events_preview.pipeline_events

Primärschlüssel: (account_id, pipeline_event_id)

Ursprungsstrukturfelder

Fehlerstrukturfelder

Detailsfeld

Die details Spalte ist ein VARIANT, und die darin enthaltenen Felder hängen von der event_type. Die unter den einzelnen Ereignistypen verfügbaren Felder finden Sie unter Pipeline-Ereignisprotokollschema. Verwenden Sie die variant_get Funktions- oder Punktsyntax, um geschachtelte Werte zu lesen. In den folgenden Beispielabfragen finden Sie typische Zugriffsmuster.

Der event_type Schlüssel umhüllt die Nutzlast. Zum Beispiel sind die Metriken eines Ereignisses flow_progress bei $.flow_progress.metrics, nicht $.metrics. Fügen Sie den Ereignistypschlüssel in jeden Pfad ein.

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

Beispielabfragen

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

Allgemeine Verknüpfungsmuster

Mit der pipelines Tabelle verbinden, um nach Pipelinenamen zu filtern

Die pipelines Tabelle ist eine langsam ändernde Dimension (SCD2). Nehmen Sie die neueste Version jeder Pipeline vor dem Beitritt vor.

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

Teilnehmen mit pipeline_update_timeline am 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

Benachrichtigungen einrichten

Sie können Benachrichtigungen über pipeline_events die Verwendung von Databricks SQL-Warnungen erstellen. Schreiben Sie eine SQL-Abfrage anhand pipeline_events (optional verknüpft mit anderen Lakeflow-Systemtabellen), planen Sie sie in einem SQL-Lager, und konfigurieren Sie ein Benachrichtigungsziel (E-Mail, Slack, Webhook, PagerDuty).

Einige nützliche Ausgangspunkte:

Alarm, wenn in den letzten N Minuten keine Ereignisse für eine laufende Pipeline eingetreten sind

Verwenden Sie diese Eigenschaft, um hängen gebliebene oder im Hintergrund fehlerhafte Pipelines zu erkennen.

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

Warnung, wenn der Backlog für einen bestimmten Fluss zu hoch ist

Backlog wird bei Ereignissen als flow_progressbacklog_bytesgemeldet, und für Dateiquellen auch als backlog_files. Auslösen, wenn die aktuellste Messung eine Schwelle überschreitet (zum Beispiel 100 MB unverarbeitete Arbeit). Nicht jede Quelle meldet jede Metrik, also filtere nach der Kennzahl, die deine Quelle befüllt.

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

Um zu sehen, welche Quelle dahinterliegt, $.flow_progress.metrics.source_metrics gibt es ein Array von Pro-Source-Lesungen, jede mit source_name neben dem von dieser Quelle backlog_bytes, backlog_records oder backlog_files.

Warnung über Datenqualitätsverluste in einer Pipeline

Jedes flow_progress Ereignis meldet die Anzahl der Zeilen, die nach EXPECT … DROP Erwartungen verworfen wurden. Summieren Sie diese pro Datensatz über ein Aktualisierungsfenster und warnen Sie, wenn die Gesamtzahl eine Schwelle überschreitet.

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

Alarmieren, wenn ein Flow zu wenige Zeilen verarbeitet

flow_progress Ereignisse werden als Menge pro Mikro-Batch gemeldet metrics.num_output_rows , sodass die Summe der Ereignisse in einem Fenster die über diesem Fenster geschriebenen Zeilen ergibt. Erstellen Sie eine Warnung, wenn der Durchsatz unter einen erwarteten Unterboden fällt. Zum Beispiel kann ein Flow, der normalerweise Tausende von Zeilen pro Stunde schreibt, aber nahezu null erzeugt, auf eine falsch konfigurierte Quelle hinweisen.

Diese Abfrage meldet nur Flüche, die ein flow_progress Ereignis mit einer Zeilenanzahl im Fenster ausgelöst haben. Ein vollständig gestoppter Fluss sendet keine Ereignisse, also kombinieren Sie diese Warnung mit der oben genannten Warnung für fehlende Ereignisse.

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

Tipps für Produktionswarnungen

• Filtern Sie nach pipeline_id (und workspace_id wenn Sie Warnungen pro Arbeitsbereich verwalten), sodass jede Warnung auf einen bestimmten Bereich anstatt auf das gesamte Konto ausgerichtet ist.
• Wählen Sie einen Auswertungsrhythmen aus, der der Vertraulichkeit der Warnung entspricht. Verwenden Sie ein kurzes Intervall (z. B. alle 5 Minuten) für Schnelle-Fail-Signale und ein längeres Intervall (z. B. stündlich) für Backlog- und Datenqualitätstrends. Eine "Trigger when query returns more than 0 rows" condition works for most cases.

Referenzwerte

Levelwerte

Reifestufenwerte

Ereignistypwerte

Das event_type Feld ist eine Aufzählung. Der vollständige Wertesatz: