Pozorovatelnost v Azure Databricks pro úlohy, kanály Lakeflow a Lakeflow Connect

Monitorování výkonu, nákladů a stavu streamovaných aplikací je nezbytné k vytváření spolehlivých a efektivních kanálů ETL. Azure Databricks poskytuje bohatou sadu funkcí pozorovatelnosti napříč úlohami, kanály Lakeflow a Lakeflow Connect, které pomáhají diagnostikovat kritické body, optimalizovat výkon a spravovat využití a náklady na prostředky.

Tyto osvědčené postupy zahrnují následující oblasti:

Klíčové metriky výkonu streamování
Schémata protokolu událostí a ukázkové dotazy
Monitorování streamovacích dotazů
Pozorovatelnost nákladů pomocí systémových tabulek
Export protokolů a metrik do externích nástrojů

Klíčové metriky pro pozorovatelnost streamování

Při provozu streamovacích kanálů monitorujte následující klíčové metriky:

Metric	Purpose
Backpressure	Monitoruje počet souborů a posunů (velikostí). Pomáhá identifikovat kritické body a zajišťuje, aby systém mohl zpracovávat příchozí data, aniž by zapadal.
Throughput	Sleduje počet zpráv zpracovaných v mikrodávce. Vyhodnoťte efektivitu kanálu a zkontrolujte, jestli udržuje krok s příjmem dat.
Duration	Měří průměrnou dobu trvání mikrodávky. Označuje rychlost zpracování a pomáhá ladit dávkové intervaly.
Latency	Určuje, kolik záznamů a zpráv se zpracovává v průběhu času. Pomáhá pochopit zpoždění mezi koncovými kanály a optimalizovat je pro nižší latence.
Využití clusteru	Odráží využití procesoru a paměti (%). Zajišťuje efektivní využití prostředků a pomáhá škálovat clustery tak, aby splňovaly požadavky na zpracování.
Network	Měří přenášená a přijatá data. Užitečné pro identifikaci kritických bodů sítě a zlepšení výkonu přenosu dat.
Checkpoint	Identifikuje zpracovávaná data a posuny. Zajišťuje konzistenci a umožňuje odolnost proti chybám během selhání.
Cost	Zobrazuje hodinovou, denní a měsíční náklady na streamovací aplikaci. Pomáhá při rozpočtování a optimalizaci zdrojů.
Lineage	Zobrazí datové sady a vrstvy vytvořené v aplikaci streamování. Usnadňuje transformaci dat, kontrolu kvality, sledování a ladění.

Protokoly a metriky clusteru

Protokoly a metriky clusteru Azure Databricks poskytují podrobné přehledy o výkonu a využití clusteru. Mezi tyto protokoly a metriky patří informace o procesoru, paměti, vstupně-výstupních operacích disku, síťovém provozu a dalších systémových metrikách. Monitorování těchto metrik je zásadní pro optimalizaci výkonu clusteru, efektivní správu prostředků a řešení potíží.

Protokoly a metriky clusteru Azure Databricks nabízejí podrobné přehledy o výkonu clusteru a využití prostředků. Patří sem využití procesoru a paměti, vstupně-výstupní operace disku a síťový provoz. Monitorování těchto metrik je důležité pro:

Optimalizace výkonu clusteru
Efektivní správa prostředků
Řešení provozních problémů

Metriky je možné využít prostřednictvím uživatelského rozhraní Databricks nebo exportovat do osobních monitorovacích nástrojů. Viz příklad poznámkového bloku: Metriky služby Datadog.

Uživatelské rozhraní Sparku

Uživatelské rozhraní Sparku zobrazuje podrobné informace o průběhu úloh a fází, včetně počtu dokončených, čekajících a neúspěšných úkolů. To vám pomůže pochopit tok provádění a identifikovat kritické body.

U streamovacích aplikací se na kartě Streamování zobrazují metriky, jako je vstupní rychlost, rychlost zpracování a doba trvání dávky. Pomáhá monitorovat výkon úloh streamování a identifikovat případné problémy s příjmem nebo zpracováním dat.

Další informace najdete v tématu Ladění pomocí uživatelského rozhraní Sparku .

Výpočetní metriky

Výpočetní metriky vám pomůžou porozumět využití clusteru. Při spuštění úlohy uvidíte, jak se škáluje a jak jsou vaše prostředky ovlivněny. Budete schopni najít zatížení paměti, které by mohlo vést k selhání kvůli nedostatku paměti (OOM), nebo zatížení procesoru, které by mohlo způsobit dlouhá zpoždění. Tady jsou konkrétní metriky, které uvidíte:

Distribuce zatížení serveru: Využití procesoru každého uzlu za poslední minutu
Využití procesoru: Procento času stráveného procesorem v různých režimech (například uživatel, systém, nečinný a iowait).
Využití paměti: Celkové využití paměti v jednotlivých režimech (například využité, volné, vyrovnávací paměti a ukládání do mezipaměti).
Využití paměti swapu: Celkové využití paměti swapu
Volné místo systému souborů: Celkové využití systému souborů každým přípojným bodem
Propustnost sítě: Počet přijatých a přenášených bajtů prostřednictvím sítě jednotlivými zařízeními.
Počet aktivních uzlů: Počet aktivních uzlů v každém časovém razítku daného výpočetního prostředí.

Další informace najdete v grafech hardwarových metrik .

Systémové tabulky

Monitorování nákladů

Systémové tabulky Azure Databricks poskytují strukturovaný přístup k monitorování nákladů na úlohy a výkonu. Mezi tyto tabulky patří:

Podrobnosti o spuštění úlohy
Využití prostředků
Přidružené náklady.

Tyto tabulky slouží k pochopení provozního stavu a finančního dopadu.

Requirements

Použití systémových tabulek pro monitorování nákladů:

Správce účtu musí povolit system.lakeflow schema.
Uživatelé musí:
- Být správcem metastoru i správcem účtu nebo
- Mít oprávnění USE a SELECT k systémovým schématům.

Příklad dotazu: Nejnákladnější úlohy (posledních 30 dnů)

Tento dotaz identifikuje nejnákladnější úlohy za posledních 30 dnů, což pomáhá při analýze nákladů a optimalizaci.

WITH list_cost_per_job AS (
     SELECT
       t1.workspace_id,
       t1.usage_metadata.job_id,
       COUNT(DISTINCT t1.usage_metadata.job_run_id) AS runs,
       SUM(t1.usage_quantity * list_prices.pricing.default) AS list_cost,
       FIRST(identity_metadata.run_as, true) AS run_as,
       FIRST(t1.custom_tags, true) AS custom_tags,
       MAX(t1.usage_end_time) AS last_seen_date
     FROM system.billing.usage t1
     INNER JOIN system.billing.list_prices list_prices ON
       t1.cloud = list_prices.cloud AND
       t1.sku_name = list_prices.sku_name AND
       t1.usage_start_time >= list_prices.price_start_time AND
       (t1.usage_end_time <= list_prices.price_end_time OR list_prices.price_end_time IS NULL)
     WHERE
       t1.billing_origin_product = "JOBS"
       AND t1.usage_date >= CURRENT_DATE() - INTERVAL 30 DAY
     GROUP BY ALL
   ),
   most_recent_jobs AS (
     SELECT
       *,
       ROW_NUMBER() OVER(PARTITION BY workspace_id, job_id ORDER BY change_time DESC) AS rn
     FROM
       system.lakeflow.jobs QUALIFY rn=1
   )
   SELECT
     t2.name,
     t1.job_id,
     t1.workspace_id,
     t1.runs,
     t1.run_as,
     SUM(list_cost) AS list_cost,
     t1.last_seen_date
   FROM list_cost_per_job t1
   LEFT JOIN most_recent_jobs t2 USING (workspace_id, job_id)
   GROUP BY ALL
   ORDER BY list_cost DESC

Kanály Lakeflow

Protokol událostí kanálů Lakeflow zaznamenává komplexní záznam všech událostí kanálu, včetně:

Protokoly auditu
Kontroly kvality dat.
Průběh pipeline
Původ dat.

Protokol událostí je automaticky povolený pro všechny kanály Lakeflow a dá se k němu přistupovat prostřednictvím:

Rozhraní Pipeline: Prohlížet protokoly přímo.
API pro pipelines: Programatický přístup.
Přímý dotaz: Zadejte dotaz na tabulku protokolu událostí.

