Použití vodoznaků k řízení prahových hodnot zpracování dat

Tato stránka popisuje koncepty vodoznaků a obsahuje doporučení k používání vodoznaků v běžných stavových operacích streamování.

Streamované dotazy hromadí stavová data v průběhu času. Vodoznaky automaticky odeberou stará data o stavu, aby se zabránilo chybám paměti a vyšší latenci zpracování.

Co je vodoznak?

Strukturované streamování během zpracování uchovává stav napříč mikrodávkovými dávkami. Streamovací dotazy používají stav k průběžné aktualizaci výsledků místo toho, aby po každé mikrodávce znovu přepočítávaly vše. Vodoznaky určují prahovou hodnotu, kdy dotaz přestane zpracovávat entitu stavu.

Mezi běžné příklady stavových entit patří:

• Agregace v časovém intervalu
• Jedinečné klíče v propojení mezi dvěma datovými proudy

Pokud chcete deklarovat vodoznak u streamovaného datového rámce, zadejte pole časového razítka a prahovou hodnotu zpoždění. Při příchodu nových dat správce stavu sleduje poslední časové razítko v zadaném poli a zpracovává pouze záznamy v rámci prahové hodnoty zpoždění.

Dotazy vždy zpracovávají záznamy, které přicházejí do prahové hodnoty. Dotazy můžou stále zpracovávat záznamy, které přicházejí mimo prahovou hodnotu, ale to není zaručené.

Následující příklad použije prahovou hodnotu meze 10 minut na počet oken:

Python

from pyspark.sql.functions import window

(df
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window("event_time", "5 minutes"),
    "id")
  .count()
)

Scala

import org.apache.spark.sql.functions.window

df
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window($"event_time", "5 minutes"),
    $"id")
  .count()

V tomto příkladu:

• Sloupec event_time slouží k definování 10minutového vodoznaku a 5minutového přeskakujícího okna.
• Pro každý pozorovaný id je zaznamenán počet pro každý nepřekrývající se 5minutový interval.
• Informace o stavu jsou pro každé počítadlo uchovávány, dokud není konec okna o 10 minut starší než nejnověji pozorované event_time.

Důležité

Při operaci groupBy() a window() odkazujte na sloupce podle názvu, "<colName>" nebo col("<colName>"), aby se zachovala značka času události. Ve Scale můžete také použít $colName.

Jak vodoznaky ovlivňují dobu zpracování a propustnost?

Režimy výstupu určují, kdy dotaz s vodoznaky zapisuje data do jímky. Vodoznaky jsou nezbytné pro řízení propustnosti ve stavovém streamování, protože snižují celkové množství informací o stavu v paměti. Pro všechny stavové operace se nepodporují všechny režimy výstupu. viz Vodoznaky a výstupní režim pro okenní agregace.

Výběr doby trvání vodoznaku má kompromisy:

• Kratší vodoznaky snižují latenci dotazů, protože dotazy ukládají méně informací o stavu a po dokončení každé doby trvání vodoznaku zapisují výsledky. Krátké vodoznaky ale mají nízkou toleranci pro pozdní data.
• Delší vodoznaky mají vysokou toleranci pro pozdní data. Dlouhé vodoznaky ale zvyšují latenci dotazů, protože dotazy musí ukládat další informace o stavu a čekat na zápis výsledků po delší době trvání vodoznaku.

Vodoznaky a výstupní režim pro okenní agregace

Následující tabulka ukazuje chování zpracování dotazů s agregací nad časovým razítkem a vodoznakem:

Výstupní režim	Chování
Připojit	Dotaz zapíše řádky do cílové tabulky po překročení prahové hodnoty vodoznaku. Všechny zápisy jsou zpožděné na základě prahové hodnoty zpoždění. Starý stav agregace se po uplynutí prahové hodnoty zahodí.
Aktualizace	Dotaz zapíše řádky do cílové tabulky při výpočtu výsledků a dotaz může aktualizovat a přepsat řádky při příchodu nových dat. Starý stav agregace se po uplynutí prahové hodnoty zahodí.
Dokončit	Stav agregace se nezahodí. Dotaz přepíše cílovou tabulku pro každou aktivační událost.

Vodoznaky a výstupní režimy pro spojení mezi streamy

Spojení více datových proudů podporují pouze režim přidávání. Dotazy zapisují odpovídající záznamy pro každou dávku.

Pro vnitřní spojení databricks doporučuje nastavit prahovou hodnotu meze u každého streamovaného zdroje dat, aby dotaz mohl zahodit informace o stavu starých záznamů. Bez použití vodoznaků se Structured Streaming při každém spuštění pokusí spojit všechny klíče z obou stran spojení, což může ovlivnit výkon.

U vnějších spojení je vodoznak povinný. Pokud je záznam nesouvisený, dotaz zapíše hodnotu null pro tento klíč. Protože spojení podporují pouze režim přidávání, nespárované záznamy se nezapíšou, dokud není překročen práh zpoždění.

Řiďte práh pozdních dat pomocí zásady více vodoznaků

U více vstupů strukturovaného streamování můžete nastavit více vodoznaků pro řízení prahových hodnot tolerance pro pozdní data. Vodoznaky umožňují řídit informace o stavu a latenci.

Streamovací dotaz může mít více vstupních datových proudů, které jsou sjednocovány nebo spojeny dohromady. U stavových operací může každý vstupní datový proud vyžadovat jinou prahovou hodnotu pro pozdní toleranci dat. Tyto prahové hodnoty zadejte pomocí withWatermark("eventTime", delay) pro každý vstupní proud. Následuje příklad dotazu s spojeními stream-stream.

Python

input_stream1 = ...      # delays up to 1 hour
input_stream2 = ...      # delays up to 2 hours

(input_stream1.withWatermark("eventTime1", "1 hour")
  .join(
    input_stream2.withWatermark("eventTime2", "2 hours"),
    joinCondition)
)

Scala

val inputStream1 = ...      // delays up to 1 hour
val inputStream2 = ...      // delays up to 2 hours

inputStream1.withWatermark("eventTime1", "1 hour")
  .join(
    inputStream2.withWatermark("eventTime2", "2 hours"),
    joinCondition)

Při spouštění dotazu se stavovými operacemi sleduje strukturované streamování jednotlivě maximální dobu události pro každý vstupní datový proud, vypočítá vodoznaky na základě odpovídajícího zpoždění a určí jeden globální vodoznak. Strukturované streamování ve výchozím nastavení používá jako globální vodoznak minimum. Pokud některý datový proud zaostává za ostatními, minimální globální vodotisk zabrání tomu, aby dotaz omylem označil data jako opožděná. K tomu může dojít například v případě, že některý z datových proudů přestane přijímat data kvůli selháním nadřazeného datového proudu. Globální vodoznak se bezpečně pohybuje tempem nejpomalejšího datového proudu a v případě potřeby zpožďuje výstup dotazu.

Chcete-li snížit latenci, nastavte spark.sql.streaming.multipleWatermarkPolicy na hodnotu max (výchozí hodnota je min), aby se jako globální vodoznak použil vodoznak nejrychlejšího datového proudu. Tato konfigurace však zahodí data z nejpomalejších datových proudů. Databricks doporučuje, abyste tuto konfiguraci použili opatrně.

Použití vodoznaků u jedinečných operací

Operace distinct sleduje všechny jedinečné záznamy v rámci stavu. Bez vodoznaku se stav natrvalo rozrůstá a může způsobit problémy s pamětí. Zadejte vodoznak v poli časového razítka pro svázaný stav a po uplynutí prahové hodnoty odeberte staré záznamy.

Následující příklad použije vodoznak na distinct operaci:

Python

streamingDf = spark.readStream. ...  # columns: eventTime, id, value, ...

# Apply watermark before distinct operation
(streamingDf
  .withWatermark("eventTime", "1 hour")
  .distinct()
)

Scala

val streamingDf = spark.readStream. ...  // columns: eventTime, id, value, ...

// Apply watermark before distinct operation
streamingDf
  .withWatermark("eventTime", "1 hour")
  .distinct()

V tomto příkladu dotaz streamování odebere duplicitní záznamy, které přicházejí do 1 hodiny od posledního zjištěného stavu eventTime. Dotaz zahodí informace o stavu pro odstranění duplicit po uplynutí prahové hodnoty.

Důležité

Chcete-li místo všech sloupců odstranit duplicity v konkrétních sloupcích, použijte dropDuplicates() nebo dropDuplicatesWithinWatermark() místo distinct. Viz Odstranění duplicit vodoznaku.

Přetažení duplicit v rámci vodoznaku

Ve službě Databricks Runtime 13.3 LTS nebo vyšší můžete k odstranění duplicitních záznamů v rámci prahové hodnoty meze použít jedinečný identifikátor.

Strukturované streamování zaručuje zpracování přesně jednou, ale neodstraňuje záznamy ze zdrojů dat. Slouží dropDuplicatesWithinWatermark k odebrání duplicit u libovolného pole, i když se pole liší mezi duplicitními záznamy, jako je čas události nebo čas příjezdu.

U dropDuplicatesWithinWatermark dotazů se záznamy, které dorazí v rámci prahové hodnoty watermarku, vždy deduplikují. Dotazy můžou také odstranit duplicitní záznamy, které přicházejí mimo prahovou hodnotu, ale to není zaručené. Pokud chcete zaručit, že dotazy zahodí všechny duplicity, nastavte prahovou hodnotu meze tak, aby byla větší než maximální rozdíl časového razítka mezi duplicitními událostmi.

Chcete-li použít metodu dropDuplicatesWithinWatermark , musíte zadat vodoznak:

Python

streamingDf = spark.readStream. ...

# deduplicate using guid column with watermark based on eventTime column
(streamingDf
  .withWatermark("eventTime", "10 hours")
  .dropDuplicatesWithinWatermark(["guid"])
)

Scala

val streamingDf = spark.readStream. ...  // columns: guid, eventTime, ...

// deduplicate using guid column with watermark based on eventTime column
streamingDf
  .withWatermark("eventTime", "10 hours")
  .dropDuplicatesWithinWatermark(Seq("guid"))

Příklady případů použití

Následující příklady ukazují pokročilé případy použití oken:

Použití přeskakujících oken k výpočtu celkových tržeb za hodinu

Překlápěcí okna mají pevnou velikost a nepřekrývající se intervaly. Každý vstupní řádek patří přesně do jednoho okna. Pomocí přeskakujících oken můžete vypočítat diskrétní agregace časového období, jako jsou celkové tržby po hodinách:

Python

from pyspark.sql.functions import window, sum

hourly_sales = (orders
  .withWatermark("timestamp", "1 hour")
  .groupBy(window("timestamp", "1 hour"))
  .agg(sum("amount").alias("total_sales"))
)

Scala

import org.apache.spark.sql.functions.{window, sum}

val hourlySales = orders
  .withWatermark("timestamp", "1 hour")
  .groupBy(window($"timestamp", "1 hour"))
  .agg(sum($"amount").alias("total_sales"))

V tomto příkladu:

• window("timestamp", "1 hour") seskupuje objednávky do nepřekrývajících se jednohodinových intervalů, například od 5 do 6 hodin a od 6 do 7 hodin.
• withWatermark("timestamp", "1 hour") uchovává agregaci každého okna ve stavu, dokud časové razítko konce okna nebude o 1 hodinu starší než maximální časové razítko objednávky.

Pomocí posuvných oken vypočítejte průběžné agregace

Posuvná okna mají pevnou velikost s intervaly, které se můžou překrývat. Jeden řádek může patřit do více oken. Pomocí posuvných oken můžete vypočítat souhrnné agregace, jako je prodej za 6 hodin:

Python

from pyspark.sql.functions import window, sum

rolling_sales = (orders
  .withWatermark("timestamp", "1 hour")
  .groupBy(window("timestamp", "6 hours", slideDuration="1 hour"))
  .agg(sum("amount").alias("total_sales"))
)

Scala

import org.apache.spark.sql.functions.{window, sum}

val rollingSales = orders
  .withWatermark("timestamp", "1 hour")
  .groupBy(window($"timestamp", "6 hours", "1 hour"))
  .agg(sum($"amount").alias("total_sales"))

V tomto příkladu:

• window("timestamp", "6 hours", slideDuration="1 hour") seskupuje objednávky do 6hodinových intervalů, které se posouvají o 1 hodinu, například od 5:00 do 11:00 a od 6:00 do 12:00.
• withWatermark("timestamp", "1 hour") udržuje agregovanou hodnotu každého okna ve stavu, dokud časové razítko konce okna není o 1 hodinu starší než maximální časové razítko objednávky.
• slideDuration musí být menší než nebo rovno hodnotě windowDuration.

Pomocí oken relace můžete kontrolovat aktivitu uživatelů

Okna relací nemají pevnou velikost. Okno se otevře, když dorazí řádek, a uzavře se po uplynutí doby, během níž nedorazí žádné nové řádky. Pomocí oken relací můžete agregovat shluky aktivit oddělené dlouhými obdobími nečinnosti, například zobrazení stránek uživatelem během 30 minut:

Python

from pyspark.sql.functions import session_window, sum

sessionized_page_views = (activity
  .withWatermark("timestamp", "1 hour")
  .groupBy("user_id", session_window("timestamp", gapDuration="30 minutes"))
  .agg(sum("page_views").alias("total_page_views"))
)

Scala

import org.apache.spark.sql.functions.{session_window, sum}

val sessionizedPageViews = activity
  .withWatermark("timestamp", "1 hour")
  .groupBy($"user_id", session_window($"timestamp", "30 minutes"))
  .agg(sum($"page_views").alias("total_page_views"))

V tomto příkladu:

• session_window("timestamp", gapDuration="30 minutes") otevře okno při příchodu prvního zobrazení stránky. Každé další zobrazení stránky, které dorazí do 30 minut, rozšiřuje okno. Když do 30 minut nedorazí žádné zobrazení stránky, okno se zavře a další zobrazení stránky spustí nové okno.
• withWatermark("timestamp", "1 hour") uchovává agregovaná data každé relace ve stavu, dokud časové razítko konce okna není o 1 hodinu starší než nejvyšší časové razítko zobrazení stránky.
• Argument timeColumn pro window() a session_window() musí být typu TimestampType nebo TimestampNTZType.
• Slouží current_timestamp() k definování oken na základě času zpracování, nikoli času události.
• Dobu trvání okna můžete nastavit od mikrosekund až po dny. Doba trvání měsíce a delší nejsou podporována.
• Použijte výstupní režim complete s okenními agregacemi, aby se veškerý stav oken uchovával na neomezenou dobu. Pomocí append výstupního režimu s odpovídajícím vodoznakem můžete svázat růst stavu a zabránit problémům s pamětí u velkých datových sad. Další podrobnosti o chování výstupního režimu najdete v tématu Vodoznaky a režim výstupu pro agregace s okny.