カスタム ステートフル アプリケーションを構築する transformWithState

transformWithStateを使用してステートフル ストリーミング アプリケーションを構築し、待機時間が短く、ほぼリアルタイムのソリューションを実装できます。 カスタムステートフル演算子を使用すると、従来の構造化ストリーミング処理では不可能な新しい運用ユース ケースを構築できる任意のステートフル ロジックを作成できます。

要求事項

transformWithState演算子とtransformWithStateInPandas演算子には、次の要件があります。

• Databricks Runtime 16.2 以降で使用できます。
  • リアルタイム モードの場合は、Databricks Runtime 17.3 LTS 以降を使用します。 リアルタイム モードの概念を参照してください。
  • 標準アクセス モードの場合、Pythonは Databricks Runtime 16.3 以降で使用でき、Scala は Databricks Runtime 17.3 以降で使用できます。
• RocksDB は、Databricks Runtime 17.3 以降の既定の状態ストア プロバイダーです。

  • Databricks Runtime 17.2 以降では、RocksDB ステート ストア プロバイダーを構成する必要があります。 Databricks では、Spark 構成で RocksDB を有効にすることをお勧めします。

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

transformWithState とは

transformWithState演算子は、構造化ストリーミング クエリにカスタム ステートフル プロセッサを適用します。 transformWithStateを使用するには、カスタム ステートフル プロセッサを実装する必要があります。 構造化ストリーミングには、Python、Scala、またはJavaを使用してステートフル プロセッサを構築するための API が含まれています。

transformWithStateを使用して、グループ化キーにカスタム ロジックを適用します。 次に、大まかな設計について説明します。

• 1 つ以上の状態変数を定義します。
• 状態情報は、グループ化キーごとに保持されます。 ユーザー定義コードで各状態変数にアクセスできます。
• 処理されたマイクロ バッチごとに、キーのすべての行を反復子として使用できます。
• タイマーとユーザー定義の条件で StatefulProcessorHandle を使用して、行の出力方法を制御します。
• 状態の有効期限と状態サイズを管理するために、状態値は個々の Time-to-Live (TTL) 定義をサポートします。

transformWithStateは状態ストアでのスキーマの進化をサポートしているため、状態の履歴情報を失うことなく、運用アプリケーションを反復処理および更新できます。 状態スキーマを更新した後、行を再処理する必要はありません。これにより、コードのデプロイとメンテナンスが簡略化されます。 状態ストアでのスキーマの進化を参照してください。

StatefulProcessor の定義

ステートフル プロセッサを定義する場合は、 StatefulProcessor クラスを拡張し、そのメソッドを実装します。

Spark は、お使いのStatefulProcessorHandleのinitメソッドにStatefulProcessorを渡します。 ハンドルを使用して状態変数を作成し、状態ストアと対話します。

transformWithState では、 ValueState、 ListState、 MapStateの 3 つの状態の種類がサポートされています。 各型は、異なる基になるデータ構造を使用して、各グループ化キーの状態を格納します。

カスタム ロジックを定義するには、次のメソッドを実装します。

• アプリケーションがデータを処理し、状態を更新し、マイクロバッチごとに行を出力する方法を制御する handleInputRows を実装します。 「 入力行の処理」を参照してください。
• グループ化キーがマイクロバッチで新しい行を受け取るかどうかに関係なく、時間ベースのロジックを実行する handleExpiredTimer を実装します。 「期限切れのタイマーを処理する」を参照してください。
• 必要に応じて、アプリケーションが入力行を処理する前に状態を事前設定する handleInitialState を実装します。 「 初期状態の処理」を参照してください。

次の表は、これらのメソッドの機能動作を比較しています。

handleInputRowsとhandleExpiredTimerの両方を組み合わせて、必要に応じて複雑なロジックを実装できます。

たとえば、 handleInputRows を使用して各マイクロ バッチの状態値を更新し、タイマーを 10 秒後に設定するアプリケーションを実装できます。 追加の行が処理されない場合は、 handleExpiredTimer を使用して状態ストアの現在の値を出力できます。 グループ化キーに対して新しい行が処理される場合は、既存のタイマーをクリアして新しいタイマーを設定できます。