Další informace najdete ve schématu protokolu událostí pro kanály Lakeflow.

Ukázkové dotazy

Tyto ukázkové dotazy pomáhají monitorovat výkon a stav datových toků tím, že poskytují klíčové metriky, jako je doba trvání dávky, propustnost, zpětný tlak a využití prostředků.

Každý dotaz čte z event_log_raw, což je pohled na tabulku logu událostí vašeho pipeline. Pro vytvoření viz Dotaz do event logu.

Průměrná doba trvání dávky

Tento dotaz vypočítá průměrnou dobu trvání dávek zpracovaných kanálem.

SELECT
  (max_t - min_t) / batch_count as avg_batch_duration_seconds,
  batch_count,
  min_t,
  max_t,
  date_hr,
  message
FROM
  -- /60 for minutes
  (
    SELECT
      count(*) as batch_count,
      unix_timestamp(
        min(timestamp)
      ) as min_t,
      unix_timestamp(
        max(timestamp)
      ) as max_t,
      date_format(timestamp, 'yyyy-MM-dd:HH') as date_hr,
      message
    FROM
      event_log_raw
    WHERE
      event_type = 'flow_progress'
      AND level = 'METRICS'
    GROUP BY
      date_hr,
      message
  )
ORDER BY
  date_hr desc

Průměrná propustnost

Tento dotaz vypočítá průměrnou propustnost kanálu z hlediska zpracovaných řádků za sekundu.

SELECT
  (max_t - min_t) / total_rows as avg_throughput_rps,
  total_rows,
  min_t,
  max_t,
  date_hr,
  message
FROM
  -- /60 for minutes
  (
    SELECT
      sum(
        details:flow_progress:metrics:num_output_rows
      ) as total_rows,
      unix_timestamp(
        min(timestamp)
      ) as min_t,
      unix_timestamp(
        max(timestamp)
      ) as max_t,
      date_format(timestamp, 'yyyy-MM-dd:HH') as date_hr,
      message
    FROM
      event_log_raw
    WHERE
      event_type = 'flow_progress'
      AND level = 'METRICS'
    GROUP BY
      date_hr,
      message
  )
ORDER BY
  date_hr desc

Backpressure

Tento dotaz měří zpětný tlak kanálu kontrolou čekačky dat.

SELECT
  timestamp,
  DOUBLE(
    details:flow_progress:metrics:backlog_bytes
  ) AS backlog_bytes,
  DOUBLE(
    details:flow_progress:metrics:backlog_files
  ) AS backlog_files
FROM
  event_log_raw
WHERE
  event_type = 'flow_progress'

Využití clusterů a slotů

Tento dotaz poskytuje informace o využití clusterů nebo slotů, které používá systém. Platí pouze pro klasické výpočetní clustery. Podrobnosti o cluster_resources akci.

SELECT
  date_trunc("hour", timestamp) AS hour,
  AVG (
    DOUBLE (
      details:cluster_resources:num_task_slots
    )
  ) AS num_task_slots,
  AVG (
    DOUBLE (
      details:cluster_resources:avg_num_task_slots
    )
  ) AS avg_num_task_slots,
  AVG (
    DOUBLE (
      details:cluster_resources:num_executors
    )
  ) AS num_executors,
  AVG (
    DOUBLE (
      details:cluster_resources:avg_task_slot_utilization
    )
  ) AS avg_utilization,
  AVG (
    DOUBLE (
      details:cluster_resources:avg_num_queued_tasks
    )
  ) AS queue_size
FROM
  event_log_raw
WHERE
  details : cluster_resources : avg_num_queued_tasks IS NOT NULL
  AND origin.update_id = (
    SELECT origin.update_id
    FROM event_log_raw
    WHERE event_type = 'create_update'
    ORDER BY timestamp DESC
    LIMIT 1
  )
GROUP BY
  1;

Jobs

Streamovací dotazy můžete sledovat v úlohách prostřednictvím Streamovacího Query Listeneru.

Připojte k relaci Spark naslouchací proces a povolte tak v Azure Databricks naslouchací proces pro streamované dotazy. Tento naslouchací modul sleduje průběh a metriky vašich streamovacích dotazů. Dá se použít k nasdílení metrik externím monitorovacím nástrojům nebo k jejich protokolování pro další analýzu.

