Vytvořte vlastní stavovou aplikaci

Můžete použít transformWithState k vytváření stavových streamovacích aplikací a implementaci řešení s nízkou latencí a téměř v reálném čase. Pomocí vlastních stavových operátorů můžete vytvořit libovolnou stavovou logiku, která umožňuje vytvářet nové případy použití provozu, které nejsou možné pomocí tradičního zpracování strukturovaného streamování.

Poznámka

V případě stavových operací, jako jsou agregace, odstranění duplicitních dat a spojení streamování, databricks doporučuje místo vlastní logiky používat integrované operátory strukturovaného streamování. Pro více informací viz Co je stavové streamování?.

Databricks doporučuje používat transformWithState namísto zastaralých operátorů, jako jsou flatMapGroupsWithState a mapGroupsWithState, pro transformace libovolného stavu. Viz zastaralé libovolné stavové operátory.

Požadavky

Operátory transformWithState a transformWithStateInPandas mají následující požadavky:

• K dispozici v Databricks Runtime 16.2 a novějších.
  • V režimu v reálném čase použijte Databricks Runtime 17.3 LTS nebo vyšší. Viz režim v reálném čase ve strukturovaném streamování.
  • Pro standardní režim přístupu je Python k dispozici ve službě Databricks Runtime 16.3 a novější a scala je dostupná v Databricks Runtime 17.3 a vyšší.
• RocksDB je výchozí zprostředkovatel úložiště stavů v Databricks Runtime 17.3 a novější.

  • Pro Databricks Runtime 17.2 a novější musíte nakonfigurovat poskytovatele úložiště stavů RocksDB. Databricks doporučuje povolit RocksDB v konfiguraci Sparku.

    spark.conf.set("spark.sql.streaming.stateStore.providerClass", "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")
    

Co je transformWithState?

Operátor transformWithState použije vlastní stavový procesor na dotaz strukturovaného streamování. Je nutné implementovat vlastní stavový procesor pro použití transformWithState. Strukturované streamování zahrnuje rozhraní API pro vytváření stavových procesorů s použitím Pythonu, Scaly nebo Javy.

Pomocí transformWithState použijete vlastní logiku na seskupovací klíč. Následující informace popisují návrh vysoké úrovně:

• Definujte jednu nebo více proměnných stavu.
• Informace o stavu se zachovají pro každý klíč seskupení. Ke každé stavové proměnné můžete přistupovat v uživatelem definovaném kódu.
• Pro každou zpracovávanou mikrodávku jsou všechny řádky pro daný klíč k dispozici ve formě iterátoru.
• Použijte StatefulProcessorHandle s časovači a uživatelsky definovanými podmínkami k určení, jak se mají řádky emitovat.
• Pokud chcete spravovat vypršení platnosti stavu a velikost stavu, hodnoty stavu podporují jednotlivé definice TTL (Time to Live).

Protože transformWithState podporuje vývoj schématu v úložišti stavů, můžete iterovat a aktualizovat produkční aplikace bez ztráty historických informací o stavu. Po aktualizaci schématu stavu nemusíte znovu zpracovávat řádky, což zjednodušuje nasazení kódu a údržbu. Viz vývoj schématu v úložišti stavů.

Důležitý

Dokumentace ke službě Azure Databricks používá transformWithState k popisu implementací v jazycích Python a Scala:

• PySpark podporuje jak na řádcích založené rozhraní API transformWithState, tak na Pandas založený operátor transformWithStateInPandas.
  • transformWithStateInPandas není podporován v režimu v reálném čase. Místo toho použijte transformWithState. Podrobnosti najdete transformWithState v režimu v reálném čase.
• Scala podporuje pouze rozhraní API založené na transformWithState řádcích.

Implementace Scala a Python transformWithState mají stejné funkce, ale s některými rozdíly v syntaxi.

Definovat StatefulProcessor

Definujete stavový procesor rozšířením StatefulProcessor třídy a implementací jejích metod.

Spark předá StatefulProcessorHandle metodě init vašeho StatefulProcessor. Pomocí identifikátoru můžete vytvářet proměnné stavu a interagovat s úložištěm stavu.

transformWithState podporuje tři typy stavů: ValueState, ListStatea MapState. Každý typ ukládá stav pro každý klíč seskupení pomocí jiné podkladové datové struktury.

Implementujte následující metody pro definování vlastní logiky:

• Implementujte handleInputRows , abyste mohli řídit, jak aplikace zpracovává data, aktualizuje stav a generuje řádky pro každou mikrodávku. Viz Zpracování vstupních řádků.
• Implementujte handleExpiredTimer logiku založenou na čase bez ohledu na to, jestli klíč seskupení přijímá nové řádky v mikrodávce. Viz Popisovač časovačů s vypršenou platností.
• Volitelně můžete implementovat handleInitialState, aby se stav předem naplnil dříve, než aplikace zpracuje jakékoli vstupní řádky. Viz Popisovač počátečního stavu.

Následující tabulka porovnává funkční chování těchto metod:

Chování	handleInputRows	handleExpiredTimer
Získání, vložení, aktualizace nebo vymazání hodnot stavu	Ano	Ano
Vytvoření nebo odstranění časovače	Ano	Ano
Generování řádků	Ano	Ano
Procházet řádky v aktuální mikrodávce	Ano	Ne
Logika triggeru založená na uplynulé době	Ne	Ano

Podle potřeby můžete zkombinovat jak handleInputRows, tak handleExpiredTimer, a implementovat tak složitou logiku.

Můžete například implementovat aplikaci, která používá handleInputRows k aktualizaci hodnot stavu pro každou mikrodávku a nastavit časovač 10 sekund v budoucnu. Pokud se nezpracují žádné další řádky, můžete použít handleExpiredTimer k vygenerování aktuálních hodnot v úložišti stavů. Pokud se pro klíč seskupení zpracovávají nové řádky, můžete vymazat existující časovač a nastavit nový časovač.

StatefulProcessorHandle

V PySpark umožňuje StatefulProcessorHandle třída přístup k funkcím, které řídí, jak váš kód používá informace o stavu.

Při inicializaci StatefulProcessor je nutné vždy importovat a předat StatefulProcessorHandle do proměnné handle. Proměnná handle sváže místní proměnnou ve vaší třídě Python s proměnnou stavu.

Poznámka

Scala používá metodu getHandle.

vlastní typy stavů

V jednom stavovém operátoru můžete implementovat více stavových objektů.

Zvolte typ stavu na základě kompletní logiky aplikace. Můžete například sledovat relace pomocí ValueState, seskupené podle user_id a session_id. Nebo pokud chcete vyhodnotit podmínky napříč více relacemi, použijte MapState seskupený podle user_id, kde session_id bude klíčem mapy.

Pokud váš stavový objekt používá StructType, musíte definovat jedinečné názvy pro každé pole ve struktuře schématu. Tyto názvy jsou viditelné při čtení úložiště stavu. Viz Informace o stavu strukturovaného streamování.

Následující části popisují typy stavů, které transformWithStatepodporují:

ValueState

ValueState ukládá hodnotu pro každý klíč seskupení.

Stav hodnoty může zahrnovat komplexní typy, například strukturu nebo řazenou kolekci členů. Pro ValueState musíte implementovat logiku pro nahrazení celé hodnoty.

Doba životnosti stavu hodnoty se při aktualizaci hodnoty resetuje. Pokud zpracujete zdrojový klíč pro ValueState, aniž byste aktualizovali uložený ValueState, doba životnosti se neobnoví.

ListState

ListState ukládá seznam pro každý klíč seskupení.

Stav seznamu je kolekce hodnot, z nichž každý může obsahovat složité typy. Každá hodnota v seznamu má vlastní dobu životnosti.

Položky můžete přidat do seznamu tak, že připojíte jednotlivé položky, připojíte seznam položek nebo přepíšete celý seznam pomocí put. Pokud chcete resetovat čas naživo, musíte použít put operaci.

MapState

MapState uloží mapu pro každý klíč seskupení. Mapy jsou ekvivalentem Apache Sparku ke slovníku Python (dict).

Stav mapy je kolekce jedinečných klíčů, které každý mapuje na hodnotu, z nichž každý může obsahovat komplexní typy. Každý pár klíč-hodnota v mapě má vlastní dobu platnosti.

Můžete aktualizovat hodnotu konkrétního klíče nebo můžete klíč a jeho hodnotu odebrat. Jednotlivou hodnotu můžete vrátit pomocí klíče, zobrazit seznam všech klíčů, zobrazit seznam všech hodnot nebo vrátit iterátor pro práci s úplnou sadou párů klíč-hodnota v mapě.

Důležitý

Klíče seskupení popisují pole zadaná v klauzuli GROUP BY dotazu strukturovaného streamování. Stavy mapy mohou obsahovat libovolný počet dvojic klíč–hodnota pro seskupovací klíč.

Pokud například dotaz používá GROUP BY user_id a chcete definovat mapu pro každý z nich session_id, klíč seskupení je user_id a MapState klíč je session_id:

Python

class SessionTracker(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    self.sessions = handle.getMapState("sessions", StringType(), LongType())

  def handleInputRows(self, key, rows: Iterator[Row], timerValues) -> Iterator[Row]:
    for row in rows:
      session_id = row["session_id"]  # session_id is the MapState key
      count = self.sessions.getValue(session_id)[0] if self.sessions.containsKey(session_id) else 0
      new_count = count + 1
      self.sessions.updateValue(session_id, (new_count,))
    yield from []

  def close(self) -> None:
    pass

df.groupBy("user_id").transformWithState(SessionTracker(), ...) # user_id is the grouping key

Scala

case class Event(userId: String, sessionId: String)

class SessionTracker extends StatefulProcessor[String, Event, Row] {
  @transient private var sessions: MapState[String, Long] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    sessions = getHandle.getMapState[String, Long]("sessions", Encoders.STRING, Encoders.scalaLong, TTLConfig.NONE)
  }

  override def handleInputRows(
      key: String,
      rows: Iterator[Event],
      timerValues: TimerValues): Iterator[Row] = {
    rows.foreach { event =>
      val count = if (sessions.containsKey(event.sessionId)) sessions.getValue(event.sessionId) else 0L
      sessions.updateValue(event.sessionId, count + 1) // sessionId is the MapState key
    }
    Iterator.empty
  }
}

df.as[Event]
  .groupByKey(_.userId) // userId is the grouping key
  .transformWithState(new SessionTracker(), TimeMode.None(), OutputMode.Update())

Vytvořte vlastní proměnnou stavu v StatefulProcessor

Při inicializaci StatefulProcessorvytvoříte místní proměnnou pro každý objekt stavu, který umožňuje interakci se stavovými objekty ve vlastní logice. Definujte a inicializujete stavové proměnné přepsáním předdefinované init metody ve StatefulProcessor třídě.

Můžete definovat libovolný počet stavových objektů pomocí getValueState, getListStatea getMapState metody ve vašem StatefulProcessor.

Každý objekt stavu musí mít následující:

• Jedinečný název
• Schéma
  • V Python je nutné zadat schéma.
  • V jazyce Scala můžete předat Encoder k určení schématu stavu.

Volitelně můžete také zadat dobu trvání TTL (Time to Live) v milisekundách. Pokud implementujete stav mapování, musíte zadat samostatnou definici schématu pro klíče mapování a hodnoty.

Poznámka

StatefulProcessor zpracovává logiku odděleně pro dotazování, aktualizaci a publikování informací o stavu. Viz Použití stavových proměnných v metodách s vlastní logikou.

Použijte stavové proměnné v metodách s vlastní logikou

Objekty stavu mají metody pro získání stavu, aktualizaci existujících informací o stavu a vymazání aktuálního stavu.

Každý klíč seskupení má vyhrazené informace o stavu.

• StatefulProcessor generuje řádky na základě vaší vlastní logiky a zadaného výstupního schématu. Viz Generování řádků.
• K přístupu k hodnotám v úložišti stavu použijte čtecí modul statestore. Tato čtečka je určená pro dávkové úlohy a není určená pro úlohy s nízkou latencí. Viz Informace o stavu strukturovaného streamování.
• Logika zadaná pomocí handleInputRows se spustí jenom v případě, že jsou řádky pro klíč přítomné v mikrodávce. Viz Zpracování vstupních řádků.
• Použijte handleExpiredTimer k implementaci logiky založené na čase, která pro spuštění nezávisí na sledování řádků. Viz Popisovač časovačů s vypršenou platností.