StatefulProcessorHandle

PySpark では、 StatefulProcessorHandle クラスを使用して、コードでの状態情報の使用方法を制御する関数にアクセスできます。

StatefulProcessorを初期化するときは、常にStatefulProcessorHandleをインポートしてhandle変数に渡す必要があります。 handle 変数は、Python クラスのローカル変数を状態変数に結び付けます。

カスタム状態の種類

1 つのステートフル演算子に複数の状態オブジェクトを実装できます。

完全なアプリケーション ロジックに基づいて状態の種類を選択します。 たとえば、ValueStateとuser_idでグループ化されたsession_idを使用してセッションを追跡できます。 または、複数のセッション間で条件を評価するには、マップ キーとして MapState を使用して、user_id で グループ化された session_id を使用します。

状態オブジェクトで StructTypeを使用する場合は、スキーマの構造体内の各フィールドに一意の名前を定義する必要があります。 これらの名前は、状態ストアを読み取るときに表示されます。 「構造化ストリーミング状態情報の読み取り」をご覧ください。

次のセクションでは、 transformWithStateでサポートされる状態の種類について説明します。

ValueState

ValueState には、グループ化キーごとに値が格納されます。

値の状態には、構造体やタプルなどの複合型を含めることができます。 ValueStateの場合は、値全体を置き換えるロジックを実装する必要があります。

値の状態の有効期間は、値が更新されるとリセットされます。 保存されているValueStateを更新せずにValueStateのソース キーを処理した場合、有効期限はリセットされません。

ListState

ListState は、グループ化キーごとにリストを格納します。

リスト状態は値のコレクションであり、それぞれに複合型を含めることができます。 リスト内の各値には、独自の有効期限があります。

リストに項目を追加するには、個々の項目を追加するか、項目のリストを追加するか、リスト全体を putで上書きします。 有効期間をリセットするには、 put 操作を使用する必要があります。

MapState

MapState では、グループ化キーごとにマップが格納されます。 マップは、Python ディクショナリ (dict) と同等の Apache Spark です。

マップ状態は、それぞれが値にマップされる個別のキーのコレクションであり、それぞれに複合型を含めることができます。 マップ内の各キーと値のペアには、独自の有効期限があります。

特定のキーの値を更新することも、キーとその値を削除することもできます。 個々の値は、そのキーを使用して返したり、すべてのキーを一覧表示したり、すべての値を一覧表示したり、マップ内のキーと値のペアの完全なセットを操作するための反復子を返したりすることができます。

たとえば、クエリで GROUP BY user_id を使用し、各 session_idのマップを定義する場合、グループ化キーは user_id され、 MapState キーは session_id。

Python

class SessionTracker(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    self.sessions = handle.getMapState("sessions", "session_id string", "count long")

  def handleInputRows(self, key, rows: Iterator[Row], timerValues) -> Iterator[Row]:
    for row in rows:
      session_key = (row["session_id"],)  # session_id is the MapState key
      count = self.sessions.getValue(session_key)[0] if self.sessions.containsKey(session_key) else 0
      new_count = count + 1
      self.sessions.updateValue(session_key, (new_count,))
    yield from []

  def close(self) -> None:
    pass

df.groupBy("user_id").transformWithState(SessionTracker(), ...) # user_id is the grouping key

スカラ (プログラミング言語)

case class Event(userId: String, sessionId: String)

class SessionTracker extends StatefulProcessor[String, Event, (String, Long)] {
  @transient private var sessions: MapState[String, Long] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    sessions = getHandle.getMapState[String, Long]("sessions", Encoders.STRING, Encoders.scalaLong, TTLConfig.NONE)
  }

  override def handleInputRows(
      key: String,
      rows: Iterator[Event],
      timerValues: TimerValues): Iterator[(String, Long)] = {
    rows.foreach { event =>
      val count = if (sessions.containsKey(event.sessionId)) sessions.getValue(event.sessionId) else 0L
      sessions.updateValue(event.sessionId, count + 1) // sessionId is the MapState key
    }
    Iterator.empty
  }
}

