Dokumentacja tabeli systemu zdarzeń potoku

Ten artykuł jest odwołaniem do tabeli systemowej pipeline_events , która rejestruje wpisy dziennika zdarzeń potoków lakeflow dla potoków na twoim koncie. Każdy wiersz jest niezmiennym zdarzeniem z dziennika zdarzeń potoku, przechwytywaniem przejść cyklu życia, postępem przepływu, metrykami jakości danych, błędami, zasobami klastra i innymi danymi operacyjnymi we wszystkich potokach i obszarach roboczych w regionie.

Requirements

• Aby uzyskać dostęp do tej tabeli systemowej, użytkownicy muszą:
  • Bądź zarówno administratorem metadanych, jak i administratorem konta lub
  • Mieć uprawnienia USE i SELECT na schematach systemowych. Zobacz Udzielanie dostępu do tabel systemowych.

Pulpit zdarzeń pipeline

Gotowy dashboard wizualizuje zdrowie potoku, przepustowość, backlog, jakość danych oraz błędy z tej tabeli systemowej.

Aby zaimportować i korzystać z dashboardu, zobacz Monitor i zapytanie o zdarzenia.

Dostępne tabele zdarzeń potoków

Szczegółowe odniesienie do schematu

Schemat tabeli zdarzeń potoku

Tabela zdarzeń potoku jest tylko do dołączania. Każdy wiersz rejestruje pojedyncze zdarzenie emitowane przez aktualizację potoku w momencie jego emisji, a wiersze nigdy nie są modyfikowane ani usuwane.

Pola, które są wypełniane w wierszu, zależą od typu zdarzenia. error, update_idi wiele origin.* pól podrzędnych są ustawiane tylko na zdarzeniach, w których mają zastosowanie, a struktura details pola również różni się w zależności od event_type.

Ta tabela umożliwia wykonywanie zapytań dotyczących historycznych działań potoku, tworzenie alertów dotyczących błędów potoku oraz korelowanie zachowania potoku z innymi tabelami systemowymi usługi Lakeflow.

Ścieżka tabeli: system.lakeflow_pipeline_events_preview.pipeline_events

Klucz podstawowy: (account_id, pipeline_event_id)

Pola struktury początkowej

Pola struktury błędu

Pole szczegółów

Kolumna details to VARIANT, a pola, które zawiera, zależą od .event_type Aby uzyskać pola dostępne w poszczególnych typach zdarzeń, zobacz Schemat dziennika zdarzeń potoku. variant_get Użyj funkcji lub składni kropki, aby odczytać zagnieżdżone wartości. Zapoznaj się z poniższymi przykładowymi zapytaniami, aby uzyskać typowe wzorce dostępu.

Klucz event_type owija ładunek. Na przykład flow_progress metryki zdarzenia są w , a nie $.flow_progress.metrics.$.metrics Umieść klucz typu zdarzenia w każdej ścieżce.

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

Duże wydarzenia

Gdy zdarzenie jest zbyt duże, by go przechowywać w całości, tabela przechowuje skróconą kopię swojego details. Skrócone zdarzenie ma truncation pole w , $.truncationobok klucza typu zdarzenia. Jego tablica wymienia truncated_fields każde skrócone pole o field_name, na przykład details. Inne wydarzenia nie mają truncation pola.

Przykłady zapytań

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

Typowe wzorce sprzężenia

Dołącz do tabeli, pipelines aby filtrować według nazwy potoku

Tabela pipelines jest powoli zmieniającym się wymiarem (SCD2). Przed dołączeniem weź najnowszą wersję każdego potoku.

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

Dołącz do pipeline_update_timelineupdate_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

Ustawianie alertów

Alerty pipeline_events dotyczące usługi Databricks można tworzyć przy użyciu alertów SQL usługi Databricks. Napisz zapytanie SQL względem pipeline_events (opcjonalnie połączone z innymi tabelami systemu Lakeflow), zaplanuj je w usłudze SQL Warehouse i skonfiguruj miejsce docelowe powiadomień (e-mail, Slack, webhook, PagerDuty).

Kilka przydatnych punktów początkowych:

Alarm, gdy w ciągu ostatnich N minut nie pojawiły się żadne zdarzenia dotyczące działającego pipeline'u

Służy do wykrywania zablokowanych lub dyskretnie zakończonych niepowodzeniem potoków.

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

Alert, gdy zaległości dla określonego przepływu są zbyt wysokie

Zaległości są raportowane przy flow_progress zdarzeniach jako , backlog_bytesa dla źródeł plików także jako backlog_files. Wyzwalaj, gdy ostatni odczyt przekroczy próg (na przykład 100 MB nieprzetworzonej pracy). Nie każde źródło podaje wszystkie wskaźniki, więc filtruj ten, który Twoje źródło wypełnia.

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

Aby zobaczyć, które źródło jest za nim, $.flow_progress.metrics.source_metrics jest to zestaw odczytów dla źródła, z których każde jest obok source_name tego backlog_bytesbacklog_records źródła , lub backlog_files.

Alert dotyczący spadku jakości danych w potoku

Każde flow_progress zdarzenie zgłasza liczbę wierszy porzuconych przez EXPECT … DROP oczekiwania. Sumuj je na zbiór danych w oknie aktualizacji i ostrzegaj, gdy łączna liczba przekroczy próg.

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

Bądź ostrożny, gdy przepływ przetwarza zbyt mało wierszy

flow_progress Zdarzenia raportują metrics.num_output_rows jako liczba na mikro-partię, więc sumowanie zdarzeń w oknie daje wiersze zapisane nad tym oknem. Stwórz alert, gdy przepustowość spadnie poniżej oczekiwanego progu. Na przykład przepływ, który normalnie zapisuje tysiące wierszy na godzinę, ale generuje niemal zero, może wskazywać na źle skonfigurowane źródło.

To zapytanie raportuje tylko przepływy, które wygenerowały zdarzenie flow_progress z liczbą wierszy w oknie. Całkowicie zatrzymany przepływ nie emituje żadnych zdarzeń, więc połącz ten alert z powyższym alertem o brakujących zdarzeniach.

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

Porady dotyczące alertów produkcyjnych

• Filtruj według pipeline_id (i workspace_id jeśli utrzymujesz alerty dla obszaru roboczego), aby każdy alert był przeznaczony dla określonego zakresu, a nie całego konta.
• Wybierz cykl oceny zgodny z poufnością alertu. Użyj krótkiego interwału (na przykład co 5 minut) dla sygnałów szybkiej awarii i dłuższego interwału (na przykład godzinowego) dla trendów listy prac i jakości danych. W większości przypadków działa warunek "wyzwalacz, gdy zapytanie zwraca więcej niż 0 wierszy".

Wartości odniesienia

Wartości poziomów

Wartości poziomu dojrzałości

Wartości typu zdarzenia

Pole event_type jest wyliczeniem. Pełny zestaw wartości: