Asynchronní zpracování s transformWithState

Python transformWithState podporuje asynchronní zpracování založené na asyncio. Současným spouštěním stavových operací a uživatelské logiky napříč seskupovacími klíči a dávkováním meziprocesové komunikace má asynchronní zpracování vyšší propustnost než synchronní zpracování s pouze drobnými změnami kódu. Toto zvýšení propustnosti nevyžaduje žádné asynchronní knihovny třetích stran. Pokročilí uživatelé mohou dále optimalizovat své aplikace pomocí asynchronních programovacích vzorů a asynchronních knihoven.

Asynchronní zpracování podporuje jak mikrodávkové, tak dotazy v režimu reálného času. Pro jeden seskupovací klíč asynchronní zpracování zachovává pořadí řádků. Různé seskupovací klíče běží současně.

Pro použití asynchronního zpracování implementujte AsyncStatefulProcessor místo synchronního StatefulProcessor. AsyncStatefulProcessor API zrcadlí synchronní StatefulProcessor API, takže většina aplikací vyžaduje pouze malé změny pro použití asynchronního API. Viz implementaci AsyncStatefulProcessor.

Pro synchronní transformWithState API a základní koncepty viz Vytvoření vlastní stavové aplikace pomocí transformWithState.

Requirements

Asynchronní zpracování má následující požadavky:

  • Dostupné v Databricks Runtime 19 a novějším, nebo na serverless výpočetním režimu s prostředím verze 6 a vyšším.
  • Dostupné pouze pro Python transformWithState API založené na řádcích. Asynchronní zpracování není podporováno pro transformWithStateInPandas ani pro rozhraní Scala transformWithState API.
  • V Lakeflow pipeline nakonfigurujte pipeline tak, aby používala kanál PREVIEW. Tento požadavek platí jak pro klasické, tak pro serverless výpočetní prostředí pro kanály. Kanál CURRENT nepodporuje asynchronní zpracování. Viz channel v konfiguracích kanálů a kanálech modulu runtime pro kanály Lakeflow.

Zaveďte AsyncStatefulProcessor

Chcete-li převést synchronní StatefulProcessor na AsyncStatefulProcessor, proveďte následující změny:

  • Definujte API metody (init, close, handleInputRows, handleExpiredTimer, a handleInitialState) pomocí async def klíčového slova.
  • Čtěte a aktualizujte hodnoty stavu a časovače pomocí await, nebo je spusťte pomocí knihovny asyncio v Pythonu. To platí pro stavové operace, jako jsou valueState.get(), a pro časové operace, například registerTimer. Vytváření stavových objektů, jako je handle.getValueState, zůstává synchronní.

Následující úvahy se vztahují na asynchronní zpracování:

  • Pokud vaše aplikace ukládá data do proměnných členů nebo externích systémů, Databricks doporučuje přepsat logiku tak, aby byla bezpečná pro souběžné spouštění. Protože handleInputRows a handleExpiredTimer mohou běžet současně napříč klíči pro seskupování, prokládané běhy nesmí poškozovat sdílená data. Většina aplikací už tuto podmínku splňuje.
  • Databricks doporučuje, abyste nechytali ani nepotláčeli chyby ze stavových operací. Apache Spark tyto chyby řeší za vás. Pokud stavová operace selže, Apache Spark úlohu označí jako neúspěšnou a zkusí ji znovu.
    • Ve stavu AsyncStatefulProcessor jsou chyby operací spravovány za vás a nikdy se neobjeví ve vašem kódu.
    • V synchronním StatefulProcessor se ve vašem kódu vyvolávají chyby operací se stavem, ale jejich potlačení může ohrozit správnost dat.

Příklad: spočítat řádky pro každý seskupovací klíč

Následující příklad definuje agregaci AsyncCountProcessor, která počítá počet řádků pro každý seskupovací klíč. Proměnná value_schema definuje schéma prvku ValueState, který ukládá průběžný počet. Ve srovnání se synchronním StatefulProcessor jsou rozdíly klíčové slovo async def u každé metody a await u operací čtení a aktualizace stavu. Volání v getValueState v init zůstává synchronní. Definujte procesor, jak je uvedeno v následujícím kódu:

from pyspark.sql import Row
from pyspark.sql.streaming import AsyncStatefulProcessor, AsyncStatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, LongType

value_schema = StructType([StructField("count", LongType(), True)])

class AsyncCountProcessor(AsyncStatefulProcessor):
  async def init(self, handle: AsyncStatefulProcessorHandle) -> None:
    self.count = handle.getValueState("count", value_schema)

  async def handleInputRows(self, key, rows, timerValues):
    total = (await self.count.get() or (0,))[0]
    for _ in rows:
      total += 1
    await self.count.update((total,))
    yield Row(action=key[0], count=total)

  async def close(self) -> None:
    pass

Spusť dotaz s asynchronním procesorem