df.as[Event]
  .groupByKey(_.userId) // userId is the grouping key
  .transformWithState(new SessionTracker(), TimeMode.None(), OutputMode.Update())

でカスタム状態変数を作成します。 StatefulProcessor

StatefulProcessorを初期化するときは、カスタム ロジックで状態オブジェクトを操作できる状態オブジェクトごとにローカル変数を作成します。 init クラスの組み込みのStatefulProcessor メソッドをオーバーライドして、状態変数を定義および初期化します。

getValueStateのgetListState、getMapState、およびStatefulProcessorメソッドを使用して、任意の数の状態オブジェクトを定義できます。

各状態オブジェクトには、次のものが必要です。

• ユニークな名前
• スキーマ
  • Pythonでは、スキーマを指定する必要があります。
  • Scala では、 Encoder を渡して状態スキーマを指定できます。

必要に応じて、Time to Live (TTL) 期間をミリ秒単位で指定することもできます。 マップ状態を実装する場合は、マップ キーと値に別のスキーマ定義を指定する必要があります。

カスタム ロジックを持つメソッドで状態変数を使用する

状態オブジェクトには、状態の取得、既存の状態情報の更新、および現在の状態のクリアを行うメソッドがあります。

各グループ化キーには、専用の状態情報があります。

• StatefulProcessorは、カスタム ロジックと指定された出力スキーマに基づいて行を出力します。 行を出力を参照してください。
• statestore リーダーを使用して、状態ストアの値にアクセスします。 このリーダーはバッチ ワークロードを対象としており、待ち時間の短いワークロードを対象としていません。 「構造化ストリーミング状態情報の読み取り」をご覧ください。
• handleInputRowsを使用して指定されたロジックは、キーの行がマイクロバッチ内に存在する場合にのみ実行されます。 「 入力行の処理」を参照してください。
• handleExpiredTimer を使用すると、行を監視して発火させることに依存しない時間ベースのロジックを実装できます。 「期限切れのタイマーを処理する」を参照してください。

グループ化キー内の値を比較できます。 MapStateを使用して、カスタム ロジックで使用できる 2 番目のキーを持つロジックを実装します。 たとえば、user_idでグループ化し、ip_address キーにMapStateを使用すると、同時にユーザー セッションを追跡できます。

状態を操作するための高度な考慮事項

状態の更新は耐障害性があります。 マイクロバッチの処理が完了する前にタスクがクラッシュした場合、再試行では最後に成功したマイクロバッチの値が使用されます。

パフォーマンスを最適化するために、Databricks では、特定のキーの反復子内のすべての値を処理し、更新を 1 回の書き込みでコミットすることをお勧めします。 状態変数に書き込むと、RocksDB への書き込みがトリガーされます。

状態の値には既定値がありません。 ロジックで既存の状態情報を読み取る必要がある場合は、 exists メソッドを使用します。

null 状態のロジックを実装するために、 MapState 変数を使用すると、個々のキーを確認したり、すべてのキーを一覧表示したりできます。

入力行の処理

handleInputRowsメソッドを使用して、アプリケーションが行を処理し、状態値を更新する方法を定義します。 このメソッドは、構造化ストリーミング クエリがグループ化キーの行を処理するたびに実行されます。

transformWithStateで実装されているほとんどのステートフル アプリケーションでは、コア ロジックは handleInputRows を使用して定義されます。

処理されたマイクロバッチ更新ごとに、特定のグループ化キーのマイクロバッチ内のすべての行を反復子を使用して使用できます。 ユーザー定義ロジックは、現在のマイクロバッチのすべての行とステートストア内の値と対話できます。

期限切れのタイマーを処理する

handleExpiredTimer メソッドを使用して、経過時間に基づいてカスタム ロジックを実装します。

グループ化キー内では、タイマーはタイムスタンプによって一意に識別されます。

タイマーの有効期限が切れると、結果はアプリケーションに実装されているロジックによって決まります。 一般的なパターンは次のとおりです。

• 状態変数に格納されている情報を出力します。
• 格納されている状態情報の削除します。
• 新しいタイマーの作成。

