Pular para o conteúdo principal
Última atualização em

Exemplos de aplicativos com estado

Esta página contém exemplos de código para aplicações de transmissão com estado personalizadas usando o operador transformWithState. A Databricks recomenda o uso de métodos com estado integrados para operações comuns, como agregações e joins.

Consulte Criar um aplicativo com estado personalizado com transformWithState.

nota

O Python suporta tanto a API baseada em linhas transformWithState (disponível no modo microbatch e no modo em tempo real) quanto o operador transformWithStateInPandas baseado em Pandas. Os exemplos abaixo fornecem código usando transformWithStateInPandas em Python e transformWithState em Scala.

nota

Os exemplos executáveis nesta página criam tabelas em um esquema main.stateful_examples dedicado para que possam ser executados sem afetar seus dados existentes. Se você não tiver permissão para criar esquemas no catálogo main, altere o catálogo e o esquema nos exemplos para um local onde você possa criar tabelas.

Requisitos​

O operador transformWithState e as APIs e classes relacionadas têm os seguintes requisitos:

  • Disponível em Databricks Runtime 16.2 e acima.
  • O modo de acesso padrão é suportado para Python (transformWithStateInPandas e baseado em linha transformWithState) no Databricks Runtime 16.3 e superior, e para Scala (transformWithState) no Databricks Runtime 17.3 e superior.
  • RocksDB é o provedor default de armazenamento do estado no Databricks Runtime 17.3 e acima. Para versões Databricks Runtime abaixo de 17.3, você deve configurar o provedor de armazenamento do estado RocksDB . Databricks recomenda habilitar RocksDB como parte da configuração compute .
nota

Em versões Databricks Runtime anteriores à 17.3, habilite o provedor de armazenamento de estado RocksDB para a sessão atual executando o seguinte comando:

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

dimensões que mudam lentamente (SCD) (SCD) tipo 1​

O código a seguir é um exemplo de implementação do SCD tipo 1 usando transformWithState. O SCD tipo 1 rastreia apenas o valor mais recente de um determinado campo.

nota

Você pode usar tabelas de transmissão e AUTO CDC ... INTO para implementar SCD tipo 1 ou tipo 2 usando tabelas com suporte Delta Lake. Este exemplo implementa SCD tipo 1 no armazenamento do estado, o que proporciona menor latência para aplicações reais próximas do tempo de execução.

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

import uuid

# Create a dedicated schema for the example tables
spark.sql("CREATE SCHEMA IF NOT EXISTS main.stateful_examples")

# Seed a small Delta table to use as the streaming source
spark.sql("DROP TABLE IF EXISTS main.stateful_examples.scd1_source")
spark.createDataFrame(
    [("u1", 1, "NYC"), ("u1", 3, "SF"), ("u1", 2, "LA"), ("u2", 5, "London")],
    "user string, time long, location string",
).write.saveAsTable("main.stateful_examples.scd1_source")

df = spark.readStream.table("main.stateful_examples.scd1_source")

# Apply the stateful transformation to the input DataFrame
q = (
    df.groupBy("user")
    .transformWithStateInPandas(
        statefulProcessor=SCDType1StatefulProcessor(),
        outputStructType=output_schema,
        outputMode="Update",
        timeMode="None",
    )
    .writeStream.format("memory")
    .queryName("scd1_output")
    .option("checkpointLocation", f"/tmp/checkpoint_{uuid.uuid4()}")
    .trigger(availableNow=True)
    .start()
)

q.awaitTermination()

# Each user keeps only its latest location by time: u1 -> SF (time 3), u2 -> London (time 5)
display(spark.sql("SELECT user, time, location FROM scd1_output ORDER BY user"))

dimensões que mudam lentamente (SCD) (SCD) tipo 2​

O Notebook a seguir contém um exemplo de implementação do SCD tipo 2 usando transformWithState em Python ou Scala.

SCD Tipo 2 Python

Abrir notebook em uma nova abaOpen in Databricks

SCD Tipo 2 Scala

Abrir notebook em uma nova abaOpen in Databricks

Detector de tempo de inatividade​

transformWithState implementa temporizadores para permitir que o usuário tome medidas com base no tempo decorrido, mesmo que nenhum registro de um determinado key seja processado em um micro-lote.

O exemplo a seguir implementa um padrão para um detector de tempo de inatividade. Cada vez que um novo valor é visto para um determinado key, ele atualiza o valor do estado lastSeen, limpa todos os temporizadores existentes e reinicia um temporizador para o futuro.

Quando um cronômetro expira, o aplicativo emite o tempo decorrido desde o último evento observado para o key. Em seguida, ele define um novo cronômetro para emitir uma atualização 10 segundos depois.

Para executar o exemplo de ponta a ponta, semeie uma única leitura de sensor como a fonte de transmissão. Como os temporizadores usam tempo de processamento, o driver usa um trigger processingTime e aguarda antes de interromper a query para que os temporizadores sejam disparados.

