Przetwarzanie asynchroniczne z transformWithState

Python transformWithState obsługuje asynchroniczne przetwarzanie oparte na asyncio. Dzięki jednoczesnemu wykonywaniu operacji stanu i logiki użytkownika równolegle dla kluczy grupowania oraz łączeniu komunikacji międzyprocesowej w partie przetwarzanie asynchroniczne osiąga wyższą przepustowość niż przetwarzanie synchroniczne, przy jedynie drobnych zmianach w kodzie. Ten wzrost przepustowości nie wymaga żadnych bibliotek asynchronicznych firm trzecich. Zaawansowani użytkownicy mogą dodatkowo optymalizować swoje aplikacje dzięki wzorcom programowania asynchronicznego i bibliotekom wspierającym asynchronię.

Przetwarzanie asynchroniczne obsługuje zarówno zapytania w trybie mikro-wsadowym, jak i zapytania w trybie czasu rzeczywistego. Dla pojedynczego klucza grupowania przetwarzanie asynchroniczne zachowuje kolejność wierszy. Różne klucze grupowania działają równolegle.

Aby użyć przetwarzania asynchronicznego, zaimplementuj AsyncStatefulProcessor zamiast synchronicznego StatefulProcessor. AsyncStatefulProcessor API odzwierciedla synchroniczne StatefulProcessor API, więc większość aplikacji wymaga jedynie drobnych zmian, aby korzystać z asynchronicznego API. Zobacz Wdrażanie AsyncStatefulProcessor.

Informacje o synchronicznego transformWithState API i podstawowych pojęciach znajdziesz w artykule Zbuduj niestandardową aplikację stanową z transformWithState.

Requirements

Przetwarzanie asynchroniczne spełnia następujące wymagania:

• Dostępne w Databricks Runtime 19 i nowszych lub w obliczeniach bezserwerowych ze środowiskiem w wersji 6 i nowszej.
• Dostępne tylko dla interfejsu API języka Python opartego transformWithState na wierszach. Przetwarzanie asynchroniczne nie jest obsługiwane dla transformWithStateInPandas ani dla interfejsu API Scala transformWithState.
• W potoku Lakeflow konfiguruj go tak, aby korzystał z kanału PREVIEW. Ten wymóg dotyczy zarówno klasycznych, jak i bezserwerowych mocy obliczeniowej potoków. Kanał CURRENT nie obsługuje przetwarzania asynchronicznego. Zobacz channel, sekcje konfiguracje potoków i kanały uruchomieniowe potoków Lakeflow.

Wdróż AsyncStatefulProcessor

Aby przekształcić synchroniczny StatefulProcessor na AsyncStatefulProcessor, wykonaj następujące zmiany:

• Zdefiniuj metody API (init, close, handleInputRows, handleExpiredTimer, oraz handleInitialState) za pomocą async def słowa kluczowego.
• Czytaj i aktualizuj wartości stanu i timera za pomocą await, albo uruchamiaj je za pomocą biblioteki Pythona asyncio. Dotyczy to operacji stanów, takich jak valueState.get(), oraz operacji timera, takich jak registerTimer. Tworzenie obiektów stanu, takich jak handle.getValueState, pozostaje synchroniczne.

Następujące kwestie dotyczą przetwarzania asynchronicznego:

• Jeśli Twoja aplikacja przechowuje dane w zmiennych członkowskich lub w zewnętrznych systemach, Databricks zaleca przepisanie logiki, aby była bezpieczna do współbieżnego wykonywania. Ponieważ handleInputRows i handleExpiredTimer mogą działać równocześnie na kluczach grupowania, przeplatane przebiegi nie mogą uszkadzać danych współdzielonych. Większość aplikacji już spełnia ten wymóg.
• Databricks zaleca, aby nie przechwytywać ani nie tłumić błędów w operacjach stanowych. Apache Spark radzi sobie z tymi błędami za Ciebie. Jeśli operacja stanu nie powiodła się, Apache Spark oznacza zadanie jako nieudane i ponawia je.
  • W operacji stanu AsyncStatefulProcessor błędy są obsługiwane automatycznie i nigdy nie są przekazywane do Twojego kodu.
  • W synchronicznym stanie StatefulProcessor błędy operacji pojawiają się w kodzie, ale ich tłumienie może zagrozić poprawności danych.

