Aller au contenu principal
Dernière mise à jour le

Exemple d'applications avec état

Cet article contient des exemples de code pour des applications avec état personnalisées. Databricks recommande d'utiliser des méthodes avec état intégrées pour les opérations courantes, telles que les agrégations et les jointures.

Les modèles de cet article utilisent l'opérateur transformWithState et les classes associées disponibles dans Databricks Runtime 16.2 et versions ultérieures. Consultez Créer une application avec état personnalisée.

remarque

Python prend en charge à la fois l'API transformWithState basée sur les lignes (disponible en mode micro-batch et en mode temps réel) et l'opérateur transformWithStateInPandas basé sur Pandas. Les exemples ci-dessous fournissent du code utilisant transformWithStateInPandas en Python et transformWithState en Scala.

Exigences​

L'opérateur transformWithState et les APIs et classes associées ont les exigences suivantes :

  • Disponible dans Databricks Runtime 16.2 et versions ultérieures.
  • Le mode d'accès standard est pris en charge pour Python (transformWithStateInPandas et basé sur les lignes transformWithState) dans Databricks Runtime 16.3 et versions supérieures, et pour Scala (transformWithState) dans Databricks Runtime 17.3 et versions supérieures.
  • RocksDB est le fournisseur de magasin d'état default dans Databricks Runtime 17.3 et les versions ultérieures. Pour les versions de Databricks Runtime antérieures à 17.3, vous devez configurer le fournisseur de magasin d'état RocksDB. Databricks recommande d'activer RocksDB dans le cadre de la configuration compute.
remarque

Sur les versions de Databricks Runtime inférieures à 17.3, activez le fournisseur de magasin d'état RocksDB pour la session actuelle en exécutant ce qui suit :

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

Dimension à évolution lente (SCD) de type 1​

Le code suivant est un exemple de mise en œuvre du SCD de type 1 à l'aide de transformWithState. Le SCD de type 1 ne suit que la valeur la plus récente pour un champ donné.

remarque

Vous pouvez utiliser des tables de streaming et AUTO CDC ... INTO pour implémenter le SCD de type 1 ou de type 2 en utilisant des tables basées sur Delta Lake. Cet exemple implémente le SCD de type 1 dans le magasin d'état, ce qui offre une latence plus faible pour les applications quasi-temps réel.

Python
# Import the necessary libraries
import pandas as pd
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, LongType, StringType
from typing import Iterator

# Set the state store provider to RocksDB
spark.conf.set("spark.sql.streaming.stateStore.providerClass", "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

# Define the output schema for the streaming query
output_schema = StructType([
    StructField("user", StringType(), True),
    StructField("time", LongType(), True),
    StructField("location", StringType(), True)
])

# Define a custom StatefulProcessor for slowly changing dimension type 1 (SCD1) operations
class SCDType1StatefulProcessor(StatefulProcessor):
    def init(self, handle: StatefulProcessorHandle) -> None:
        self.handle = handle
        # Define the schema for the state value
        value_state_schema = StructType([...])
            StructField("user", StringType(), True),
            StructField("time", LongType(), True),
            StructField("location", StringType(), True)
        ])
        # Initialize the state to store the latest location for each user
        self.latest_location = handle.getValueState("latestLocation", value_state_schema)

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        # Find the row with the maximum time value
        max_row = None
        max_time = float('-inf')
        for pdf in rows:
            for _, pd_row in pdf.iterrows():
                time_value = pd_row["time"]
                if time_value > max_time:
                    max_time = time_value
                    max_row = tuple(pd_row)

        # Check whether state exists and update if necessary
        exists = self.latest_location.exists()
        if not exists or max_row[1] > self.latest_location.get()[1]:
            # Update the state with the new max row
            self.latest_location.update(max_row)
            # Yield the updated row
            yield pd.DataFrame(
                {"user": (max_row[0],), "time": (max_row[1],), "location": (max_row[2],)}
            )
        # Yield an empty DataFrame if no update is needed
        yield pd.DataFrame()

    def close(self) -> None:
        # No cleanup needed
        pass