Python
import datetime
import time
import uuid
import pandas as pd
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, StringType, TimestampType
from typing import Iterator

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

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 as the elapsed time between the last observed event and now
        downtime_duration = timerValues.getCurrentProcessingTimeInMs() - int(latest_from_existing[0].timestamp() * 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()[0]
        else:
            latest_from_existing = datetime.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

# Create a dedicated schema for the example tables
spark.sql("CREATE SCHEMA IF NOT EXISTS main.stateful_examples")

# Seed a small Delta table with a sensor reading to use as the streaming source
spark.sql("DROP TABLE IF EXISTS main.stateful_examples.sensor_events")
spark.createDataFrame(
    [("sensor1", datetime.datetime(2024, 1, 1, 12, 0, 0))],
    "id string, timestamp timestamp",
).write.saveAsTable("main.stateful_examples.sensor_events")

df = spark.readStream.table("main.stateful_examples.sensor_events")

# Output schema: the key and a time value (processing time or elapsed downtime)
output_schema = StructType([
    StructField("id", StringType(), True),
    StructField("timeValues", StringType(), True),
])

# ProcessingTime mode enables the timers that detect downtime
q = (
    df.groupBy("id")
    .transformWithStateInPandas(
        statefulProcessor=DownTimeDetectorStatefulProcessor(),
        outputStructType=output_schema,
        outputMode="Update",
        timeMode="ProcessingTime",
    )
    .writeStream.format("memory")
    .queryName("downtime_output")
    .option("checkpointLocation", f"/tmp/checkpoint_{uuid.uuid4()}")
    .trigger(processingTime="5 seconds")
    .start()
)

# Wait past the timers so they fire, then stop the query
time.sleep(30)
q.stop()

# When a timer fires, it emits the elapsed time in milliseconds since the last observed event
display(spark.sql("SELECT * FROM downtime_output"))

Migrar informações existentes sobre o estado​

O exemplo a seguir demonstra como implementar um aplicativo com estado que aceita um estado inicial. Você pode adicionar o tratamento do estado inicial a qualquer aplicativo com estado, mas o estado inicial só pode ser definido ao inicializar o aplicativo pela primeira vez.

Este exemplo usa o leitor statestore para carregar informações de estado existentes de um caminho de ponto de verificação. Um exemplo de caso de uso desse padrão é a migração de aplicativos legados com estado para 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

Migrar a tabela Delta para o armazenamento do estado para inicialização​

O Notebook a seguir contém um exemplo de inicialização de valores de armazenamento do estado de uma tabela Delta usando transformWithState em Python ou Scala.

Inicializar o estado a partir do Delta Python

Abrir notebook em uma nova abaOpen in Databricks

Inicializar o estado a partir do Delta Scala

Abrir notebook em uma nova abaOpen in Databricks

Sessão de acompanhamento​

O Notebook a seguir contém um exemplo de acompanhamento de sessão usando transformWithState em Python ou Scala.

Sessão de acompanhamento Python

Abrir notebook em uma nova abaOpen in Databricks

Sessão de acompanhamento Scala

Abrir notebook em uma nova abaOpen in Databricks

Transmissão-transmissão personalizada join usando transformWithState​

O código a seguir demonstra uma transmissão-transmissão personalizada join em várias transmissões usando transformWithState. O senhor pode usar essa abordagem em vez de um operador integrado join pelos seguintes motivos:

  • O senhor precisa usar o modo de saída de atualização que não suporta a união de transmissão-transmissão. Isso é especialmente útil para aplicativos de baixa latência.
  • O senhor precisa continuar a executar a união para as linhas que chegam mais tarde (após a expiração da marca d'água).
  • O senhor precisa realizar uma união de transmissão-transmissão de muitos para muitos.

Este exemplo oferece controle total sobre a lógica de expiração de estado, permitindo a extensão dinâmica do período de retenção para lidar com eventos fora de ordem, mesmo após o watermark.

No exemplo a seguir, eventos de perfil, preferência e atividade chegam em uma única transmissão, cada um marcado com uma record_type tag. O processador armazena cada tipo de registro em buffer no estado, e um temporizador de tempo de processamento emite o join enriquecido pouco tempo após a chegada de um evento de atividade. O estado de perfil e preferência expira após uma hora de inatividade usando um TTL, e cada atividade é limpa do estado assim que é feito o join.

nota

Este exemplo mantém uma atividade por usuário e a limpa após o join emitir. Para manter o foco, ele não lida com múltiplos eventos de atividade chegando para o mesmo usuário antes que o temporizador dispare: uma atividade posterior substitui a anterior, e cada temporizador lê a atividade armazenada em buffer mais recente em vez daquela que a agendou. Para preservar cada atividade, armazene as atividades em um estado de valor de lista ou valor de mapa indexado pelo tempo do evento.

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

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

# 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):
    # Buffer each user's profile, preference, and activity records in state.
    def init(self, handle: StatefulProcessorHandle) -> None:
        self.handle = handle

        profile_schema = StructType([
            StructField("name", StringType(), True),
            StructField("email", StringType(), True)
        ])
        preferences_schema = StructType([
            StructField("preferred_category", StringType(), True)
        ])
        activity_schema = StructType([
            StructField("event_type", StringType(), True),
            StructField("timestamp", TimestampType(), True)
        ])

        # One value state per record type. The grouping key is user_id, so each
        # state holds the latest record of that type for the user.
        # Profile and preference state expire after an hour of inactivity via TTL
        self.profile_state = handle.getValueState("userProfile", profile_schema, ttlDurationMs=3600000)
        self.preferences_state = handle.getValueState("userPreferences", preferences_schema, ttlDurationMs=3600000)
        self.activity_state = handle.getValueState("userActivity", activity_schema)

    # Route each incoming record by its type and buffer it in state. When an
    # activity event arrives, set a timer to emit the enriched join after a delay.
    def handleInputRows(self, key, rows: Iterator[pd.DataFrame], timerValues) -> Iterator[pd.DataFrame]:
        for pdf in rows:
            for _, row in pdf.iterrows():
                record_type = row["record_type"]
                if record_type == "activity":
                    self.activity_state.update((row["event_type"], row["timestamp"]))
                    # Set a timer to process this event after a 10-second delay
                    self.handle.registerTimer(timerValues.getCurrentProcessingTimeInMs() + 10000)
                elif record_type == "profile":
                    self.profile_state.update((row["name"], row["email"]))
                elif record_type == "preference":
                    self.preferences_state.update((row["preferred_category"],))

        # No immediate output; the enriched row is emitted when the timer expires
        return iter([])

    # Perform the lookup after the delay, handling out-of-order and late-arriving records.
    def handleExpiredTimer(self, key, timerValues, expiredTimerInfo) -> Iterator[pd.DataFrame]:
        if not self.activity_state.exists():
            return iter([])

        activity = self.activity_state.get()
        profile = self.profile_state.get() if self.profile_state.exists() else None
        preferences = self.preferences_state.get() if self.preferences_state.exists() else None

        # Combine data from the different states into a single output row
        output_row = {
            "user_id": key[0],
            "event_type": activity[0],
            "timestamp": activity[1],
            "profile_name": profile[0] if profile else None,
            "email": profile[1] if profile else None,
            "preferred_category": preferences[0] if preferences else None
        }
        # The activity has been consumed by this join, so clear it from state
        self.activity_state.clear()
        return iter([pd.DataFrame([output_row])])

    def close(self) -> None:
        pass

# Create a dedicated schema for the example tables
spark.sql("CREATE SCHEMA IF NOT EXISTS main.stateful_examples")

# Seed a small Delta table with profile, preference, and activity records for one user
spark.sql("DROP TABLE IF EXISTS main.stateful_examples.user_events")
input_schema = StructType([
    StructField("user_id", StringType()),
    StructField("record_type", StringType()),
    StructField("event_type", StringType()),
    StructField("timestamp", TimestampType()),
    StructField("name", StringType()),
    StructField("email", StringType()),
    StructField("preferred_category", StringType())
])
spark.createDataFrame(
    [
        ("u1", "profile", None, None, "Alice", "alice@example.com", None),
        ("u1", "preference", None, None, None, None, "electronics"),
        ("u1", "activity", "purchase", datetime(2024, 1, 1, 12, 0, 0), None, None, None),
    ],
    input_schema,
).write.saveAsTable("main.stateful_examples.user_events")

df = spark.readStream.table("main.stateful_examples.user_events")

# Apply transformWithState. ProcessingTime mode enables the timer that fires the join.
q = (
    df.groupBy("user_id")
    .transformWithStateInPandas(
        statefulProcessor=CustomStreamJoinProcessor(),
        outputStructType=output_schema,
        outputMode="Append",
        timeMode="ProcessingTime",
    )
    .writeStream.format("memory")
    .queryName("enriched_events")
    .option("checkpointLocation", f"/tmp/checkpoint_{uuid.uuid4()}")
    .trigger(processingTime="5 seconds")
    .start()
)

# Wait past the 10-second timer so it fires, then stop the query
time.sleep(30)
q.stop()

# The enriched row joins the activity with the buffered profile and preference
display(spark.sql("SELECT * FROM enriched_events"))

Computação Top-K​

O exemplo a seguir usa um ListState com uma fila de prioridade para manter e atualizar os K elementos principais em uma transmissão para cada grupo key em tempo real próximo.

Top-K Python

Abrir notebook em uma nova abaOpen in Databricks

Top-K Scala

Abrir notebook em uma nova abaOpen in Databricks
Nesta página