Creare un'applicazione personalizzata con stato

È possibile costruire applicazioni di streaming usando operatori personalizzati con stato per implementare soluzioni a bassa latenza e quasi in tempo reale che utilizzano logica arbitraria stato-dipendente. Gli operatori con stato personalizzati sbloccano nuovi casi d'uso operativi e modelli non disponibili tramite l'elaborazione tradizionale di Structured Streaming.

Requisiti

L'operatore transformWithState e le API e le classi correlate hanno i requisiti seguenti:

• Disponibile in Databricks Runtime 16.2 e versioni successive.
• La modalità di accesso standard è supportata per Python (transformWithStateInPandas e per righe transformWithState) in Databricks Runtime 16.3 e versioni successive e per Scala (transformWithState) in Databricks Runtime 17.3 e versioni successive.
• È necessario usare il provider dell'archivio stati RocksDB. Databricks consiglia di abilitare RocksDB come parte della configurazione di calcolo.

spark.conf.set("spark.sql.streaming.stateStore.providerClass", "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

Che cos'è transformWithState?

L'operatore transformWithState applica un processore con stato personalizzato a una query Structured Streaming. È necessario implementare un processore con stato personalizzato per usare transformWithState. Structured Streaming include API per costruire il tuo processore con stato usando Python, Scala o Java.

Si usa transformWithState per applicare la logica personalizzata a una chiave di raggruppamento per i record elaborati in modo incrementale con Structured Streaming. Di seguito viene descritta la progettazione generale:

• Definire una o più variabili di stato.
• Le informazioni sullo stato vengono rese persistenti per ogni chiave di raggruppamento e possono essere accessibili per ogni variabile di stato in base alla logica definita dall'utente.
• Per ogni micro batch elaborato, tutti i record per la chiave sono disponibili come iteratore.
• Utilizzare maniglie integrate per controllare quando e come i record vengono emessi, in base a timer e condizioni definite dall'utente.
• I valori di stato supportano singole definizioni TTL (Time-to-Live), consentendo la flessibilità nella gestione della scadenza dello stato e delle dimensioni dello stato.

Poiché transformWithState supporta l'evoluzione dello schema nell'archivio stati, è possibile eseguire l'iterazione e aggiornare le applicazioni di produzione senza perdere informazioni sullo stato cronologico o dover rielaborare i record, offrendo flessibilità per lo sviluppo e la facilità di manutenzione. Vedere Evoluzione dello schema nell'archivio stati.

Capacità di elaborazione integrate

Implementate la logica di base per la vostra applicazione con stato personalizzata implementando i gestori di utilizzando gli handle predefiniti .

• Gli handle forniscono i metodi per interagire con i valori di stato e i timer, elaborare i record in ingresso e generare record.
• I gestori definiscono la logica personalizzata basata su eventi.

Gli handle per ogni tipo di stato vengono implementati in base alla struttura di dati sottostante, ma ognuno contiene funzionalità per ottenere, inserire, aggiornare ed eliminare record.

I gestori vengono implementati in base a entrambi gli eventi osservati nei record di input o nei timer, usando la semantica seguente:

• Definire un gestore usando il metodo handleInputRows per controllare la modalità di elaborazione dei dati, lo stato viene aggiornato e i record vengono generati per ogni micro batch di record elaborati per la chiave di raggruppamento. Vedere Gestire le righe di input.
• Definire un gestore usando il metodo handleExpiredTimer per usare le soglie basate sul tempo per eseguire la logica indipendentemente dal fatto che vengano elaborati record aggiuntivi per la chiave di raggruppamento. Vedere Eventi programmati.

La tabella seguente include un confronto tra i comportamenti funzionali supportati da questi gestori:

È possibile combinare handleInputRows e handleExpiredTimer per implementare logica complessa in base alle esigenze.

Ad esempio, è possibile implementare un'applicazione che usa handleInputRows per aggiornare i valori di stato per ogni micro batch e impostare un timer di 10 secondi in futuro. Se non vengono elaborati record aggiuntivi, è possibile usare handleExpiredTimer per generare i valori correnti nell'archivio dello stato. Se vengono elaborati nuovi record per la chiave di raggruppamento, è possibile cancellare il timer esistente e impostare un nuovo timer.

Tipi di stato personalizzati

È possibile implementare più oggetti di stato in un singolo operatore con stato. I nomi che si assegnano a ogni oggetto di stato vengono mantenuti nell'archivio stati, a cui è possibile accedere con il lettore dell'archivio stati. Se l'oggetto di stato utilizza un StructType, fornisci i nomi per ogni campo nella struttura mentre passi lo schema. Questi nomi sono visibili anche durante la lettura dell'archivio degli stati. Consulta le informazioni sullo stato dello Streaming Strutturato.

La funzionalità fornita da classi e operatori predefiniti è destinata a offrire flessibilità ed estendibilità e le scelte di implementazione devono essere informate dalla logica completa necessaria per l'esecuzione dell'applicazione. Ad esempio, è possibile implementare una logica quasi identica usando un ValueState raggruppato per campi user_id e session_id o un MapState raggruppato per user_id dove session_id è la chiave per l'MapState. In questo caso, un MapState potrebbe essere l'implementazione preferita se la logica deve valutare le condizioni tra più session_ids.

Nelle sezioni seguenti vengono descritti i tipi di stato supportati da transformWithState.

ValueState

Per ogni chiave di raggruppamento è presente un valore associato.

Uno stato valore può includere tipi complessi, ad esempio un struct o una tupla. Quando si aggiorna un ValueState, si implementa la logica per sostituire l'intero valore. Il TTL per uno stato di valore viene reimpostato quando il valore viene aggiornato, ma non si reimposta se una chiave di origine corrispondente a un ValueState viene elaborata senza aggiornare un ValueStatearchiviato.

ListState

Per ogni chiave di raggruppamento è presente un elenco associato.

Uno stato elenco è una raccolta di valori, ognuno dei quali può includere tipi complessi. Ogni valore di un elenco ha una durata (TTL) specifica. È possibile aggiungere elementi a un elenco aggiungendo singoli elementi, aggiungendo un elenco di elementi o sovrascrivendo l'intero elenco con un put. Solo l'operazione put viene considerata un aggiornamento per la reimpostazione del tempo di vita (TTL).

MapState

Per ogni chiave di raggruppamento è presente una mappa associata. Le mappe sono l'equivalente funzionale di Apache Spark a un dict Python.

Uno stato della mappa è una raccolta di chiavi distinte mappate a un valore che può includere tipi complessi. Ogni coppia chiave-valore in una mappa ha una durata TTL specifica. È possibile aggiornare il valore di una chiave specifica o rimuovere una chiave e il relativo valore. È possibile restituire un singolo valore usando la chiave, elencare tutte le chiavi, elencare tutti i valori o restituire un iteratore per lavorare con il set completo di coppie chiave-valore nella mappa.

Inizializzare una variabile di stato personalizzata

Quando si inizializza il StatefulProcessor, si crea una variabile locale per ogni oggetto stato che consente di interagire con gli oggetti di stato nella logica personalizzata. Le variabili di stato vengono definite e inizializzate eseguendo l'override del metodo init predefinito nella classe StatefulProcessor.

Si definisce una quantità arbitraria di oggetti di stato usando i metodi getValueState, getListStatee getMapState durante l'inizializzazione dell'StatefulProcessor.

Ogni oggetto di stato deve avere quanto segue:

• Nome univoco
• Un schema specificato
  • In Python lo schema viene specificato in modo esplicito.
  • In Scala passare un Encoder per specificare lo schema dello stato.

È anche possibile fornire una durata facoltativa (TTL) in millisecondi. Se si implementa uno stato della mappa, è necessario fornire una definizione di schema separata per le chiavi della mappa e i valori.

Esempio di applicazione con stato

Di seguito viene illustrata la sintassi di base per la definizione e l'uso di un processore con stato personalizzato con transformWithState, incluse le variabili di stato di esempio per ogni tipo supportato. Per altri esempi, vedere Esempio di applicazioni con stato.

current_value_tuple = value_state.get() # Returns the value state as a tuple
current_value = current_value_tuple[0]  # Extracts the first item in the tuple
new_value = current_value + 1           # Calculate a new value
value_state.update((new_value,))        # Pass the new value formatted as a tuple

Pitone

import pandas as pd
from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from typing import Iterator

spark.conf.set("spark.sql.streaming.stateStore.providerClass","org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

output_schema = StructType(
    [
        StructField("id", StringType(), True),
        StructField("countAsString", StringType(), True),
    ]
)

class SimpleCounterProcessor(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    value_state_schema = StructType([StructField("count", IntegerType(), True)])
    list_state_schema = StructType([StructField("count", IntegerType(), True)])
    self.value_state = handle.getValueState(stateName="valueState", schema=value_state_schema)
    self.list_state = handle.getListState(stateName="listState", schema=list_state_schema)
    # Schema can also be defined using strings and SQL DDL syntax
    self.map_state = handle.getMapState(stateName="mapState", userKeySchema="name string", valueSchema="count int")

  def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
    count = 0
    for pdf in rows:
      list_state_rows = [(120,), (20,)] # A list of tuples
      self.list_state.put(list_state_rows)
      self.list_state.appendValue((111,))
      self.list_state.appendList(list_state_rows)
      pdf_count = pdf.count()
      count += pdf_count.get("value")
    self.value_state.update((count,)) # Count is passed as a tuple
    iter = self.list_state.get()
    list_state_value = next(iter1)[0]
    value = count
    user_key = ("user_key",)
    if self.map_state.exists():
      if self.map_state.containsKey(user_key):
        value += self.map_state.getValue(user_key)[0]
    self.map_state.updateValue(user_key, (value,)) # Value is a tuple
    yield pd.DataFrame({"id": key, "countAsString": str(count)})

q = (df.groupBy("key")
  .transformWithStateInPandas(
    statefulProcessor=SimpleCounterProcessor(),
    outputStructType=output_schema,
    outputMode="Update",
    timeMode="None",
  )
  .writeStream...
)

Linguaggio di programmazione Scala

import org.apache.spark.sql.streaming._
import org.apache.spark.sql.{Dataset, Encoder, Encoders , DataFrame}
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

spark.conf.set("spark.sql.streaming.stateStore.providerClass","org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

class SimpleCounterProcessor extends StatefulProcessor[String, (String, String), (String, String)] {
  @transient private var countState: ValueState[Int] = _
  @transient private var listState: ListState[Int] = _
  @transient private var mapState: MapState[String, Int] = _

  private val longEncoder = Encoders.scalaLong
  private val intEncoder = Encoders.scalaInt
  private val stringEncoder = Encoders.STRING

  override def init(
      outputMode: OutputMode,
      timeMode: TimeMode): Unit = {
    countState = getHandle.getValueState[Int]("countState",
      intEncoder, TTLConfig.NONE)
    listState = getHandle.getListState[Int]("listState",
      intEncoder, TTLConfig.NONE)
    mapState = getHandle.getMapState[String, Int]("mapState",
      stringEncoder, intEncoder, TTLConfig.NONE)
  }

  override def handleInputRows(
      key: String,
      inputRows: Iterator[(String, String)],
      timerValues: TimerValues): Iterator[(String, String)] = {
    var count = countState.getOption().getOrElse(0)
    for (row <- inputRows) {
      val listData = Array(120, 20)
      listState.put(listData)
      listState.appendValue(count)
      listState.appendList(listData)
      count += 1
    }
    val iter = listState.get()
    var listStateValue = 0
    if (iter.hasNext) {
      listStateValue = iter.next()
    }
    countState.update(count)
    var value = count
    val userKey = "userKey"
    if (mapState.exists()) {
      if (mapState.containsKey(userKey)) {
        value += mapState.getValue(userKey)
      }
    }
    mapState.updateValue(userKey, value)
    Iterator((key, count.toString))
  }
}

val q = spark
        .readStream
        .format("delta")
        .load("$srcDeltaTableDir")
        .as[(String, String)]
        .groupByKey(x => x._1)
        .transformWithState(
            new SimpleCounterProcessor(),
            TimeMode.None(),
            OutputMode.Update(),
        )
        .writeStream...

StatefulProcessorHandle

PySpark include la classe StatefulProcessorHandle per fornire l'accesso alle funzioni che controllano il modo in cui il codice Python definito dall'utente interagisce con le informazioni sullo stato. È sempre necessario importare e passare il StatefulProcessorHandle alla variabile handle durante l'inizializzazione di un StatefulProcessor.

La variabile handle collega la variabile locale nella classe Python alla variabile di stato.

Specificare lo stato iniziale

Facoltativamente, è possibile fornire uno stato iniziale da usare con il primo micro-batch. Ciò può essere utile quando si esegue la migrazione di un flusso di lavoro esistente a una nuova applicazione personalizzata, si aggiorna un operatore con stato per modificare lo schema o la logica oppure si ripristina un errore che non può essere riparato automaticamente e richiede un intervento manuale.

Si specifica uno stato iniziale usando un dataframe con lo stesso schema di chiave di raggruppamento delle righe di input.

Usare le variabili di stato

Gli oggetti di stato supportati forniscono metodi per ottenere lo stato, aggiornare le informazioni sullo stato esistenti o cancellare lo stato corrente. Ogni tipo di stato supportato ha un'implementazione univoca di metodi che corrispondono alla struttura dei dati implementata.

Ogni chiave di raggruppamento osservata ha informazioni sullo stato dedicate.

• I record vengono generati in base alla logica implementata e usando lo schema di output specificato. Visualizza Emmettere record.
• È possibile accedere ai valori nell'archivio stati usando il lettore statestore. Questo lettore ha funzionalità batch e non è destinato ai carichi di lavoro a bassa latenza. Consulta le informazioni sullo stato dello Streaming Strutturato.
• La logica specificata tramite handleInputRows viene attivata solo se i record per la chiave sono presenti in un micro batch. Vedere Gestire le righe di input.
• Usare handleExpiredTimer per implementare la logica basata sul tempo senza dipendere dall'osservazione dei record per attivare. Vedere Eventi programmati.

Considerazioni avanzate per l'uso dello stato

La scrittura in una variabile di stato attiva una scrittura in RocksDB. Per ottimizzare le prestazioni, Databricks consiglia di elaborare tutti i valori nell'iteratore per una determinata chiave e di eseguire il commit degli aggiornamenti in un'unica scrittura quando possibile.

I valori di stato non hanno impostazioni predefinite. Se la logica richiede la lettura delle informazioni sullo stato esistenti, usare il metodo exists durante l'implementazione della logica.

Emettere record

La logica definita dall'utente controlla il modo in cui transformWithState genera record. I record vengono emessi per ogni chiave di raggruppamento.

Le applicazioni con stato personalizzato non presupmettono il modo in cui le informazioni sullo stato vengono usate per determinare come generare record e il numero restituito di record per una determinata condizione può essere nessuno, uno o molti.

Si applica la logica per generare record usando handleInputRows o handleExpiredTimer. Consultare Gestire le righe di input e programmare gli eventi temporizzati .

Pitone

In Python si definisce lo schema di output usando la parola chiave outputStructType chiamando transformWithStateInPandas.

Emetti i record usando un oggetto DataFrame pandas e yield.

Facoltativamente, è possibile yield un dataframe vuoto. Quando combinato con la modalità di output update, l'emissione di un DataFrame vuoto porta i valori della chiave di raggruppamento a null.

Linguaggio di programmazione Scala

In Scala, si emettono record usando un oggetto Iterator. Lo schema dell'output è derivato dai record generati.

È possibile, se si desidera, generare un Iteratorvuoto. Quando combinato con la modalità di output update, l'emissione di un Iterator vuoto aggiorna i valori per la chiave di raggruppamento portando la chiave di raggruppamento a essere null.

Gestire le righe di input

Usare il metodo handleInputRows per definire la logica per il modo in cui i record osservati nella query di streaming interagiscono con e aggiornano i valori dello stato. Il gestore definito con il metodo handleInputRows viene eseguito ogni volta che qualsiasi record viene elaborato tramite una query di Structured Streaming.

Per la maggior parte delle applicazioni con stato implementate con transformWithState, la logica di base viene definita usando handleInputRows.

Per ogni aggiornamento micro batch elaborato, tutti i record nel micro batch per una determinata chiave di raggruppamento sono disponibili usando un iteratore. La logica definita dall'utente può interagire con tutti i record del microbatch e dei valori correnti nel statestore.

eventi programmati

È possibile usare i timer per implementare la logica personalizzata in base al tempo trascorso da una condizione specificata.

È possibile usare i timer implementando un metodo handleExpiredTimer.

All'interno di una chiave di raggruppamento, i timer vengono identificati in modo univoco dal timestamp.

Quando un timer scade, il risultato viene determinato dalla logica implementata nell'applicazione. I modelli comuni includono:

• Emissione di informazioni archiviate in una variabile di stato.
• Rimozione delle informazioni sullo stato archiviate.
• Creazione di un nuovo timer.

I timer scaduti vengono attivati anche se non viene elaborato alcun record per la chiave associata in un micro batch.

Specificare il modello temporale

Quando si passa il StatefulProcessor a transformWithState, è necessario specificare il modello temporale. Sono supportate le opzioni seguenti:

• ProcessingTime
• EventTime
• NoTime o TimeMode.None()

Se si specifica NoTime, i timer non sono supportati per il processore.

Valori timer predefiniti

Databricks sconsiglia di richiamare l'orologio di sistema nella tua applicazione personalizzata con stato, poiché ciò può portare a tentativi di ripetizione inaffidabili in caso di fallimento del task. Usare i metodi nella classe TimerValues quando è necessario accedere al tempo di elaborazione o alla filigrana:

Che cos'è il tempo di durata (TTL)?

I valori di stato usati da transformWithState supportano una specifica TTL (Time to Live) facoltativa. Quando la durata (TTL) scade, il valore viene rimosso dall'archivio dello stato. TTL interagisce solo con i valori nell'archivio stati, ovvero è possibile implementare la logica per rimuovere le informazioni sullo stato, ma non è possibile attivare direttamente la logica perché TTL rimuove i valori di stato.

La durata (TTL) viene applicata per ciascun valore di stato, seguendo regole diverse per ogni tipo di stato.

• Le variabili di stato hanno come ambito il raggruppamento delle chiavi.
• Per gli oggetti ValueState, viene archiviato solo un singolo valore per ogni chiave di raggruppamento. Il valore di TTL si applica a questo parametro.
• Per gli oggetti ListState, l'elenco può contenere molti valori. La durata (TTL) si applica a ogni valore in un elenco in modo indipendente.
• Per gli oggetti MapState, ogni chiave della mappa ha un valore di stato associato. La durata (TTL, tempo di vita) si applica in modo indipendente a ciascuna coppia chiave-valore di una mappa.

Per tutti i tipi di stato, il tempo di vita (TTL) viene reimpostato se le informazioni sullo stato vengono aggiornate.

Qual è la differenza tra timer e TTL?

Esistono alcune sovrapposizioni tra timer e durata (TTL) per le variabili di stato, ma i timer forniscono un set più ampio di funzionalità rispetto alla durata (TTL).

TTL rimuove le informazioni sullo stato che non sono state aggiornate per il periodo specificato dall'utente. In questo modo gli utenti possono impedire l'aumento dello stato non controllato e rimuovere voci di stato non aggiornate. Poiché le mappe e gli elenchi implementano TTL per ogni valore, è possibile implementare funzioni che considerano solo i valori di stato aggiornati di recente impostando TTL.

I timer consentono di definire logica personalizzata oltre la rimozione dello stato, inclusa la creazione di record. Facoltativamente, è possibile usare timer per cancellare le informazioni sullo stato per un determinato valore di stato, con la flessibilità aggiuntiva per generare valori o attivare altre logiche condizionali in base al timer.