Poznámka

Objekty stavu jsou izolované pomocí seskupení klíčů s následujícími důsledky:

• Hodnoty stavu nemohou být ovlivněny řádky přidruženými k jinému klíči seskupení.
• nemůžete implementovat logiku, která závisí na porovnávání hodnot nebo aktualizaci stavu napříč klíči seskupení.

Hodnoty v rámci klíče seskupení můžete porovnat. Použijte MapState k implementaci logiky s druhým klíčem, který může vaše vlastní logika použít. Například seskupení podle user_id a použití ip_address pro váš klíč MapState umožňuje sledovat souběžné uživatelské relace.

Pokročilé aspekty práce se stavem

Aktualizace stavu jsou odolné proti chybám. Pokud se úloha chybově ukončí před dokončením zpracování mikrodávkové dávky, použije opakování hodnotu z poslední úspěšné mikrodávkové dávky.

Pro optimalizovaný výkon doporučuje Databricks zpracovat všechny hodnoty v iterátoru pro daný klíč a potvrdit aktualizace v jednom zápisu. Při zápisu do stavové proměnné se tím aktivuje zápis do RocksDB.

Hodnoty stavu nemají výchozí hodnoty. Pokud vaše logika vyžaduje čtení existujících informací o stavu, použijte metodu exists .

Chcete-li implementovat logiku pro hodnotu null, proměnné MapState vám umožňují ověřovat jednotlivé klíče nebo vypsat všechny klíče.

Zpracování vstupních řádků

handleInputRows Pomocí metody definujte, jak aplikace zpracovává řádky a aktualizuje hodnoty stavu. Tato metoda se spustí pokaždé, když váš dotaz Structured Streaming zpracovává řádky pro daný klíč seskupení.

Pro většinu stavových aplikací implementovaných s transformWithStateje základní logika definována pomocí handleInputRows.

Pro každou zpracovanou aktualizaci mikrodávkové dávky jsou všechny řádky v mikrodávce pro daný klíč seskupení k dispozici pomocí iterátoru. Uživatelsky definovaná logika může pracovat se všemi řádky z aktuální mikrobatch a hodnot v úložišti stavů.

Zpracování časovačů s vypršenou platností

Použijte metodu handleExpiredTimer k implementaci vlastní logiky na základě uplynulého času.

V rámci klíče seskupení jsou časovače jednoznačně identifikovány jejich časovým razítkem.

Když vyprší platnost časovače, výsledek se určuje logikou implementovanou ve vaší aplikaci. Mezi běžné vzory patří:

• Vysílání informací uložených v proměnné stavu
• Vyřazení uložených informací o stavu
• Vytvoření nového časovače

Časovače s vypršenou platností se aktivují i v případě, že se v mikrodávce nezpracují žádné řádky pro jejich přidružený klíč.

Určení časového režimu

Při předávání StatefulProcessor do transformWithState, musíte určit časový režim pomocí parametru timeMode .

Podporují se následující možnosti:

Časový režim	Popis
ProcessingTime	Podporovány jsou jak časovače, tak hodnoty TTL, a vyhodnocují se na základě reálného času v okamžiku, kdy Apache Spark zpracovává každou mikrodávku. Použijte ProcessingTime, když chcete, aby se časovače spouštěly v pevných intervalech vzhledem k okamžiku zpracování řádků, bez ohledu na časová razítka v datech.
EventTime	Časovače jsou podporovány a vyhodnocovány na základě vodoznaku času události. Vodoznak se posune tak, jak Apache Spark sleduje časové razítka ve vstupních datech. TTL není podporováno u EventTime. Použijte EventTime, když vaše data obsahují časová razítka a chcete, aby se časovače spouštěly podle průběhu těchto časových razítek. Při použití EventTime je nutné také zadat parametr eventTimeColumnName. Viz eventTimeColumnName.
NoTime nebo TimeMode.None()	Časovače a hodnota TTL se nepodporují. Použijte NoTime , když stavová aplikace nevyžaduje logiku založenou na čase.

eventTimeColumnName

Při použití EventTime časového režimu eventTimeColumnName určuje parametr název sloupce ve výstupním schématu, který obsahuje časové razítko události. Apache Spark používá tento sloupec k přenášení vodoznaku do výstupního datového toku, což umožňuje správné následné operace založené na čase.