# Apply the stateful transformation to the input DataFrame
(df.groupBy("user")
  .transformWithStateInPandas(
      statefulProcessor=SCDType1StatefulProcessor(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
  )
  .writeStream...  # Continue with stream writing configuration
)

Dimension à évolution lente (SCD) de type 2​

Les notebooks suivants contiennent un exemple d'implémentation du SCD de type 2 à l'aide de transformWithState en Python ou Scala.

SCD de type 2 Python

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

SCD Type 2 Scala

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Détecteur de temps d'arrêt​

transformWithState implémente des minuteurs pour vous permettre d'agir en fonction du temps écoulé, même si aucun enregistrement pour une clé donnée n'est traité dans un micro-batch.

L'exemple suivant met en œuvre un modèle pour un détecteur de temps d'arrêt. Chaque fois qu'une nouvelle valeur est vue pour une clé donnée, elle met à jour la valeur d'état lastSeen, efface tous les minuteurs existants, et Reset un minuteur pour l'avenir.

Lorsqu'un temporisateur expire, l'application émet le temps écoulé depuis le dernier événement observé pour la clé. Il définit ensuite un nouveau minuteur pour émettre une mise à jour 10 secondes plus tard.

Python
import datetime
import time

class DownTimeDetectorStatefulProcessor(StatefulProcessor):
    def init(self, handle: StatefulProcessorHandle) -> None:
        # Define the schema for the state value (timestamp)
        state_schema = StructType([StructField("value", TimestampType(), True)])
        self.handle = handle
        # Initialize state to store the last seen timestamp for each key
        self.last_seen = handle.getValueState("last_seen", state_schema)

    def handleExpiredTimer(self, key, timerValues, expiredTimerInfo) -> Iterator[pd.DataFrame]:
        latest_from_existing = self.last_seen.get()
        # Calculate downtime duration
        downtime_duration = timerValues.getCurrentProcessingTimeInMs() - int(time.time() * 1000)
        # Register a new timer for 10 seconds in the future
        self.handle.registerTimer(timerValues.getCurrentProcessingTimeInMs() + 10000)
        # Yield a DataFrame with the key and downtime duration
        yield pd.DataFrame(
            {
                "id": key,
                "timeValues": str(downtime_duration),
            }
        )

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        # Find the row with the maximum timestamp
        max_row = max((tuple(pdf.iloc[0]) for pdf in rows), key=lambda row: row[1])

        # Get the latest timestamp from the existing state or use epoch start if a timestamp doesn't exist
        if self.last_seen.exists():
            latest_from_existing = self.last_seen.get()
        else:
            latest_from_existing = datetime.fromtimestamp(0)

        # If the new data is more recent than the existing state
        if latest_from_existing < max_row[1]:
            # Delete all existing timers
            for timer in self.handle.listTimers():
                self.handle.deleteTimer(timer)
            # Update the last seen timestamp
            self.last_seen.update((max_row[1],))

        # Register a new timer for 5 seconds in the future
        self.handle.registerTimer(timerValues.getCurrentProcessingTimeInMs() + 5000)

        # Get current processing time in milliseconds
        timestamp_in_millis = str(timerValues.getCurrentProcessingTimeInMs())

        # Yield a DataFrame with the key and current timestamp
        yield pd.DataFrame({"id": key, "timeValues": timestamp_in_millis})

    def close(self) -> None:
        # No cleanup needed
        pass

Migrer les informations d'état existantes​

L'exemple suivant montre comment implémenter une application avec état qui accepte un état initial. Vous pouvez ajouter une gestion de l'état initial à n'importe quelle application avec état, mais l'état initial ne peut être défini que lors de la première initialisation de l'application.

Cet exemple utilise l'API de lecture statestore pour charger les informations d'état existantes à partir d'un chemin de point de contrôle. Un cas d'utilisation exemple pour ce modèle est la migration d'applications avec état existantes vers transformWithState.

Python
# Import the necessary libraries
import pandas as pd
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, LongType, StringType, IntegerType
from typing import Iterator

# Set RocksDB as the state store provider for better performance
spark.conf.set("spark.sql.streaming.stateStore.providerClass", "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

"""
Input schema is as below

input_schema = StructType(
    [StructField("id", StringType(), True)],
    [StructField("value", StringType(), True)]
)
"""

# Define the output schema for the streaming query
output_schema = StructType([
    StructField("id", StringType(), True),
    StructField("accumulated", StringType(), True)
])

class AccumulatedCounterStatefulProcessorWithInitialState(StatefulProcessor):

    def init(self, handle: StatefulProcessorHandle) -> None:
        # Define the schema for the state value (integer)
        state_schema = StructType([StructField("value", IntegerType(), True)])
        # Initialize state to store the accumulated counter for each id
        self.counter_state = handle.getValueState("counter_state", state_schema)
        self.handle = handle

    def handleInputRows(self, key, rows, timerValues) -> Iterator[pd.DataFrame]:
        # Check if state exists for the current key
        exists = self.counter_state.exists()
        if exists:
            value_row = self.counter_state.get()
            existing_value = value_row[0]
        else:
            existing_value = 0

        accumulated_value = existing_value

        # Process input rows and accumulate values
        for pdf in rows:
            value = pdf["value"].astype(int).sum()
            accumulated_value += value

        # Update the state with the new accumulated value
        self.counter_state.update((accumulated_value,))

        # Yield a DataFrame with the key and accumulated value
        yield pd.DataFrame({"id": key, "accumulated": str(accumulated_value)})

    def handleInitialState(self, key, initialState, timerValues) -> None:
        # Initialize the state with the provided initial value
        init_val = initialState.at[0, "initVal"]
        self.counter_state.update((init_val,))

    def close(self) -> None:
        # No cleanup needed
        pass

# Load initial state from a checkpoint directory
initial_state = spark.read.format("statestore")
  .option("path", "$checkpointsDir")
  .load()

# Apply the stateful transformation to the input DataFrame
df.groupBy("id")
  .transformWithStateInPandas(
      statefulProcessor=AccumulatedCounterStatefulProcessorWithInitialState(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
      initialState=initial_state,
  )
  .writeStream...  # Continue with stream writing configuration

Migrer la table Delta vers un magasin d'états pour l'initialisation​

Les Notebooks suivants contiennent un exemple d'initialisation des valeurs de magasin d'état à partir d'une table Delta à l'aide de transformWithState en Python ou Scala.

Initialiser l'état à partir de Delta Python

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Initialiser l'état à partir de Delta Scala

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Suivi de session​

Les notebooks suivants contiennent un exemple de suivi de session utilisant transformWithState en Python ou Scala.

Suivi des sessions Python

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Suivi de session Scala

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Jointure personnalisée de Stream à Stream à l'aide de transformWithState​

Le code suivant démontre une jointure stream-stream personnalisée sur plusieurs streams à l'aide de transformWithState. Vous pouvez utiliser cette approche à la place d’un opérateur de jointure intégré pour les raisons suivantes :

  • Vous devez utiliser le mode de sortie de mise à jour qui ne prend pas en charge les jointures de Stream à Stream. Ceci est particulièrement utile pour les applications à faible latence.
  • Vous devez continuer à effectuer des jointures pour les lignes arrivant tardivement (après l'expiration du watermark).
  • Vous devez effectuer des jointures Stream-Stream de plusieurs à plusieurs.

Cet exemple donne à l'utilisateur un contrôle total sur la logique d'expiration de l'état, ce qui permet une extension dynamique de la période de rétention pour gérer les événements hors séquence même après le watermark.

Python
# Import the necessary libraries
import pandas as pd
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, StringType, TimestampType
from typing import Iterator

# Define output schema for the joined data
output_schema = StructType([
    StructField("user_id", StringType(), True),
    StructField("event_type", StringType(), True),
    StructField("timestamp", TimestampType(), True),
    StructField("profile_name", StringType(), True),
    StructField("email", StringType(), True),
    StructField("preferred_category", StringType(), True)
])

class CustomStreamJoinProcessor(StatefulProcessor):
    # Initialize stateful storage for user profiles, preferences, and event tracking.
    def init(self, handle: StatefulProcessorHandle) -> None:

        # Define schemas for different types of state data
        profile_schema = StructType([
            StructField("name", StringType(), True),
            StructField("email", StringType(), True),
            StructField("updated_at", TimestampType(), True)
        ])
        preferences_schema = StructType([
            StructField("preferred_category", StringType(), True),
            StructField("updated_at", TimestampType(), True)
        ])
        activity_schema = StructType([
            StructField("event_type", StringType(), True),
            StructField("timestamp", TimestampType(), True)
        ])

        map_state_key_schema = StructType([
            StructField("user_id", StringType(), True)
        ])

        # Initialize state storage for user profiles, preferences, and activity
        self.profile_state = handle.getMapState("user_profiles", map_state_key_schema, profile_schema)
        self.preferences_state = handle.getMapState("user_preferences", map_state_key_schema, preferences_schema)
        self.activity_state = handle.getMapState("user_activity", map_state_key_schema, activity_schema)

    # Process incoming events and update the state
    def handleInputRows(self, key, rows: Iterator[pd.DataFrame], timerValues) -> Iterator[pd.DataFrame]:
        df = pd.concat(rows, ignore_index=True)
        output_rows = []

        for _, row in df.iterrows():
            user_id = row["user_id"]

            if "event_type" in row:  # User activity event
                self.activity_state.updateValue(user_id, row.to_dict())
                # Set a timer to process this event after a 10-second delay
                self.handle.registerTimer(timerValues.get_current_processing_time_in_ms() + (10 * 1000))

            elif "name" in row:  # Profile update
                self.profile_state.updateValue(user_id, row.to_dict())

            elif "preferred_category" in row:  # Preference update
                self.preferences_state.updateValue(user_id, row.to_dict())

        # No immediate output; processing will happen when the timer expires
        return iter([])

    # Perform lookup after delay, handling out-of-order and late-arriving events.
    def handleExpiredTimer(self, key, timerValues, expiredTimerInfo) -> Iterator[pd.DataFrame]:

        # Retrieve stored state for the user
        user_activity = self.activity_state.getValue(key)
        user_profile = self.profile_state.getValue(key)
        user_preferences = self.preferences_state.getValue(key)

        if user_activity:
            # Combine data from different states into a single output row
            output_row = {
                "user_id": key,
                "event_type": user_activity["event_type"],
                "timestamp": user_activity["timestamp"],
                "profile_name": user_profile.get("name") if user_profile else None,
                "email": user_profile.get("email") if user_profile else None,
                "preferred_category": user_preferences.get("preferred_category") if user_preferences else None
            }
            return iter([pd.DataFrame([output_row])])

        return iter([])

    def close(self) -> None:
        # No cleanup needed
        pass

# Apply transformWithState to the input DataFrame
(df.groupBy("user_id")
  .transformWithStateInPandas(
      statefulProcessor=CustomStreamJoinProcessor(),
      outputStructType=output_schema,
      outputMode="Append",
      timeMode="ProcessingTime"
  )
  .writeStream...  # Continue with stream writing configuration
)

Calcul Top-K​

L'exemple suivant utilise un ListState avec une file d'attente prioritaire pour maintenir et mettre à jour les K éléments supérieurs dans un Stream pour chaque clé de groupe en temps réel.

Top-K Python

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks

Top-K Scala

Ouvrir le notebook dans un nouvel ongletOuvrir dans Databricks
Sur cette page