Monitorování kanálů DLT

Tento článek popisuje použití integrovaných funkcí monitorování a pozorovatelnosti pro kanály DLT. Tyto funkce podporují například tyto úlohy:

• Sledování průběhu a stavu aktualizací potrubí. Viz Jaké podrobnosti pipelinu jsou k dispozici v uživatelském rozhraní?.
• Upozornění na události v kanálu, jako je úspěch nebo neúspěch aktualizací kanálu. Viz Přidání e-mailových oznámení o událostech kanálu.
• Zobrazení metrik pro streamované zdroje, jako jsou Apache Kafka a Auto Loader (Public Preview). Viz Zobrazení metrik streamování.
• Extrahování podrobných informací o aktualizacích kanálu, jako jsou rodokmen dat, metriky kvality dat a využití prostředků Viz Co je protokol událostí DLT?.
• Definování vlastních akcí, které se provedou, když nastanou konkrétní události. Viz Definování vlastního monitorování kanálů DLT pomocí událostních háků.

Informace o kontrole a diagnostice výkonu dotazů najdete v tématu Historie dotazů accessu pro kanály DLT. Tato funkce je ve verzi Public Preview.

Přidejte e-mailová oznámení pro události pipeline

Můžete nakonfigurovat jednu nebo více e-mailových adres pro příjem oznámení, když dojde k následujícímu:

• Aktualizace pipeline byla úspěšně provedena.
• Aktualizace pipeline selže, buď s chybou, kterou lze opakovat, nebo s chybou, kterou nelze opakovat. Tuto možnost vyberte, pokud chcete dostávat oznámení o všech selháních kanálu.
• Aktualizace potrubí selže s neopravitelnou (fatální) chybou. Tuto možnost vyberte, pokud chcete dostávat oznámení pouze v případě, že dojde k neopakovatelné chybě.
• Jeden datový tok selže.

Pro nastavení e-mailových oznámení při vytvoření nebo úpravě potrubí:

1. Klikněte na Přidat oznámení.
2. Zadejte jednu nebo více e-mailových adres pro příjem oznámení.
3. Kliknutím na zaškrtávací políčko u každého typu oznámení odešlete na nakonfigurované e-mailové adresy.
4. Klikněte na Přidat oznámení.

Jaké podrobnosti o potrubí (pipelině) jsou k dispozici v uživatelském rozhraní?

Graf pipeline se zobrazí, jakmile se úspěšně spustí aktualizace pipeline. Šipky představují závislosti mezi datovými sadami v rámci vašeho potrubí. Ve výchozím nastavení se na stránce podrobností datového toku zobrazuje nejnovější aktualizaci tabulky, ale starší aktualizace můžete vybrat z rozevírací nabídky.

Podrobnosti zahrnují ID kanálu, zdrojový kód, náklady na výpočetní prostředky, edici produktu a kanál nakonfigurovaný pro kanál.

Pokud chcete zobrazit tabulkové zobrazení datových sad, klikněte na kartu Seznam. Zobrazení Seznam umožňuje zobrazit všechny datové sady v kanálu reprezentované jako řádek v tabulce a je užitečné, když je DAG kanálu příliš velký, aby se vizualizoval v zobrazení Grafu. Datové sady zobrazené v tabulce můžete řídit pomocí více filtrů, jako je název datové sady, typ a stav. Pokud chcete přepnout zpět na vizualizaci DAG, klikněte na graph.

Uživatel Spustit jako je vlastníkem potrubí a aktualizace potrubí se spouštějí s oprávněními tohoto uživatele. Chcete-li změnit uživatele run as, klikněte na Oprávnění a změňte vlastníka potrubí.

Jak můžete zobrazit podrobnosti datové sady?

Kliknutím na datovou sadu v grafu kanálu nebo seznamu datových sad zobrazíte podrobnosti o datové sadě. Podrobnosti zahrnují schéma datové sady, metriky kvality dat a odkaz na zdrojový kód definující datovou sadu.

zobrazení historie aktualizací

Pokud chcete zobrazit historii a stav aktualizací kanálu, klikněte na rozevírací nabídku historie aktualizací v horním panelu.

Výběrem aktualizace v rozevírací nabídce zobrazíte graf, podrobnosti a události aktualizace. Chcete-li se vrátit k nejnovější aktualizaci, klepněte na tlačítko Zobrazit nejnovější aktualizace.

Zobrazení metrik streamování

