Ereignisse überwachen und abfragen

Jede Pipeline sendet Ereignisse, die Audit-Protokolle, Datenqualitätsprüfungen, den Fortschritt und die Datenherkunft erfassen. Sie können diese Ereignisse aus zwei Quellen abfragen:

• Die pipeline_events Systemtabelle enthält Ereignisse für alle Pipelines über die Arbeitsbereiche einer Region hinweg und ist die empfohlene Methode zur Abfrage von Ereignissen. Es befindet sich in der Beta-Phase.
• Das Ereignislog pro Pipeline ist eine Delta-Tabelle, die Ereignisse für eine einzelne Pipeline enthält.

Du fragst Ereignisse mit Standard-SQL ab. Um Ihnen den Einstieg zu erleichtern, bietet diese Seite außerdem ein Beispiel-Dashboard und gängige Abfragen für die Systemtabelle.

Anforderungen

Um auf diese Systemtabelle zuzugreifen, müssen Benutzer entweder:

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

Beispieldashboard

Dieses Beispiel-Dashboard liest die pipeline_events Systemtabelle, um Pipeline-Aktualisierungen, Flussdurchsatz, Rückstau, Datenqualität und Fehler in jeder Pipeline einer Region zu verfolgen. Filtere jede Seite nach Pipeline, Tabelle, Tag und Zeitbereich.

Importieren des Dashboards

1. Laden Sie die Dashboard-JSON-Datei herunter.
2. Importieren Sie das Dashboard in Ihren Arbeitsbereich. Für Anweisungen siehe Importiere eine Dashboard-Datei.

Überwachungsanfragen

Die folgenden Abfragen vom Dashboard zeigen gängige Anwendungsfälle für Pipeline-Überwachung.

Letzter Fehler pro Pipeline

Diese Abfrage liefert den aktuellsten Fehler für jede Pipeline, die in den letzten 7 Tagen fehlerhaft war, mit der äußersten Ausnahme.

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

Stundenfehlerrate

Diese Abfrage zählt Fehler pro Pipeline und Stunde, damit Sie Spitzen erkennen können.

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

Reihen wurden pro Fluss geändert

Diese Abfrage summiert die Zeilen, die ein Flow pro Stunde angehängt und gelöscht bzw. ein Upsert ausgeführt hat. Jede Metrik ist die Anzahl pro Mikrobatch, sodass die Summe über das Fenster den Durchsatz ergibt.

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

Rückstau pro Fluss

Diese Abfrage liefert die aktuellste Backlog-Lesung pro Flow. Ein COMPLETED-Flow ist auf dem neuesten Stand, sodass sein Backlog als 0 gemeldet wird.

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

Fehlgeschlagene Erwartungen nach Datensatz

Diese Abfrage gibt die Erwartungen zurück, bei denen Datensätze pro Datensatz und pro Update innerhalb des letzten Tages fehlgeschlagen sind. Es basiert auf dem eigenen dataset der Erwartung, das auch dann gefüllt ist, wenn das origin.dataset_name des Ereignisses nicht gefüllt ist.

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