Asynchrone Verarbeitung mit transformWithState (Beta)

Asynchrone Verarbeitung ist ab Databricks Runtime 19 oder auf serverlosem Compute mit Umgebungsversion 6 oder höher verfügbar.

Python transformWithState unterstützt asynchrone Verarbeitung auf Basis von asyncio. Durch das gleichzeitige Ausführen von Zustandsoperationen und Benutzerlogik über Gruppierungsschlüssel hinweg und das Batching der Kommunikation zwischen Prozessen hat asynchrone Verarbeitung einen höheren Durchsatz als synchrone Verarbeitung mit nur geringfügigen Codeänderungen. Dieser Durchsatzgewinn erfordert keine asynchronen Drittanbieter-Bibliotheken. Fortgeschrittene Nutzer können ihre Anwendungen weiter mit asynchronen Programmiermustern und asynk-fähigen Bibliotheken optimieren.

Um asynchrone Verarbeitung zu verwenden, implementieren Sie AsyncStatefulProcessor anstelle des synchronen StatefulProcessor. Die AsyncStatefulProcessor API spiegelt die synchrone StatefulProcessor API wider, sodass die meisten Anwendungen nur geringe Änderungen benötigen, um die asynchrone API zu nutzen. Siehe Implementieren eines AsyncStatefulProcessor

Für die synchrone transformWithState API und Kernkonzepte siehe Erstellen einer benutzerdefinierten zustandsbehafteten Anwendung mit transformWithState.

Implementiere ein AsyncStatefulProcessor

Um eine synchrone StatefulProcessor in eine AsyncStatefulProcessor umzuwandeln, nehmen Sie folgende Änderungen vor:

• Definiere die API-Methoden (init, close, handleInputRows, handleExpiredTimer, und handleInitialState) mit dem async def Schlüsselwort.
• Lesen und aktualisieren Sie Zustands- und Timerwerte mit await, oder verwenden Sie sie mit der Python-Bibliothek asyncio. Dies gilt für Zustandsoperationen wie valueState.get() und für Timer-Operationen wie registerTimer. Das Erstellen von Zustandsobjekten, wie handle.getValueState, bleibt synchron.

Die folgenden Überlegungen gelten für asynchrone Verarbeitung:

• Wenn Ihre Anwendung Daten in Mitgliedsvariablen oder in externen Systemen speichert, empfiehlt Databricks, die Logik so umzuschreiben, dass sie für gleichzeitige Ausführung sicher ist. Da handleInputRows und handleExpiredTimer gleichzeitig über Gruppierungsschlüssel hinweg ausgeführt werden können, dürfen interleaved-Läufe die gemeinsamen Daten nicht beschädigen. Die meisten Anwendungen erfüllen diese Anforderung bereits.
• Databricks empfiehlt, Fehler bei Zustandsoperationen nicht abzufangen oder zu unterdrücken. Apache Spark behandelt diese Fehler für Sie. Wenn eine Zustandsoperation fehlschlägt, markiert Apache Spark die Aufgabe als fehlgeschlagen und führt sie erneut aus.
  • In einem AsyncStatefulProcessorstatusbehafteten Vorgang werden Betriebsfehler für Sie verwaltet und tauchen nie in Ihrem Code auf.
  • In einem synchronen StatefulProcessorZustand werden Fehler bei Statusvorgängen in Ihrem Code ausgelöst, aber deren Unterdrückung kann die Datenkorrektheit beeinträchtigen.

Beispiel: Zeilen für jeden Gruppierungsschlüssel zählen

Das folgende Beispiel definiert ein AsyncCountProcessor, der die Anzahl der Zeilen für jeden Gruppierungsschlüssel zählt. Die Variable value_schema definiert das Schema von der Tabelle ValueState, die die laufende Anzahl speichert. Im Vergleich zu einer synchronen StatefulProcessorMethode sind die Änderungen das async def Schlüsselwort bei jeder Methode sowie await bei den Zustandslese- und Aktualisierungsoperationen. Der Aufruf von getValueState in init bleibt synchron. Definieren Sie den Prozessor wie im folgenden 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

Führe eine Abfrage mit einem asynchronen Prozessor aus

Um eine Abfrage mit einem asynchronen Prozessor auszuführen, übergebe dein AsyncStatefulProcessor an transformWithState. Die Abfrage verwendet dieselbe Syntax wie der synchrone Pfad. Die asynchronen und synchronen APIs teilen dasselbe Zustandsformat, sodass man eine bestehende Abfrage zwischen einer AsyncStatefulProcessor und einer synchronen StatefulProcessor wechseln kann und dabei denselben Checkpoint wiederverwenden.

Beispiel: Zähle Ereignisse im events Beispieldatensatz

Das folgende Beispiel führt AsyncCountProcessor für den events-Beispieldatensatz aus. Jeder Datensatz hat ein Feld time (Epochensekunden) und ein action Feld mit dem Wert Open oder Close. Die Abfrage gruppiert nach action und zählt die Ereignisse für jeden Aktionstyp. Weitere Beispieldatensätze finden Sie unter Beispieldatensätze.

Die Variable input_schema definiert das Schema der Quelldatensätze, und die Variable output_schema definiert das Schema der Zeilen, die der Prozessor ausgibt. Um den Beispieldatensatz als Strom auszulesen, definieren Sie beide Schemata und starten dann die Abfrage wie im folgenden 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()

Nachdem die Abfrage abgeschlossen ist, sehen Sie sich die laufende Anzahl für jeden Aktionstyp wie im folgenden Code an:

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

Asynchrone Zustands- und Timeroperationen

In einem AsyncStatefulProcessor, sind Variablen- und Timeroperationen, die Werte lesen oder schreiben, asynchron. Die meisten dieser Operationen liefern ein einziges Ergebnis, das Sie mit awaitabrufen. Operationen, die eine Sammlung zurückgeben, geben stattdessen einen asynchronen Iterator zurück, den Sie mit async forkonsumieren. Eine Einführung in async/await und asynchrone Iteratoren in Python finden Sie in der Python asyncio-Dokumentation.

Die folgende Tabelle listet Operationen auf, die ein einzelnes Ergebnis zurückgeben, das Sie mit awaitabrufen können:

Die folgende Tabelle listet Operationen auf, die einen asynchronen Iterator zurückgeben, den Sie mit async for abrufen können:

Beispiel: async for

Zum Beispiel, um die Werte in einem AsyncListState zu lesen, iterieren Sie mit async for wie im folgenden Code:

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

Die Methoden, die Zustandsobjekte erzeugen und Zustandsvariablen löschen, bleiben synchron: getValueState, getMapState, getListState und deleteIfExists.

Für eine Beschreibung jedes Zustandstyps siehe Benutzerdefinierte Zustandstypen.

Optimieren Sie mit asynchronen Programmierungsmustern

Asynchrone Verarbeitung ist nützlich, wenn Ihre Logik auf externe Operationen wie Netzwerkanfragen wartet. Anstatt auf jede Anfrage nacheinander zu warten, sollten Sie asyncio verwenden, um die Anfragen gleichzeitig auszuführen und die Leerlaufzeit zu reduzieren.

Beispiel: gleichzeitige Anfragen ausführen mit asyncio.gather

Das folgende Beispiel verwendet asyncio.gather , um alle HTTP-Anfragen pro Zeile gleichzeitig auszulösen und auf deren Abschluss zu warten, dann speichert es den maximalen Score im Zustand. Definieren Sie den Prozessor wie im folgenden 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()