Asynchrone verwerking met transformWithState

Asynchrone verwerking voor de Python-rijgebaseerde transformWithState API vereist Databricks Runtime 19 en hoger, of serverless compute met omgevingsversie 6 en hoger.

Om asynchrone verwerking in een Lakeflow-pijplijn te gebruiken, configureer je de pijplijn om het PREVIEW-kanaal te gebruiken. Zie channel in Pijplijnconfiguraties.

Python transformWithState ondersteunt asynchrone verwerking gebouwd op asyncio. Door toestandsoperaties en gebruikerslogica gelijktijdig te laten uitvoeren over meerdere groeperingssleutels heen en interprocescommunicatie te batchen, heeft asynchrone verwerking een hogere doorvoer dan synchrone verwerking met slechts kleine codewijzigingen. Deze doorvoerwinst vereist geen asynchrone bibliotheken van derden. Gevorderde gebruikers kunnen hun applicaties verder optimaliseren met asynchrone programmeerpatronen en bibliotheken met ondersteuning voor asynchrone bewerkingen.

Om asynchrone verwerking te gebruiken, implementeer je een AsyncStatefulProcessor in plaats van de synchrone StatefulProcessor. De AsyncStatefulProcessor API spiegelt de synchrone StatefulProcessor API, dus de meeste applicaties vereisen slechts kleine wijzigingen om de asynchrone API te gebruiken. Zie Implementeer een AsyncStatefulProcessor.

Voor de synchrone transformWithState API en kernconcepten, zie Bouw een aangepaste stateful applicatie met transformWithState.

Implementeer een AsyncStatefulProcessor

Om een synchrone StatefulProcessor om te zetten naar een AsyncStatefulProcessor, breng je de volgende wijzigingen aan:

• Definieer de API-methoden (init, close, handleInputRows, handleExpiredTimer, en handleInitialState) met het trefwoord async def.
• Lees en werk status- en timerwaarden bij met await, of voer ze uit met de asyncio bibliotheek van Python. Dit geldt voor toestandsbewerkingen zoals valueState.get() en voor timerbewerkingen zoals registerTimer. Het creëren van toestandsobjecten, zoals handle.getValueState, blijft synchroon.

De volgende overwegingen gelden voor asynchrone verwerking:

• Als je applicatie data opslaat in lidvariabelen of in externe systemen, raadt Databricks aan om de logica te herschrijven zodat deze veilig is voor gelijktijdige uitvoering. Omdat handleInputRows en handleExpiredTimer gelijktijdig kunnen draaien over groepsleutels, mogen interleaved runs gedeelde data niet beschadigen. De meeste applicaties voldoen al aan deze vereiste.
• Databricks raadt aan om fouten uit toestandsoperaties niet te vangen of te onderdrukken. Apache Spark regelt deze fouten voor je. Als een toestandsoperatie faalt, markeert Apache Spark de taak als mislukt en probeert het deze opnieuw.
  • In een AsyncStatefulProcessor worden fouten bij statusbewerkingen voor je beheerd en komen ze nooit naar voren in je code.
  • In een synchrone StatefulProcessortoestand worden fouten bij statusbewerkingen in je code opgeworpen, maar het onderdrukken ervan kan de correctheid van de data in gevaar brengen.

Voorbeeld: tel rijen voor elke groeperingssleutel

Het volgende voorbeeld definieert een AsyncCountProcessor die het aantal rijen voor elke groepssleutel telt. De value_schema variabele definieert het schema van de ValueState die de lopende telling opslaat. In vergelijking met een synchrone StatefulProcessor, zijn de wijzigingen het sleutelwoord async def op elke methode en await bij de toestandslees- en updateoperaties. De oproep naar getValueState binnen init blijft synchroon. Definieer de processor als in de volgende code:

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

Voer een query uit met een asynchrone processor

Om een query met een asynchrone processor uit te voeren, geef je AsyncStatefulProcessor door aan transformWithState. De query gebruikt dezelfde syntaxis als het synchrone pad. De asynchrone en synchrone API's delen hetzelfde statusformaat, dus je kunt een bestaande query schakelen tussen een AsyncStatefulProcessor en een synchrone StatefulProcessor terwijl je hetzelfde checkpoint hergebruikt.

Voorbeeld: tel gebeurtenissen in de events voorbeelddataset

Het volgende voorbeeld draait AsyncCountProcessor op de events-voorbeelddataset. Elke record heeft een time veld (epochseconden) en een action veld met de waarde Open of Close. De query groepeert op action en telt de gebeurtenissen voor elk actietype. Voor meer voorbeelddatasets, zie Voorbeelddatasets.

De input_schema variabele definieert het schema van de bronrecords, en de output_schema variabele definieert het schema van de rijen die de processor uitzendt. Om de voorbeelddataset als een stroom te lezen, definieer je beide schema's en start je de query zoals in de volgende code:

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()

Na voltooiing van de query bekijk je de lopende telling voor elk actietype zoals in de volgende code:

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

Asynchrone toestands- en timeroperaties

In een AsyncStatefulProcessor, zijn bewerkingen op statusvariabelen en timers die waarden lezen of schrijven asynchroon. De meeste van deze bewerkingen geven één resultaat dat je ophaalt met await. Bewerkingen die een collectie teruggeven, geven in plaats daarvan een asynchrone iterator die je consumeert met async for. Voor een introductie tot async/await en asynchrone iterators in Python, zie de Python asyncio-documentatie.

De volgende tabel geeft bewerkingen weer die één enkel resultaat teruggeven dat u met await kunt ophalen:

De volgende tabel bevat bewerkingen die een asynchrone iterator retourneren die je met async for kunt ophalen:

Voorbeeld:async for

Om bijvoorbeeld de waarden in een AsyncListState te lezen, itereren we met async for zoals in de volgende code:

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

De methoden die toestandsobjecten creëren en toestandsvariabelen verwijderen blijven synchroon: getValueState, getMapState, getListState en deleteIfExists.

Voor een beschrijving van elk toestandstype, zie Aangepaste toestandtypes.

Optimaliseer met asynchrone programmeerpatronen

Asynchrone verwerking is nuttig wanneer je logica wacht op externe bewerkingen, zoals netwerkverzoeken. In plaats van op elk verzoek in volgorde te wachten, gebruik asyncio om de verzoeken gelijktijdig uit te voeren en de inactieve tijd te verminderen.

Voorbeeld: voer gelijktijdige verzoeken uit met asyncio.gather

Het volgende voorbeeld gebruikt asyncio.gather om alle HTTP-verzoeken per rij gelijktijdig af te vuren en te wachten tot ze voltooid zijn, waarna de maximale score in de state wordt opgeslagen. Definieer de processor als in de volgende code:

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()