期限切れのタイマーは、関連するキーに対応する行がマイクロバッチ内で1行も処理されない場合でも起動します。

時間モードを指定する

StatefulProcessorをtransformWithStateに渡す場合は、timeMode パラメーターを使用して時間モードを指定する必要があります。

次のオプションがサポートされています。

eventTimeColumnName

EventTime時間モードを使用する場合、eventTimeColumnName パラメーターは、イベント タイムスタンプを含む出力スキーマ内の列の名前を指定します。 Apache Spark では、この列を使用してウォーターマークを出力ストリームに伝達し、ダウンストリームの時間ベースの正しい操作を有効にします。

Python

eventTimeColumnName は、 transformWithState または transformWithStateInPandasの追加引数です。

q = (
  df.groupBy("key")
    .transformWithState(
      statefulProcessor=MyProcessor(),
      outputStructType=output_schema,
      outputMode="Append",
      timeMode="EventTime",
      eventTimeColumnName="outputTimestamp",
    )
    .writeStream...
)

スカラ (プログラミング言語)

transformWithStateは、eventTimeColumnNameの代わりにtimeModeを受け入れます。 この方法では、常に EventTime モードが使用されます。

val q = spark
  .readStream
  .format("delta")
  .load(srcDeltaTableDir)
  .as[(String, String)]
  .groupByKey(x => x._1)
  .transformWithState(
    new MyProcessor(),
    "outputTimestamp",
    OutputMode.Append(),
  )
  .writeStream...

組み込みのタイマー値

Databricks では、カスタム ステートフル アプリケーションでシステム クロックを呼び出すのは推奨しません。これにより、タスクの失敗時に再試行が信頼性に欠ける可能性があります。 処理時間または基準値にアクセスする必要がある場合は、 TimerValues クラスのメソッドを使用します。

状態タイプの Time-to-Live (TTL)

メモリ不足エラーを防ぎ、古い状態の種類の値を削除するために、 transformWithState では、各状態の種類の値に対してオプションの Time to Live (TTL) 値がサポートされます。 有効期限が切れると、TTL によって状態の種類の値が自動的に削除されます。 TTL は、 handleExpiredTimer またはカスタム ロジックを実行しません。 状態の有効期限が切れたときにコードを実行するには、代わりにタイマーを使用します。

すべての状態の種類について、状態情報を更新すると TTL がリセットされます。 TTL は状態の種類の値ごとに適用され、状態の種類ごとに異なる規則が適用されます。

• 状態変数のスコープは、キーのグループ化に設定されます。
• ValueState オブジェクトの場合、グループ化キーごとに格納される値は 1 つだけです。 TTL はこの値に適用されます。
• ListStateオブジェクトの場合、リストには多数の値を含めることができます。 TTL は、リスト内の各値に個別に適用されます。
  • TTL はListStateの個々の値にスコープ設定されますが、個々の値を更新する唯一の方法は、put変数の内容全体を上書きし、リスト内のすべての値の TTL をリセットするListStateメソッドを使用することです。
• MapState オブジェクトの場合、各マップ キーには関連付けられた状態値があります。 TTL は、マップ内の各キーと値のペアに個別に適用されます。

状態保持型アプリケーションの例

次の例では、状態変数の例を含め、カスタムステートフル プロセッサ ( SimpleCounterProcessor) を定義します。 SimpleCounterProcessor では、 ValueState、 ListState、および MapState を使用して、グループ化キーごとに行をカウントします。

Python(パンダ)

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]:
    # Seed the running total from state so the count accumulates across micro-batches
    count = self.value_state.get()[0] if self.value_state.exists() else 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(iter)[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...
)