Příklad: Export metrik do externích monitorovacích nástrojů

Note

To je k dispozici v Databricks Runtime 11.3 LTS a vyšší pro Python a Scala.

Streamované metriky můžete exportovat do externích služeb pro upozorňování nebo řídicí panel pomocí StreamingQueryListener rozhraní.

Tady je základní příklad, jak implementovat posluchač:

from pyspark.sql.streaming import StreamingQueryListener

class MyListener(StreamingQueryListener):
   def onQueryStarted(self, event):
       print("Query started: ", event.id)

   def onQueryProgress(self, event):
       print("Query made progress: ", event.progress)

   def onQueryTerminated(self, event):
       print("Query terminated: ", event.id)

spark.streams.addListener(MyListener())

Příklad: Použití posluchače dotazu v Azure Databricks

Níže je příklad protokolu událostí StreamingQueryListener pro streamovací dotaz z Kafka do Delta Lake.

{
  "id": "210f4746-7caa-4a51-bd08-87cabb45bdbe",
  "runId": "42a2f990-c463-4a9c-9aae-95d6990e63f4",
  "timestamp": "2024-05-15T21:57:50.782Z",
  "batchId": 0,
  "batchDuration": 3601,
  "numInputRows": 20,
  "inputRowsPerSecond": 0.0,
  "processedRowsPerSecond": 5.55401277422938,
  "durationMs": {
    "addBatch": 1544,
    "commitBatch": 686,
    "commitOffsets": 27,
    "getBatch": 12,
    "latestOffset": 577,
    "queryPlanning": 105,
    "triggerExecution": 3600,
    "walCommit": 34
  },
  "stateOperators": [
    {
      "operatorName": "symmetricHashJoin",
      "numRowsTotal": 20,
      "numRowsUpdated": 20,
      "allUpdatesTimeMs": 473,
      "numRowsRemoved": 0,
      "allRemovalsTimeMs": 0,
      "commitTimeMs": 277,
      "memoryUsedBytes": 13120,
      "numRowsDroppedByWatermark": 0,
      "numShufflePartitions": 5,
      "numStateStoreInstances": 20,
      "customMetrics": {
        "loadedMapCacheHitCount": 0,
        "loadedMapCacheMissCount": 0,
        "stateOnCurrentVersionSizeBytes": 5280
      }
    }
  ],
  "sources": [
    {
      "description": "KafkaV2[Subscribe[topic-1]]",
      "numInputRows": 10,
      "inputRowsPerSecond": 0.0,
      "processedRowsPerSecond": 2.77700638711469,
      "metrics": {
        "avgOffsetsBehindLatest": "0.0",
        "estimatedTotalBytesBehindLatest": "0.0",
        "maxOffsetsBehindLatest": "0",
        "minOffsetsBehindLatest": "0"
      }
    },
    {
      "description": "DeltaSource[file:/tmp/spark-1b7cb042-bab8-4469-bb2f-733c15141081]",
      "numInputRows": 10,
      "inputRowsPerSecond": 0.0,
      "processedRowsPerSecond": 2.77700638711469,
      "metrics": {
        "numBytesOutstanding": "0",
        "numFilesOutstanding": "0"
      }
    }
  ]
}

Další příklady najdete v tématu: Příklady.

Metriky průběhu dotazů

Metriky průběhu dotazů jsou nezbytné pro monitorování výkonu a stavu streamovaných dotazů. Mezi tyto metriky patří počet vstupních řádků, rychlost zpracování a různé doby trvání související se spuštěním dotazu. Metriky můžete sledovat tím, že připojíte StreamingQueryListener k relaci Spark. Naslouchací modul na konci každé epochy streamování vysílá události obsahující tyto metriky.

K metrikám můžete například přistupovat pomocí StreamingQueryProgress.observedMetrics mapy v metodě posluchače onQueryProgress. Díky tomu můžete sledovat a analyzovat výkon streamovaných dotazů v reálném čase.

class MyListener(StreamingQueryListener):
   def onQueryProgress(self, event):
       print("Query made progress: ", event.progress.observedMetrics)

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-08-05