Pro spuštění dotazu s asynchronním procesorem předejte svůj dotaz AsyncStatefulProcessor do transformWithState. Dotaz používá stejnou syntaxi jako synchronní cesta. Asynchronní a synchronní API sdílejí stejný formát stavu, takže můžete přepínat existující dotaz mezi asynchronním AsyncStatefulProcessor a synchronním StatefulProcessor dotazem a zároveň znovu použít stejný kontrolní bod.

Příklad: spočítat události v events ukázkové datové sadě

Následující příklad spustí AsyncCountProcessor nad ukázkovou datovou sadou events. Každý záznam má time pole (epochsekundy) a action pole s hodnotou Open nebo Close. Dotaz seskupuje podle action a počítá události pro každý typ akce. Pro více ukázkových datových sad, viz Ukázkové datové sady.

Proměnná input_schema definuje schéma zdrojových záznamů a proměnná output_schema určuje schéma řádků, které procesor generuje. Pro čtení ukázkové datové sady jako proudu definujte obě schémata a poté zahájte dotaz podle následujícího kódu:

from pyspark.sql.types import StructType, StructField, StringType, LongType

input_schema = StructType([
  StructField("time", LongType(), True),
  StructField("action", StringType(), True),
])

output_schema = StructType([
  StructField("action", StringType(), True),
  StructField("count", LongType(), True),
])

events = (
  spark.readStream.schema(input_schema)
    .option("maxFilesPerTrigger", 10)
    .json("/databricks-datasets/structured-streaming/events")
)

q = (
  events.groupBy("action")
    .transformWithState(
      statefulProcessor=AsyncCountProcessor(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
    )
    .writeStream.format("memory")
    .queryName("async_counts")
    .trigger(availableNow=True)
    .start()
)

q.awaitTermination()

Po dokončení dotazu uvidíte průběžný počet pro každý typ akce v následujícím kódu:

display(spark.sql("SELECT action, MAX(count) AS count FROM async_counts GROUP BY action ORDER BY action"))

Asynchronní operace stavu a časovače

V AsyncStatefulProcessor jsou operace stavových proměnných a časovačů, které čtou nebo zapisují hodnoty, asynchronní. Většina těchto operací vrací jeden výsledek, který získáte pomocí await. Operace, které vracejí kolekci, místo toho vracejí asynchronní iterátor, který používáte pomocí async for. Pro úvod do async/await a asynchronních iterátorů v Pythonu viz dokumentaci Python asyncio.

Následující tabulka uvádí operace, které vracejí jediný výsledek, který lze získat pomocí await:

Class Operace, které používají await
AsyncValueState exists, get, , updateclear
AsyncMapState exists, getValue, containsKey, updateValue, removeKey, clear
AsyncListState exists, put, appendValue, , appendListclear
AsyncStatefulProcessorHandle registerTimer, deleteTimer

Následující tabulka uvádí operace, které vracejí asynchronní iterátor, který lze získat pomocí async for:

Class Operace, které používají async for
AsyncMapState iterator, keys, values
AsyncListState get
AsyncStatefulProcessorHandle listTimers

Příklad: async for

Například pro čtení hodnot v AsyncListState iterujte s async for jako v následujícím kódu:

total = 0
async for value in self.items.get():
  total += value[0]

Metody, které vytvářejí stavové objekty a mažou stavové proměnné, zůstávají synchronní: getValueState, getMapState, getListState, a deleteIfExists.

Pro popis každého typu stavu viz Vlastní typy stavů.

Optimalizujte pomocí asynchronních programovacích vzorů

Asynchronní zpracování je užitečné, když vaše logika čeká na externí operace, jako jsou síťové požadavky. Místo čekání na každý požadavek postupně použijte asyncio ke spouštění požadavků souběžně a zkraťte dobu nečinnosti.

Příklad: spustit souběžné požadavky pomocí asyncio.gather

Následující příklad používá asyncio.gather ke spuštění všech HTTP požadavků na řádek současně a čeká na jejich dokončení, poté ukládá maximální skóre do stavu. Definujte procesor, jak je uvedeno v následujícím kódu:

import asyncio
import aiohttp
from pyspark.sql import Row
from pyspark.sql.streaming import AsyncStatefulProcessor

class HttpScoreRowGatherProcessor(AsyncStatefulProcessor):
  async def init(self, handle):
    self._score_state = handle.getValueState("last_score", "score double")
    self._session = aiohttp.ClientSession()

  async def _fetch_score(self, row) -> float:
    async with self._session.get(
      f"https://api.example.com/score/{row.event_id}"
    ) as resp:
      return (await resp.json())["score"]

  async def handleInputRows(self, key, rows, timerValues):
    user_id = key[0]
    scores = await asyncio.gather(*[self._fetch_score(row) for row in rows])

    max_score = max(scores)
    await self._score_state.update((max_score,))
    yield Row(user_id=user_id, score=max_score)

  async def close(self):
    await self._session.close()
  • Poslední aktualizace v