Przykład: liczenie wierszy dla każdego klucza grupowania

Poniższy przykład definiuje agregację AsyncCountProcessor, która liczy liczbę wierszy dla każdego klucza grupowania. Zmienna value_schema definiuje schemat elementu ValueState, który przechowuje bieżący licznik. W porównaniu do synchronicznego StatefulProcessor, zmiany to słowo kluczowe async def przy każdej metodzie oraz await w operacjach odczytu i aktualizacji stanu. Wywołanie getValueState w init pozostaje synchroniczne. Zdefiniuj procesor tak jak w poniższym kodzie:

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

Wykonaj zapytanie z procesorem asynchronicznym

Aby wykonać zapytanie z procesorem asynchronicznym, przekaż AsyncStatefulProcessor do transformWithState. Zapytanie korzysta z tej samej składni co ścieżka synchroniczna. API asynchroniczne i synchroniczne mają ten sam format stanu, więc możesz przełączać istniejące zapytanie między AsyncStatefulProcessor a synchronicznym StatefulProcessor, używając tego samego punktu kontrolnego.

Przykład: zliczenie zdarzeń w przykładowym zbiorze danych events

Poniższy przykład uruchamia AsyncCountProcessor na events przykładowym zbiorze danych. Każdy rekord ma pole time (sekundy epoki) oraz action pole o wartości Open lub Close. Zapytanie grupuje według action i liczy zdarzenia dla każdego typu akcji. Więcej przykładowych zbiorów danych można znaleźć w sekcji Przykładowe zbiory danych.

Zmienna input_schema definiuje schemat rekordów źródłowych, a zmienna output_schema schemat wierszy generowanych przez procesor. Aby odczytać przykładowy zbiór danych jako strumień, zdefiniuj oba schematy, a następnie rozpocznij zapytanie zgodnie z następującym kodem:

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 zakończeniu zapytania zobacz bieżącą liczbę dla każdego typu akcji w następującym kodzie:

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

Operacje dotyczące stanu asynchronicznego i czasomierza

W AsyncStatefulProcessor operacje na zmiennych stanu i liczniku czasu, które odczytują lub zapisują wartości, są asynchroniczne. Większość tych operacji zwraca jeden wynik, który pobieramy za pomocą await. Operacje zwracające kolekcję zamiast tego zwracają asynchroniczny iterator, którego używa się z async for. Aby poznać wprowadzenie do async/await oraz iteratorów asynchronicznych w Pythonie, zobacz dokumentację Python asyncio.

Poniższa tabela przedstawia operacje zwracające pojedynczy wynik, który można uzyskać za pomocą await:

Poniższa tabela przedstawia operacje zwracające asynchroniczny iterator, który można pobrać za pomocą async for:

Przykład: async for

Na przykład, aby odczytać wartości w elemencie AsyncListState, iteruj za pomocą async for, jak w następującym kodzie:

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

Metody tworzące obiekty stanów i usuwające zmienne stanów pozostają synchroniczne: getValueState, getMapState, getListState, oraz deleteIfExists.

Opis każdego typu stanu można znaleźć w artykule Niestandardowe typy stanów.

Optymalizuj za pomocą asynchronicznych wzorców programowania

Przetwarzanie asynchroniczne jest przydatne, gdy logika czeka na operacje zewnętrzne, takie jak żądania sieciowe. Zamiast czekać na każde żądanie po kolei, użyj asyncio, aby uruchamiać żądania jednocześnie i zmniejszyć czas bezczynności.

Przykład: wykonuj żądania współbieżne z asyncio.gather

Poniższy przykład wykorzystuje asyncio.gather do jednoczesnego uruchamiania wszystkich żądań HTTP dla każdego wiersza i czekania na ich zakończenie, a następnie zapisuje maksymalny wynik w stanie. Zdefiniuj procesor tak jak w poniższym kodzie:

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