Python (行ベース)

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)
    self.map_state = handle.getMapState(stateName="mapState", userKeySchema="name string", valueSchema="count int")

  def handleInputRows(self, key, rows: Iterator[Row], timerValues) -> Iterator[Row]:
    # Seed the running total from state so the count accumulates across micro-batches
    count = self.value_state.get()[0] if self.value_state.exists() else 0
    for row 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)
      count += 1
    self.value_state.update((count,))  # Count is passed as a tuple
    iter_list = self.list_state.get()
    list_state_value = next(iter_list)[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 Row(id=key[0], countAsString=str(count))

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

スカラ (プログラミング言語)

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...

例をエンドツーエンドで実行してください

上記のプロセッサはステートフルロジックを定義しますが、クエリは開始しません。 コピー&ペーストで を実行するには、小さなDelta Lakeテーブルをストリーミングソースとしてシードし、インメモリシンクに書き込みを行うクエリを開始します。 この例では Trigger.AvailableNow を使い、クエリはシードされた行を処理し、停止します。 ソースをシードしてクエリを開始するには、以下を実行します:

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.tws_counter_source")
spark.createDataFrame(
  [("a", "1"), ("a", "2"), ("a", "3"), ("b", "1"), ("b", "2")],
  "key string, value string",
).write.saveAsTable("main.stateful_examples.tws_counter_source")

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

q = (
  df.groupBy("key")
    .transformWithState(
      statefulProcessor=SimpleCounterProcessor(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
    )
    .writeStream.format("memory")
    .queryName("counter_output")
    .option("checkpointLocation", f"/tmp/checkpoint_{uuid.uuid4()}")
    .trigger(availableNow=True)
    .start()
)

q.awaitTermination()

クエリが完了した後、各グルーピングキーのカウントを確認します:

display(spark.sql("SELECT id, countAsString FROM counter_output ORDER BY id"))

aキーは3行、bキーは2行なので、クエリは次の返しをします:

id  countAsString
a   3
b   2

その他の例については、「 ステートフル アプリケーションの例」を参照してください。

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

行を出力する

handleInputRowsまたはhandleExpiredTimerを使用して、グループ化キーごとに行transformWithState出力する方法を定義する必要があります。 「 入力行の処理」 および 「期限切れのタイマーの処理」を参照してください。

カスタム ステートフル アプリケーションでは、状態情報の使用方法を想定しません。 特定の条件では、アプリケーションは 0 行、1 行、または複数行を出力する場合があります。

Python(パンダ)

transformWithStateInPandasでは、outputStructType キーワードを使用して出力スキーマを定義します。

pandas DataFrame オブジェクトと yieldを使用して行を出力します。

必要に応じて、空の DataFrame を yield できます。 update出力モードを使用し、空の DataFrame を出力すると、グループ化キーの値がnullに更新されます。

Python (行ベース)

transformWithStateでは、outputStructType キーワードを使用して出力スキーマを定義します。

Row オブジェクトとyieldを使用して行を出力します。

必要に応じて、空の反復子を返すことができます。 出力モード update 使用して空の反復子を出力すると、グループ化キーの値が nullに更新されます。

スカラ (プログラミング言語)

Scala では、 Iterator オブジェクトを使用して行を出力します。 スキーマは、出力された行のスキーマから自動的に派生します。

必要に応じて、空の Iteratorを返すことができます。 update出力モードを使用し、空のIteratorを出力すると、グループ化キーの値がnullに更新されます。

初期状態を処理する

必要に応じて、最初のマイクロバッチに初期状態を渡すことができます。

たとえば、これを使用して次の操作を行うことができます。

• 既存のワークフローを新しいカスタム アプリケーションに移行します。
• ステートフル演算子をアップグレードして、スキーマまたはロジックを変更します。
• 自動的に修復できない、手動による介入が必要な障害を修復します。

入力行と同じグループ化キー スキーマを持つ DataFrame を使用して初期状態を指定します。

次の例では、既存の Delta テーブルからキーごとのカウンターをシードします。

Python (行ベース)

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

class CounterWithInitialState(StatefulProcessor):
  def init(self, handle: StatefulProcessorHandle) -> None:
    state_schema = StructType([StructField("count", IntegerType(), True)])
    self.count_state = handle.getValueState("countState", state_schema)

  def handleInitialState(self, key, initialState: Row, timerValues) -> None:
    self.count_state.update((initialState["count"],))

  def handleInputRows(self, key, rows: Iterator[Row], timerValues) -> Iterator[Row]:
    count = self.count_state.get()[0] if self.count_state.exists() else 0
    for _ in rows:
      count += 1
    self.count_state.update((count,))
    yield Row(id=key[0], count=count)

  def close(self) -> None:
    pass

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

import uuid

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

# Seed existing per-key counts to load as the initial state
spark.sql("DROP TABLE IF EXISTS main.stateful_examples.existing_counts")
spark.createDataFrame(
  [("x", 10)],
  "id string, count int",
).write.saveAsTable("main.stateful_examples.existing_counts")

# Seed a small Delta table to use as the streaming source
spark.sql("DROP TABLE IF EXISTS main.stateful_examples.tws_initial_source")
spark.createDataFrame(
  [("x", "a"), ("x", "b")],
  "id string, value string",
).write.saveAsTable("main.stateful_examples.tws_initial_source")

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

# Load existing counts as initial state — must use the same grouping key as the input
initial_state = spark.read.table("main.stateful_examples.existing_counts").groupBy("id")

q = (
  df.groupBy("id")
    .transformWithState(
      statefulProcessor=CounterWithInitialState(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
      initialState=initial_state,
    )
    .writeStream.format("memory")
    .queryName("initial_state_output")
    .option("checkpointLocation", f"/tmp/checkpoint_{uuid.uuid4()}")
    .trigger(availableNow=True)
    .start()
)

q.awaitTermination()

# The initial state seeds "x" with 10, and the source adds two rows, so the count is 12
display(spark.sql("SELECT id, count FROM initial_state_output ORDER BY id"))

スカラ (プログラミング言語)

import org.apache.spark.sql.streaming._
import org.apache.spark.sql.Encoders

class CounterWithInitialState
    extends StatefulProcessorWithInitialState[String, (String, String), (String, String), (String, Int)] {

  @transient private var countState: ValueState[Int] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    countState = getHandle.getValueState[Int]("countState", Encoders.scalaInt, TTLConfig.NONE)
  }

  override def handleInitialState(
      key: String, initialState: (String, Int), timerValues: TimerValues): Unit = {
    countState.update(initialState._2)
  }

  override def handleInputRows(
      key: String,
      rows: Iterator[(String, String)],
      timerValues: TimerValues): Iterator[(String, String)] = {
    val count = if (countState.exists()) countState.get() else 0
    val newCount = count + rows.size
    countState.update(newCount)
    Iterator((key, newCount.toString))
  }
}

// Load existing counts as initial state — must use the same grouping key as the input
val initialState = spark.read.table("existing_counts")
  .as[(String, Int)]
  .groupByKey(_._1)

val q = spark
  .readStream
  .format("delta")
  .load(srcDeltaTableDir)
  .as[(String, String)]
  .groupByKey(_._1)
  .transformWithState(
    new CounterWithInitialState(),
    TimeMode.None(),
    OutputMode.Update(),
    initialState,
  )
  .writeStream...

非同期処理(ベータ版)

Python transformWithStateはasyncioを用いた非同期処理をサポートし、状態操作とユーザーロジックを同時に実行します。 非同期処理は同期処理よりもスループットが高く、サードパーティの非同期ライブラリを用いず、わずかなコード変更のみで済みます。 非同期処理を使用するには、同期AsyncStatefulProcessorの代わりにStatefulProcessorを実装してください。 詳細は 「 transformWithState (ベータ版)を用いた非同期処理」を参照してください。

Lakeflow パイプラインで transformWithState を使用する

Pythonを使用してストリーミング パイプラインに任意のステートフル ロジックを実装するには、Lakeflow パイプライン内のtransformWithState演算子を使用します。

そのためには、次の手順を実行します。

1. 任意のステートフル変換の出力スキーマとステートフル プロセッサ ロジックを定義します。 例については、 ステートフル アプリケーションの例を参照してください。
2. DataFrame で transformWithState 演算子を呼び出す Lakeflow パイプライン フローを作成します。 「 チュートリアル: Lakeflow パイプライン エディターを使用して最初のパイプラインを作成する」を参照してください。
3. パイプラインを実行し、ターゲット テーブルまたはシンクで結果を検証します。

transformWithStateを使用してセンサーのハートビートを監視する例については、「例: transformWithStateを使用してセンサーのハートビートを監視する」を参照してください。