Python

eventTimeColumnName je dalším argumentem pro transformWithState nebo transformWithStateInPandas:

q = (
  df.groupBy("key")
    .transformWithState(
      statefulProcessor=MyProcessor(),
      outputStructType=output_schema,
      outputMode="Append",
      timeMode="EventTime",
      eventTimeColumnName="outputTimestamp",
    )
    .writeStream...
)

Scala

transformWithState přijímá eventTimeColumnName místo timeMode. Tento přístup vždy používá EventTime režim:

val q = spark
  .readStream
  .format("delta")
  .load(srcDeltaTableDir)
  .as[(String, String)]
  .groupByKey(x => x._1)
  .transformWithState(
    new MyProcessor(),
    "outputTimestamp",
    OutputMode.Append(),
  )
  .writeStream...

Předdefinované hodnoty časovače

Databricks doporučuje nevyvolání systémových hodin ve vlastní stavové aplikaci, protože to může vést k nespolehlivému opakování při selhání úlohy. Metody ve třídě TimerValues použijte, když potřebujete získat přístup k času zpracování nebo vodoznaku:

TimerValues	Popis
getCurrentProcessingTimeInMs	Vrátí časové razítko času zpracování aktuální dávky v milisekundách od počátku epochy.
getCurrentWatermarkInMs	Vrátí časové razítko vodotisku pro aktuální dávku v milisekundách od počátku epochy.

Poznámka

Doba zpracování popisuje čas, kdy je mikrodávka zpracována Apache Sparkem. Mnoho streamovacích zdrojů, jako je Kafka, zahrnuje také dobu zpracování systému.

Vodoznaky u streamovaných dotazů se často definují proti času události nebo času zpracování zdroje streamování. Viz Použití vodoznaků k řízení prahových hodnot zpracování dat.

Vodoznaky i okna lze použít v kombinaci s transformWithState. Ve své vlastní stavové aplikaci můžete podobné funkce implementovat s využitím TTL, časovačů a funkce MapState nebo ListState.

Doba životnosti (TTL) pro typy stavů

Chcete-li zabránit chybám nedostatku paměti a odebrat zastaralé hodnoty typu stavu, transformWithState podporuje volitelnou hodnotu TTL (Time to Live) pro každou hodnotu typu stavu. Po vypršení platnosti TTL automaticky odstraní hodnoty stavového typu. TTL nevykonává handleExpiredTimer ani žádnou vlastní logiku. Pokud chcete spustit kód po vypršení platnosti stavu, použijte místo toho časovač.

Důležitý

Pokud neimplementujete TTL, musíte zajistit odstraňování stavu, aby nedocházelo k chybám způsobeným nedostatkem paměti.

U všech typů stavů se hodnota TTL resetuje při aktualizaci informací o stavu. Hodnota TTL se vynucuje pro každou hodnotu typu stavu s různými pravidly pro každý typ stavu:

• Stavové proměnné jsou vztaženy ke seskupovacím klíčům.
• Pro ValueState objekty je uložena pouze jedna hodnota na klíč seskupení. Hodnota TTL se vztahuje na tuto hodnotu.
• Pro ListState objekty může seznam obsahovat mnoho hodnot. Hodnota TTL se vztahuje na každou hodnotu v seznamu nezávisle.
  • Zatímco hodnota TTL je vymezena na jednotlivé hodnoty v objektu ListState, jediný způsob, jak aktualizovat jednotlivé hodnoty, je put metoda, která přepíše celý obsah ListState proměnné a resetuje hodnotu TTL pro všechny hodnoty v seznamu.
• Pro MapState objekty má každý klíč mapy přidruženou hodnotu stavu. TTL se použije nezávisle na každém páru klíč-hodnota v mapě.

Poznámka

Časovače umožňují definovat vlastní logiku i mimo vyřazení stavu, včetně odesílání řádků. Volitelně lze pomocí časovačů jak vymazat stavové informace pro danou hodnotu stavu, tak emitovat hodnoty nebo spustit podmíněnou logiku. Viz Popisovač časovačů s vypršenou platností.

Příklad stavové aplikace

