Monitorowanie zdarzeń i wykonywanie zapytań

Każdy pipeline emituje zdarzenia, które rejestrują logi audytowe, kontrole jakości danych, postępy i pochodzenie danych. Możesz zapytać o te wydarzenia z dwóch źródeł:

• Tabelapipeline_events systemowa zawiera zdarzenia dla wszystkich potoków w przestrzeniach roboczych w danym regionie i jest zalecanym sposobem zapytań o zdarzenia. Jest w fazie beta.
• Dziennik zdarzeń dla potoku to tabela Delta zawierająca zdarzenia dla pojedynczego potoku.

Zapytujesz zdarzenia za pomocą standardowego SQL. Aby pomóc Ci zacząć, ta strona zawiera także przykładowy dashboard oraz typowe zapytania dla tabeli systemowej.

Requirements

Aby uzyskać dostęp do tej tabeli systemowej, użytkownicy muszą wykonać:

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

Przykładowy pulpit nawigacyjny

Ten przykładowy pulpit odczytuje tabelę systemową pipeline_events, aby śledzić aktualizacje potoków, przepustowość, zaległości, jakość danych oraz błędy we wszystkich potokach w danym regionie. Filtruj każdą stronę według pipeline'u, tabeli, tagu i zakresu czasowego.

Importowanie pulpitu nawigacyjnego

1. Pobierz plik JSON dashboardu.
2. Zaimportuj pulpit nawigacyjny do obszaru roboczego. Instrukcje dostępne są w artykule Import pliku dashboardu.

Zapytania monitorujące

Poniższe zapytania z panelu pokazują typowe zastosowania monitorowania potoków.

Ostatni błąd dla każdego potoku

To zapytanie zwraca najnowszy błąd dla każdego potoku przetwarzania, w którym wystąpił błąd w ciągu ostatnich 7 dni, wraz z najbardziej zewnętrznym wyjątkiem.

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

Wskaźnik błędów godzinowych

To zapytanie zlicza błędy dla każdego potoku w każdej godzinie, dzięki czemu możesz wykrywać skoki.

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

Rzędy zmieniane w zależności od przepływu

To zapytanie sumuje liczbę wierszy dodanych, upsertowanych i usuniętych przez przepływ w każdej godzinie. Każda metryka jest liczbą przypadającą na każdą mikropartię, więc zsumowanie wartości w obrębie okna daje przepustowość.

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

Zaległości według przepływu

To zapytanie zwraca najnowszą wartość zaległości dla każdego przepływu. Przepływ COMPLETED jest doganiany, więc jego zaległości są zgłaszane jako .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

Niespełnione oczekiwania według zbioru danych

To zapytanie zwraca reguły walidacji, dla których rekordy zakończyły się niepowodzeniem, dla każdego zbioru danych i każdej aktualizacji z ostatniego dnia. Opiera się na własnym dataset oczekiwania, które jest wypełniane nawet wtedy, gdy origin.dataset_name zdarzenia nie jest wypełnione.

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