Můžete zobrazit metriky streamování ze zdrojů dat podporovaných strukturovaným streamováním Sparku, jako je Apache Kafka, Amazon Kinesis, Auto Loader a Delta, pro každý tok streamování v kanálu DLT. Metriky se zobrazují jako grafy v pravém podokně uživatelského rozhraní DLT a zahrnují sekundy backlogu, bajty backlogu, záznamy backlogu a soubory backlogu. Grafy zobrazují maximální hodnotu agregovanou po minutě a nápověda ukazuje maximální hodnoty při najetí myší na graf. Data jsou omezená na posledních 48 hodin od aktuálního času.

Tabulky ve vašem datovém kanálu, které mají k dispozici metriky streamování, zobrazují ikonu při zobrazení DAG kanálu v uživatelském rozhraní v zobrazení Graf. Pokud chcete zobrazit metriky streamování, klikněte na ikonu grafu a zobrazte graf metrik streamování na kartě Toky v pravém podokně. Filtr můžete použít také k zobrazení pouze tabulek s metrikami streamování kliknutím na Seznam a následným kliknutím na Má metriky streamování.

Každý zdroj streamování podporuje pouze konkrétní metriky. Metriky, které zdroj streamování nepodporuje, nejsou k dispozici pro zobrazení v uživatelském rozhraní. Následující tabulka uvádí metriky dostupné pro podporované zdroje streamování:

Co je protokol událostí DLT?

Protokol událostí DLT obsahuje všechny informace související s kanálem, včetně protokolů auditu, kontrol kvality dat, průběhu kanálu a rodokmenu dat. Protokol událostí můžete použít ke sledování, pochopení a monitorování stavu datových kanálů.

Položky protokolu událostí můžete zobrazit v uživatelském rozhraní DLT, rozhraní DLT APInebo přímo dotazováním protokolu událostí. Tato část se zaměřuje na přímé dotazování protokolu událostí.

Můžete také definovat vlastní akce, které se mají spustit při protokolování událostí, například odesílání výstrah, pomocí událostních hooků .

schéma protokolu událostí

Následující tabulka popisuje schéma protokolu událostí. Některá z těchto polí obsahují data JSON, která vyžadují analýzu některých dotazů, například pole details. Azure Databricks podporuje operátor : k analýze polí JSON. Viz operátor (znak dvojtečky):.

Proveďte dotaz na protokol událostí

Ve výchozím nastavení DLT zapíše protokol událostí do skryté tabulky Delta ve výchozím katalogu a schématu nakonfigurovaného pro kanál. I když je tabulka skrytá, může se na tuto tabulku dotazovat všichni dostatečně privilegovaní uživatelé. Ve výchozím nastavení může tabulku protokolu událostí dotazovat pouze vlastník kanálu.

Ve výchozím nastavení je název skrytého protokolu událostí formátovaný jako event_log_{pipeline_id}, kde ID kanálu je UUID přiřazené systémem s pomlčkami nahrazenými podtržítky.

S konfigurací JSON můžete pracovat a publikovat protokol událostí. Při publikování protokolu událostí zadáte název protokolu událostí a volitelně můžete zadat katalog a schéma, jako v následujícím příkladu:

{
  "id": "ec2a0ff4-d2a5-4c8c-bf1d-d9f12f10e749",
  "name": "billing_pipeline",
  "event_log": {
    "catalog": "catalog_name",
    "schema": "schema_name",
    "name": "event_log_table_name"
  }
}

Umístění protokolu událostí slouží také jako umístění schématu pro všechny dotazy nástroje Auto Loader v rámci pracovního procesu. Databricks doporučuje vytvořit zobrazení tabulky protokolu událostí před úpravou oprávnění, protože některá nastavení výpočetních prostředků můžou uživatelům umožnit získat přístup k metadatům schématu, pokud se tabulka protokolu událostí sdílí přímo. Následující příklad syntaxe vytvoří zobrazení v tabulce protokolu událostí a používá se v ukázkových dotazech protokolu událostí zahrnutých v tomto článku.

CREATE VIEW event_log_raw
AS SELECT * FROM catalog_name.schema_name.event_log_table_name;

Každá instance spuštění potrubí se nazývá aktualizace. Často chcete extrahovat informace pro nejnovější aktualizaci. Spuštěním následujícího dotazu vyhledejte identifikátor poslední aktualizace a uložte ho v latest_update dočasném zobrazení. Toto zobrazení se používá v ukázkových dotazech protokolu událostí, které jsou součástí tohoto článku:

CREATE OR REPLACE TEMP VIEW latest_update AS SELECT origin.update_id AS id FROM event_log_raw WHERE event_type = 'create_update' ORDER BY timestamp DESC LIMIT 1;

Zobrazení v katalogu Unity podporují streamovací dotazy. Následující příklad používá strukturované streamování k dotazování na zobrazení, které je definováno nad tabulkou protokolu událostí.

df = spark.readStream.table("event_log_raw")

Vlastník kanálu může publikovat protokol událostí jako veřejnou tabulku Delta přepnutím možnosti Publish event log to metastore v části konfigurace kanálu Advanced. Volitelně můžete zadat nový název tabulky, katalog a schéma pro protokol událostí.

Dotázat se na trasování informací z protokolu událostí

Události obsahující informace o rodokmenu mají typ události flow_definition. Objekt details:flow_definition obsahuje output_dataset a input_datasets definující jednotlivé relace v grafu.

Pomocí následujícího dotazu můžete extrahovat vstupní a výstupní datové sady a zobrazit informace o rodokmenu:

SELECT
  details:flow_definition.output_dataset as output_dataset,
  details:flow_definition.input_datasets as input_dataset
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'flow_definition'
  AND
  origin.update_id = latest_update.id

Dotaz na kvalitu dat z protokolu událostí

Pokud definujete očekávání u datových sad ve vašem potrubí, metriky kvality dat se ukládají do objektu details:flow_progress.data_quality.expectations. Události obsahující informace o kvalitě dat mají typ události flow_progress. Následující příklad dotazuje metriky kvality dat pro poslední aktualizaci datového potrubí:

SELECT
  row_expectations.dataset as dataset,
  row_expectations.name as expectation,
  SUM(row_expectations.passed_records) as passing_records,
  SUM(row_expectations.failed_records) as failing_records
FROM
  (
    SELECT
      explode(
        from_json(
          details :flow_progress :data_quality :expectations,
          "array<struct<name: string, dataset: string, passed_records: int, failed_records: int>>"
        )
      ) row_expectations
    FROM
      event_log_raw,
      latest_update
    WHERE
      event_type = 'flow_progress'
      AND origin.update_id = latest_update.id
  )
GROUP BY
  row_expectations.dataset,
  row_expectations.name

Dotaz k událostem autoloadera z protokolu událostí

DLT generuje události při zpracování souborů pomocí Auto Loader. U událostí automatického zavaděče je event_typeoperation_progress a details:operation_progress:type je buď AUTO_LOADER_LISTING, nebo AUTO_LOADER_BACKFILL. Objekt details:operation_progress obsahuje také pole status, duration_ms, auto_loader_details:source_patha auto_loader_details:num_files_listed.

Následující příklad vyhledává události v Auto Loaderu o nejnovější aktualizaci:

SELECT
  timestamp,
  details:operation_progress.status,
  details:operation_progress.type,
  details:operation_progress:auto_loader_details
FROM
  event_log_raw,
  latest_update
WHERE
  event_type like 'operation_progress'
  AND
  origin.update_id = latest.update_id
  AND
  details:operation_progress.type in ('AUTO_LOADER_LISTING', 'AUTO_LOADER_BACKFILL')

Monitorování backlogu dat dotazováním protokolu událostí

DLT sleduje, kolik dat je v backlogu v details:flow_progress.metrics.backlog_bytes objektu. Události obsahující metriky backlogu mají typ události flow_progress. Následující příklad získává metriky backlogu pro poslední aktualizaci potrubí:

SELECT
  timestamp,
  Double(details :flow_progress.metrics.backlog_bytes) as backlog
FROM
  event_log_raw,
  latest_update
WHERE
  event_type ='flow_progress'
  AND
  origin.update_id = latest_update.id

Sledujte rozšířené události automatického škálování z protokolu událostí pro potrubí, kde není povoleno serverless

U kanálů DLT, které nepoužívají bezserverové výpočetní prostředky, zachytává protokol událostí změnu velikosti clusteru, když je ve vašich kanálech povolené rozšířené automatické škálování. Události obsahující informace o rozšířeném automatickém škálování mají typ události autoscale. Informace o změně velikosti žádosti clusteru jsou uloženy v objektu details:autoscale. Následující příklad zjišťuje žádosti o vylepšenou změnu velikosti clusteru automatického škálování pro poslední aktualizaci potrubí:

SELECT
  timestamp,
  Double(
    case
      when details :autoscale.status = 'RESIZING' then details :autoscale.requested_num_executors
      else null
    end
  ) as starting_num_executors,
  Double(
    case
      when details :autoscale.status = 'SUCCEEDED' then details :autoscale.requested_num_executors
      else null
    end
  ) as succeeded_num_executors,
  Double(
    case
      when details :autoscale.status = 'PARTIALLY_SUCCEEDED' then details :autoscale.requested_num_executors
      else null
    end
  ) as partially_succeeded_num_executors,
  Double(
    case
      when details :autoscale.status = 'FAILED' then details :autoscale.requested_num_executors
      else null
    end
  ) as failed_num_executors
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'autoscale'
  AND
  origin.update_id = latest_update.id

Monitorování využití výpočetních prostředků

cluster_resources události poskytují metriky o počtu slotů úloh v clusteru, kolik těchto slotů úloh se využívá a kolik úkolů čeká na naplánování.

Pokud je povolené rozšířené automatické škálování, cluster_resources události také obsahují metriky pro algoritmus automatického škálování, včetně latest_requested_num_executorsa optimal_num_executors. Události také zobrazují stav algoritmu jako různé stavy, jako jsou CLUSTER_AT_DESIRED_SIZE, SCALE_UP_IN_PROGRESS_WAITING_FOR_EXECUTORSa BLOCKED_FROM_SCALING_DOWN_BY_CONFIGURATION. Tyto informace je možné zobrazit ve spojení s událostmi automatického škálování a poskytnout tak celkový přehled o vylepšeném automatickém škálování.

Následující příklad zjišťuje historii velikosti fronty úkolů pro poslední aktualizaci pipeline.

SELECT
  timestamp,
  Double(details :cluster_resources.avg_num_queued_tasks) as queue_size
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'cluster_resources'
  AND
  origin.update_id = latest_update.id

Následující příklad se ptá na historii využívání při poslední aktualizaci kanálu:

SELECT
  timestamp,
  Double(details :cluster_resources.avg_task_slot_utilization) as utilization
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'cluster_resources'
  AND
  origin.update_id = latest_update.id

Následující příklad dotazuje historii počtu exekutorů, doprovázené metrikami dostupnými pouze pro rozšířené kanály automatického škálování, včetně počtu exekutorů požadovaných algoritmem v nejnovějším požadavku, optimálního počtu exekutorů doporučených algoritmem na základě nejnovějších metrik a stavu algoritmu automatického škálování:

SELECT
  timestamp,
  Double(details :cluster_resources.num_executors) as current_executors,
  Double(details :cluster_resources.latest_requested_num_executors) as latest_requested_num_executors,
  Double(details :cluster_resources.optimal_num_executors) as optimal_num_executors,
  details :cluster_resources.state as autoscaling_state
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'cluster_resources'
  AND
  origin.update_id = latest_update.id

auditování potrubí DLT

K získání kompletního přehledu o tom, jak se data aktualizují v DLT, můžete použít záznamy protokolu událostí DLT a další protokoly auditu Azure Databricks.

DlT ke spuštění aktualizací používá přihlašovací údaje vlastníka kanálu. Přihlašovací údaje můžete změnit tím, že aktualizujete vlastníka pipeline. DLT zaznamenává uživatele pro akce na potrubí, včetně vytvoření potrubí, úprav konfigurace a spuštění aktualizací.

Referenční informace o událostech auditování katalogu Unity najdete v tématu události katalogu Unity.

Dotazování akcí uživatele v protokolu událostí

Protokol událostí můžete použít k auditování událostí, například akcí uživatelů. Události obsahující informace o akcích uživatele mají typ události user_action.

Informace o akci jsou uloženy v objektu user_action v poli details. Pomocí následujícího dotazu vytvořte protokol auditu uživatelských událostí. Chcete-li vytvořit zobrazení event_log_raw použité v tomto dotazu, podívejte se na Dotaz na protokol událostí.

SELECT timestamp, details:user_action:action, details:user_action:user_name FROM event_log_raw WHERE event_type = 'user_action'

informace o modulu runtime

Můžete zobrazit informace o modulu runtime pro aktualizaci pipeline, například verzi Databricks Runtime pro aktualizaci:

SELECT details:create_update:runtime_version:dbr_version FROM event_log_raw WHERE event_type = 'create_update'