Následující příklad definuje vlastní stavový procesor, SimpleCounterProcessorvčetně ukázkových stavových proměnných. SimpleCounterProcessor používá ValueState, ListStatea MapState k počítání řádků pro každý klíč seskupení.

Python (Pandas)

import pandas as pd
from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from typing import Iterator

spark.conf.set("spark.sql.streaming.stateStore.providerClass","org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

output_schema = StructType(
    [
        StructField("id", StringType(), True),
        StructField("countAsString", StringType(), True),
    ]
)

class SimpleCounterProcessor(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    value_state_schema = StructType([StructField("count", IntegerType(), True)])
    list_state_schema = StructType([StructField("count", IntegerType(), True)])
    self.value_state = handle.getValueState(stateName="valueState", schema=value_state_schema)
    self.list_state = handle.getListState(stateName="listState", schema=list_state_schema)
    # Schema can also be defined using strings and SQL DDL syntax
    self.map_state = handle.getMapState(stateName="mapState", userKeySchema="name string", valueSchema="count int")

  def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
    count = 0
    for pdf in rows:
      list_state_rows = [(120,), (20,)] # A list of tuples
      self.list_state.put(list_state_rows)
      self.list_state.appendValue((111,))
      self.list_state.appendList(list_state_rows)
      pdf_count = pdf.count()
      count += pdf_count.get("value")
    self.value_state.update((count,)) # Count is passed as a tuple
    iter = self.list_state.get()
    list_state_value = next(iter)[0]
    value = count
    user_key = ("user_key",)
    if self.map_state.exists():
      if self.map_state.containsKey(user_key):
        value += self.map_state.getValue(user_key)[0]
    self.map_state.updateValue(user_key, (value,)) # Value is a tuple
    yield pd.DataFrame({"id": key, "countAsString": str(count)})

q = (df.groupBy("key")
  .transformWithStateInPandas(
    statefulProcessor=SimpleCounterProcessor(),
    outputStructType=output_schema,
    outputMode="Update",
    timeMode="None",
  )
  .writeStream...
)

Python (založené na řádcích)

from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from typing import Iterator

spark.conf.set("spark.sql.streaming.stateStore.providerClass", "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

output_schema = StructType(
  [
    StructField("id", StringType(), True),
    StructField("countAsString", StringType(), True),
  ]
)

class SimpleCounterProcessor(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    value_state_schema = StructType([StructField("count", IntegerType(), True)])
    list_state_schema = StructType([StructField("count", IntegerType(), True)])
    self.value_state = handle.getValueState(stateName="valueState", schema=value_state_schema)
    self.list_state = handle.getListState(stateName="listState", schema=list_state_schema)
    self.map_state = handle.getMapState(stateName="mapState", userKeySchema="name string", valueSchema="count int")

  def handleInputRows(self, key, rows: Iterator[Row], timerValues) -> Iterator[Row]:
    count = 0
    for row in rows:
      list_state_rows = [(120,), (20,)]  # A list of tuples
      self.list_state.put(list_state_rows)
      self.list_state.appendValue((111,))
      self.list_state.appendList(list_state_rows)
      count += 1
    self.value_state.update((count,))  # Count is passed as a tuple
    iter_list = self.list_state.get()
    list_state_value = next(iter_list)[0]
    value = count
    user_key = ("user_key",)
    if self.map_state.exists():
      if self.map_state.containsKey(user_key):
        value += self.map_state.getValue(user_key)[0]
    self.map_state.updateValue(user_key, (value,))  # Value is a tuple
    yield Row(id=key, countAsString=str(count))

q = (
  df.groupBy("key")
    .transformWithState(
      statefulProcessor=SimpleCounterProcessor(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
    )
    .writeStream...
)

Scala

import org.apache.spark.sql.streaming._
import org.apache.spark.sql.{Dataset, Encoder, Encoders , DataFrame}
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

spark.conf.set("spark.sql.streaming.stateStore.providerClass","org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

class SimpleCounterProcessor extends StatefulProcessor[String, (String, String), (String, String)] {
  @transient private var countState: ValueState[Int] = _
  @transient private var listState: ListState[Int] = _
  @transient private var mapState: MapState[String, Int] = _

  private val longEncoder = Encoders.scalaLong
  private val intEncoder = Encoders.scalaInt
  private val stringEncoder = Encoders.STRING

  override def init(
      outputMode: OutputMode,
      timeMode: TimeMode): Unit = {
    countState = getHandle.getValueState[Int]("countState",
      intEncoder, TTLConfig.NONE)
    listState = getHandle.getListState[Int]("listState",
      intEncoder, TTLConfig.NONE)
    mapState = getHandle.getMapState[String, Int]("mapState",
      stringEncoder, intEncoder, TTLConfig.NONE)
  }

  override def handleInputRows(
      key: String,
      inputRows: Iterator[(String, String)],
      timerValues: TimerValues): Iterator[(String, String)] = {
    var count = countState.getOption().getOrElse(0)
    for (row <- inputRows) {
      val listData = Array(120, 20)
      listState.put(listData)
      listState.appendValue(count)
      listState.appendList(listData)
      count += 1
    }
    val iter = listState.get()
    var listStateValue = 0
    if (iter.hasNext) {
      listStateValue = iter.next()
    }
    countState.update(count)
    var value = count
    val userKey = "userKey"
    if (mapState.exists()) {
      if (mapState.containsKey(userKey)) {
        value += mapState.getValue(userKey)
      }
    }
    mapState.updateValue(userKey, value)
    Iterator((key, count.toString))
  }
}

val q = spark
        .readStream
        .format("delta")
        .load("$srcDeltaTableDir")
        .as[(String, String)]
        .groupByKey(x => x._1)
        .transformWithState(
            new SimpleCounterProcessor(),
            TimeMode.None(),
            OutputMode.Update(),
        )
        .writeStream...

Další příklady najdete v tématu Příklad stavových aplikací.

Poznámka

V Pythonu jsou hodnoty stavu n-tice. Předejte n-tice do put a update a očekávejte n-tice z get.

Pokud je například schéma pro vaše ValueState celé číslo:

current_value_tuple = value_state.get() # Returns the value state as a tuple
current_value = current_value_tuple[0]  # Extracts the first item in the tuple
new_value = current_value + 1           # Calculate a new value
value_state.update((new_value,))        # Pass the new value formatted as a tuple

Tento přístup použijte také pro položky v ListState nebo hodnoty v MapState.

Generovat řádky

Musíte použít handleInputRows nebo handleExpiredTimer k definování toho, jak transformWithState generuje řádky pro každý seskupovací klíč. Viz Zpracování vstupních řádků a Zpracování expirovaných časovačů.

Vlastní stavové aplikace nijak nepředjímají způsob použití informací o stavu. Pro danou podmínku aplikace nemusí generovat žádné řádky, jeden řádek nebo mnoho řádků.

Poznámka

Můžete implementovat více hodnot stavu a definovat více podmínek pro generování řádků, ale všechny řádky musí používat stejné schéma.

Python (Pandas)

Pomocí transformWithStateInPandaspříkazu definujte výstupní schéma pomocí klíčového outputStructType slova.

Generování řádků pomocí objektu datového rámce pandas a yield.

Volitelně můžete yield použít prázdný datový rámec. Pokud použijete update výstupní režim a vygenerujete prázdný datový rámec, aktualizuje se hodnoty pro klíč seskupení, který má být null.

Python (založené na řádcích)

Pomocí transformWithStatepříkazu definujte výstupní schéma pomocí klíčového outputStructType slova.

Generování řádků pomocí objektu Row a yield.

Volitelně můžete vrátit prázdný iterátor. Pokud použijete update výstupní režim a vygenerujete prázdný iterátor, aktualizuje se hodnoty pro seskupovací klíč, který má být null.

Scala

V jazyce Scala generujete řádky pomocí objektu Iterator . Schéma se odvozuje automaticky ze schématu vygenerovaných řádků.

Případně můžete vrátit prázdné Iterator. Pokud použijete update výstupní režim a vygenerujete prázdný Iterator, aktualizuje se hodnoty pro seskupovací klíč, který má být null.

Zpracování počátečního stavu

Volitelně můžete první mikrodávce předat počáteční stav.

Můžete například použít tyto možnosti:

• Migrace existujícího pracovního postupu do nové vlastní aplikace
• Upgradujte stavový operátor a změňte schéma nebo logiku.
• Opravte chybu, která se nedá automaticky opravit a vyžaduje ruční zásah.

Poznámka

Pomocí čtečky úložiště stavů můžete dotazovat informace o stavu z existujícího kontrolního bodu. Viz Informace o stavu strukturovaného streamování.

Pokud převádíte existující tabulku Delta na stavovou aplikaci, přečtěte si tabulku pomocí spark.read.table("table_name") a předejte výsledný datový rámec. Volitelně můžete vybrat nebo upravit pole tak, aby odpovídala vaší nové stavové aplikaci.

Počáteční stav zadáte pomocí datového rámce se stejným schématem klíče seskupení jako vstupní řádky.

Poznámka

Python používá handleInitialState k určení počátečního stavu při definování StatefulProcessor. Scala používá odlišnou třídu StatefulProcessorWithInitialState.

Následující příklad inicializuje čítač pro každý klíč z existující tabulky Delta:

Python (založené na řádcích)

from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from typing import Iterator

class CounterWithInitialState(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    state_schema = StructType([StructField("count", IntegerType(), True)])
    self.count_state = handle.getValueState("countState", state_schema)

  def handleInitialState(self, key, initialState: Row, timerValues) -> None:
    self.count_state.update((initialState["count"],))

  def handleInputRows(self, key, rows: Iterator[Row], timerValues) -> Iterator[Row]:
    count = self.count_state.get()[0] if self.count_state.exists() else 0
    for _ in rows:
      count += 1
    self.count_state.update((count,))
    yield Row(id=key[0], count=count)

  def close(self) -> None:
    pass

output_schema = StructType([
  StructField("id", StringType(), True),
  StructField("count", IntegerType(), True),
])

# Load existing counts as initial state — must use the same grouping key as the input
initial_state = spark.read.table("existing_counts").groupBy("id")

q = (
  df.groupBy("id")
    .transformWithState(
      statefulProcessor=CounterWithInitialState(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
      initialState=initial_state,
    )
    .writeStream...
)

Scala

import org.apache.spark.sql.streaming._
import org.apache.spark.sql.Encoders

class CounterWithInitialState
    extends StatefulProcessorWithInitialState[String, (String, String), (String, String), (String, Int)] {

  @transient private var countState: ValueState[Int] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    countState = getHandle.getValueState[Int]("countState", Encoders.scalaInt, TTLConfig.NONE)
  }

  override def handleInitialState(
      key: String, initialState: (String, Int), timerValues: TimerValues): Unit = {
    countState.update(initialState._2)
  }

  override def handleInputRows(
      key: String,
      rows: Iterator[(String, String)],
      timerValues: TimerValues): Iterator[(String, String)] = {
    val count = if (countState.exists()) countState.get() else 0
    val newCount = count + rows.size
    countState.update(newCount)
    Iterator((key, newCount.toString))
  }
}

// Load existing counts as initial state — must use the same grouping key as the input
val initialState = spark.read.table("existing_counts")
  .as[(String, Int)]
  .groupByKey(_._1)

val q = spark
  .readStream
  .format("delta")
  .load(srcDeltaTableDir)
  .as[(String, String)]
  .groupByKey(_._1)
  .transformWithState(
    new CounterWithInitialState(),
    TimeMode.None(),
    OutputMode.Update(),
    initialState,
  )
  .writeStream...

Použití transformWithState v pipelinech Lakeflow

Pomocí operátora transformWithState v kanálech Lakeflow můžete implementovat libovolnou stavovou logiku v kanálech streamování pomocí Python.

Provedete to provedením následujících kroků:

1. Definujte výstupní schéma a logiku stavového procesoru pro libovolné stavové transformace. Příklady najdete v příkladu stavových aplikací.
2. Vytvořte tok v kanálu Lakeflow, který volá operátor transformWithState nad objektem DataFrame. Viz kurz: Vytvoření prvního kanálu pomocí Editoru kanálů Lakeflow.
3. Spusťte potrubí a ověřte výsledky v cílové tabulce nebo úložišti.

Příklad, který používá transformWithState k monitorování prezenčních signálů snímače, viz Příklad: Použití transformWithState k monitorování prezenčních signálů snímače.