メインコンテンツまでスキップ
に最終更新

ステートフルなアプリケーションの例

このページには、transformWithState 演算子を使用したカスタムステートフルストリーミングアプリケーションのコード例が記載されています。Databricks では、集計や結合などの一般的な操作には、組み込みのステートフルメソッドを使用することをお勧めします。

「transformWithState を使用したカスタム ステートフル アプリケーションの構築」を参照してください。

注記

Python は、行ベースのtransformWithState API (マイクロバッチ モードとリアルタイム モードで使用可能) と Pandas ベースのtransformWithStateInPandas演算子の両方をサポートしています。以下の例は、Python のtransformWithStateInPandasと Scala のtransformWithStateを使用したコードを示しています。

注記

このページの実行可能な例では、専用の main.stateful_examples スキーマ内にテーブルを作成するため、既存のデータに影響を与えることなく実行できます。main カタログでスキーマを作成する権限がない場合は、例のカタログとスキーマをテーブルを作成できる場所に変更してください。

必要条件​

transformWithState演算子と関連する API およびクラスには、次の要件があります。

  • Databricks Runtime 16.2 以降で使用できます。
  • 標準アクセス モードは、Databricks Runtime 16.3 以降の Python ( transformWithStateInPandasおよび行ベースtransformWithState )、および Databricks Runtime 17.3 以降の Scala ( transformWithState ) でサポートされています。
  • RocksDB は、Databricks Runtime 17.3 以降のデフォルトの状態ストア プロバイダーです。Databricks Runtime バージョン 17.3 より前のバージョンでは、RocksDB 状態ストア プロバイダーを構成する必要があります。Databricks 、コンピュート構成の一部としてRocksDB有効にすることをお勧めします。
注記

17.3 より前の Databricks Runtime バージョンでは、次のコマンドを実行して、現在のセッションの RocksDB 状態ストア プロバイダーを有効にします。

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

緩やかに変化する寸法(SCD)タイプ1​

次のコードは、 transformWithStateを使用して SCD タイプ 1 を実装する例です。 SCD タイプ 1 は、特定のフィールドの最新の値のみを追跡します。

注記

Delta Lakeをバックエンドとするテーブルを使用して、ストリーミングテーブルとAUTO CDC ... INTOを使用してSCDタイプ1またはタイプ2を実装できます。この例では、状態ストアにSCDタイプ1を実装しています。これにより、ほぼリアルタイムのアプリケーションにおいて低遅延を実現します。

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

緩やかに変化する寸法 (SCD) タイプ2​

次のノートブックには、Python または Scala で transformWithState を使用して SCD タイプ 2 を実装する例が含まれています。

SCD タイプ2 Python

ノートブックを新しいタブで開くOpen in Databricks

SCD タイプ2 Scala

ノートブックを新しいタブで開くOpen in Databricks

ダウンタイム検出器​

transformWithState タイマーを実装して、特定のキーのレコードがマイクロバッチで処理されない場合でも、経過時間に基づいてアクションを実行できるようにします。

次の例では、ダウンタイム検出機能のパターンを実装します。 特定のキーに新しい値が表示されるたびに、 lastSeen 状態値が更新され、既存のタイマーがクリアされ、将来のタイマーがリセットされます。

タイマーの有効期限が切れると、アプリケーションはキーに対して最後に観察されたイベントからの経過時間を出力します。その後、新しいタイマーを設定して、10 秒後に更新を出力します。

例をエンドツーエンドで実行するには、単一のセンサー読み取り値をストリーミングソースとしてシードします。タイマーは処理時間を使用するため、ドライバーは processingTime trigger を使用し、タイマーが起動するようにクエリーを停止する前に待機します。

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

既存の状態情報を移行する​

次の例は、初期状態を受け入れるステートフル アプリケーションを実装する方法を示しています。 初期状態処理は任意のステートフル アプリケーションに追加できますが、初期状態はアプリケーションを最初に初期化するときにのみ設定できます。

この例では、 statestore リーダーを使用して、チェックポイント パスから既存の状態情報を読み込みます。 このパターンの使用例として、従来のステートフルアプリケーションから 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

初期化のために Delta テーブルを状態ストアに移行する​

次のノートブックには、Delta transformWithStatePythonまたは の を使用して テーブルから状態ストアの値を初期化する例が含まれています。Scala

Delta Python から状態を初期化する

ノートブックを新しいタブで開くOpen in Databricks

Delta Scalaから状態を初期化する

ノートブックを新しいタブで開くOpen in Databricks

セッション追跡​

次のノートブックには、Python または Scala の transformWithState を使用したセッション追跡の例が含まれています。

セッション追跡 Python

ノートブックを新しいタブで開くOpen in Databricks

セッショントラッキング Scala

ノートブックを新しいタブで開くOpen in Databricks

カスタムストリーム-ストリーム結合 transformWithState​

次のコードは、 transformWithState. 次の理由により、組み込みの結合演算子の代わりにこの方法を使用できます。

  • ストリーム-ストリーム結合をサポートしていない更新出力モードを使用する必要があります。 これは、低レイテンシのアプリケーションに特に役立ちます。
  • 遅れて到着した行 (ウォーターマークの有効期限が切れた後) の結合を引き続き実行する必要があります。
  • 多対多のストリーム-ストリーム結合を実行する必要があります。

この例では、状態の有効期限ロジックを完全に制御できるため、ウォーターマークの後であっても順不同のイベントを処理できるように保持期間を動的に延長できます。

次の例では、プロファイル、プリファレンス、およびアクティビティイベントが単一のストリームに到着し、それぞれが record_type でタグ付けされています。プロセッサは各レコードタイプを状態内にバッファリングし、処理時間のタイマーがアクティビティイベントの到着後、短時間でエンリッチされた結合を出力します。プロファイルとプリファレンスの状態は、TTLを使用して1時間非アクティブ状態が続くと期限切れになり、各アクティビティは結合されると状態からクリアされます。

注記

この例では、ユーザーごとに1つのアクティビティを保持し、結合が発行された後にそれをクリアします。処理を集中させるため、タイマーが起動する前に同じユーザーに対して複数のアクティビティイベントが到着した場合、それらは処理されません。後のアクティビティが前のものを置き換え、各タイマーはそれをスケジュールしたものよりも、最新のバッファリングされたアクティビティを読み取ります。すべてのアクティビティを保持するには、イベント時間をキーとして、リスト値またはマップ値の状態にアクティビティをバッファリングします。

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

Top-K 計算​

次の例では、優先キューを持つ ListState を使用して、各グループ キーのストリーム内の上位 K 要素をほぼリアルタイムで維持および更新します。

トップKの Python

ノートブックを新しいタブで開くOpen in Databricks

トップKの Scala

ノートブックを新しいタブで開くOpen in Databricks
このページの見出し