Elaborazione asincrona con transformWithState

L'elaborazione asincrona per l'API Python basata transformWithState su righe richiede Databricks runtime 19 e superiori, oppure computazione serverless con l'ambiente versione 6 e superiore.

Per utilizzare l'elaborazione asincrona in una pipeline Lakeflow, configura la pipeline per utilizzare il canale PREVIEW. Vedere channel in Configurazioni della pipeline.

Python transformWithState supporta l'elaborazione asincrona basata su asyncio. Eseguendo operazioni di stato e logica utente contemporaneamente tra chiavi di raggruppamento e con la comunicazione inter-processo in batch, l'elaborazione asincrona ha un throughput superiore rispetto all'elaborazione sincrona con solo modifiche minori al codice. Questo guadagno di throughput non richiede librerie asincrone di terze parti. Gli utenti avanzati possono ottimizzare ulteriormente le proprie applicazioni con schemi di programmazione asincroni e librerie che supportano la programmazione asincrona.

Per usare l'elaborazione asincrona, implementare un AsyncStatefulProcessor invece del sincrono StatefulProcessor. L'API AsyncStatefulProcessor rispecchia l'API sincrona StatefulProcessor , quindi la maggior parte delle applicazioni richiede solo piccole modifiche per utilizzare l'API asincrona. Vedi Implementazione di un AsyncStatefulProcessor.

Per l'API sincrona transformWithState e i concetti fondamentali, vedi Creare un'applicazione personalizzata con stato con transformWithState.

Implementare un AsyncStatefulProcessor

Per convertire un elemento sincrono StatefulProcessor in un AsyncStatefulProcessor, apporta le seguenti modifiche:

• Definisci i metodi API (init, close, handleInputRows, handleExpiredTimer, e handleInitialState) con la async def parola chiave.
• Leggi e aggiorna i valori di stato e del timer con await, oppure esegui queste operazioni usando la libreria asyncio di Python. Questo si applica alle operazioni di stato come valueState.get() e alle operazioni di timer come registerTimer. La creazione di oggetti di stato, come handle.getValueState, rimane sincrona.

Le seguenti considerazioni si applicano all'elaborazione asincrona:

• Se la tua applicazione memorizza i dati in variabili membri o in sistemi esterni, Databricks raccomanda di riscrivere la logica per essere sicura per l'esecuzione concorrente. Poiché handleInputRows e handleExpiredTimer possono essere eseguiti contemporaneamente tra chiavi di raggruppamento, le esecuzioni intercalate non devono corrompere i dati condivisi. La maggior parte delle applicazioni già soddisfa questo requisito.
• Databricks raccomanda di non intercettare o sopprimere errori derivanti dalle operazioni di stato. Apache Spark gestisce questi errori per te. Se un'operazione di stato fallisce, Apache Spark contrassegna il task come non riuscito e lo ritenta.
  • In un'operazione AsyncStatefulProcessor, gli errori di operazione di stato vengono gestiti per te e non vengono mai esposti al tuo codice.
  • In un'operazione di stato sincrona StatefulProcessor, gli errori di operazione di stato sono generati nel tuo codice, ma sopprimerli può compromettere la correttezza dei dati.

Esempio: contare le righe per ogni chiave di raggruppamento

Il seguente esempio definisce un AsyncCountProcessor che conta il numero di righe per ogni chiave di raggruppamento. La value_schema variabile definisce lo schema del ValueState che memorizza il conteggio corrente. Rispetto a un StatefulProcessor sincrono, i cambiamenti sono la parola chiave async def in ciascun metodo e await nelle operazioni di lettura e aggiornamento dello stato. La chiamata a getValueState in init rimane sincrona. Definisci il processore come nel seguente codice:

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

Esegui una query con un processore asincrono

Per eseguire una query con un processore asincrono, passa il tuo AsyncStatefulProcessor a transformWithState. La query utilizza la stessa sintassi del percorso sincrono. Le API asincrone e sincrone condividono lo stesso formato di stato, quindi puoi far passare una query esistente da un'AsyncStatefulProcessor a un'StatefulProcessor sincrona riutilizzando lo stesso checkpoint.

Esempio: contare gli eventi nel events dataset campione

L'esempio seguente esegue AsyncCountProcessor sul dataset campione events. Ogni record ha un time campo (secondi di epoca) e un action campo con il valore Open o Close. La query raggruppa per action e conta gli eventi per ogni tipo di azione. Per altri set di dati di esempio, consulta Set di dati di esempio.

La input_schema variabile definisce lo schema dei record sorgente, e la output_schema variabile definisce lo schema delle righe emesse dal processore. Per leggere il dataset di esempio come un flusso, definisci entrambi gli schemi, poi avvia la query come nel seguente codice:

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

Dopo il completamento della query, visualizza il conteggio corrente per ogni tipo di azione come nel seguente codice:

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

Operazioni di stato asincrono e timer

In AsyncStatefulProcessor, le operazioni di variabile di stato e timer che leggono o scrivono valori sono asincrone. La maggior parte di queste operazioni restituisce un unico risultato che si recupera con await. Le operazioni che restituiscono una collezione restituiscono invece un iteratore asincrono da usare con async for. Per un'introduzione a async/await e agli iteratori asincroni in Python, consulta la documentazione di Python asyncio.

La tabella seguente elenca le operazioni che restituiscono un singolo risultato che puoi recuperare con await:

La tabella seguente elenca le operazioni che restituiscono un iteratore asincrono che puoi recuperare con async for:

Esempio: async for

Ad esempio, per leggere i valori in un AsyncListState, iterare con async for come nel seguente codice:

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

I metodi che creano gli oggetti di stato ed eliminano variabili di stato rimangono sincroni: getValueState, getMapState, getListState, e deleteIfExists.

Per una descrizione di ogni tipo di stato, vedi Tipi di stato personalizzati.

Ottimizza con schemi di programmazione asincroni

L'elaborazione asincrona è utile quando la tua logica attende operazioni esterne, come le richieste di rete. Invece di aspettare ogni richiesta in sequenza, usa asyncio per eseguire le richieste contemporaneamente e ridurre i tempi di inattività.

Esempio: eseguire richieste concorrenti con asyncio.gather

L'esempio seguente usa asyncio.gather per lanciare tutte le richieste HTTP per riga contemporaneamente e ad attendere che vengano completate, quindi memorizza il punteggio massimo nello stato. Definisci il processore come nel seguente codice:

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