次の方法で共有

従来の任意のステートフル演算子

注

Databricks では、 transformWithState を使用してカスタムステートフル アプリケーションを構築することをお勧めします。 カスタム ステートフル アプリケーション の構築を参照してください。

この記事では、 mapGroupsWithStateと flatMapGroupsWithStateをサポートする機能について説明します。 これらの演算子の詳細については、 リンクを参照してください。

の初期状態を指定する mapGroupsWithState

flatMapGroupsWithStateor mapGroupsWithStateを使用して、構造化ストリーミングステートフル処理のユーザー定義の初期状態を指定できます。 これにより、有効なチェックポイントなしでステートフル ストリームを開始するときに、データの再処理を回避できます。

def mapGroupsWithState[S: Encoder, U: Encoder](
    timeoutConf: GroupStateTimeout,
    initialState: KeyValueGroupedDataset[K, S])(
    func: (K, Iterator[V], GroupState[S]) => U): Dataset[U]

def flatMapGroupsWithState[S: Encoder, U: Encoder](
    outputMode: OutputMode,
    timeoutConf: GroupStateTimeout,
    initialState: KeyValueGroupedDataset[K, S])(
    func: (K, Iterator[V], GroupState[S]) => Iterator[U])

flatMapGroupsWithState演算子の初期状態を指定するユース ケースの例:

val fruitCountFunc =(key: String, values: Iterator[String], state: GroupState[RunningCount]) => {
  val count = state.getOption.map(_.count).getOrElse(0L) + valList.size
  state.update(new RunningCount(count))
  Iterator((key, count.toString))
}

val fruitCountInitialDS: Dataset[(String, RunningCount)] = Seq(
  ("apple", new RunningCount(1)),
  ("orange", new RunningCount(2)),
  ("mango", new RunningCount(5)),
).toDS()

val fruitCountInitial = initialState.groupByKey(x => x._1).mapValues(_._2)

fruitStream
  .groupByKey(x => x)
  .flatMapGroupsWithState(Update, GroupStateTimeout.NoTimeout, fruitCountInitial)(fruitCountFunc)

mapGroupsWithState演算子の初期状態を指定するユース ケースの例:

val fruitCountFunc =(key: String, values: Iterator[String], state: GroupState[RunningCount]) => {
  val count = state.getOption.map(_.count).getOrElse(0L) + valList.size
  state.update(new RunningCount(count))
  (key, count.toString)
}

val fruitCountInitialDS: Dataset[(String, RunningCount)] = Seq(
  ("apple", new RunningCount(1)),
  ("orange", new RunningCount(2)),
  ("mango", new RunningCount(5)),
).toDS()

val fruitCountInitial = initialState.groupByKey(x => x._1).mapValues(_._2)

fruitStream
  .groupByKey(x => x)
  .mapGroupsWithState(GroupStateTimeout.NoTimeout, fruitCountInitial)(fruitCountFunc)

mapGroupsWithState更新関数をテストする

TestGroupState API を使用すると、Dataset.groupByKey(...).mapGroupsWithState(...)とDataset.groupByKey(...).flatMapGroupsWithState(...)に使用される状態更新関数をテストできます。

状態更新関数は、 GroupState型のオブジェクトを使用して、以前の状態を入力として受け取ります。 Apache Spark GroupState リファレンス ドキュメントを参照してください。 例えば次が挙げられます。

import org.apache.spark.sql.streaming._
import org.apache.spark.api.java.Optional

test("flatMapGroupsWithState's state update function") {
  var prevState = TestGroupState.create[UserStatus](
    optionalState = Optional.empty[UserStatus],
    timeoutConf = GroupStateTimeout.EventTimeTimeout,
    batchProcessingTimeMs = 1L,
    eventTimeWatermarkMs = Optional.of(1L),
    hasTimedOut = false)

  val userId: String = ...
  val actions: Iterator[UserAction] = ...

  assert(!prevState.hasUpdated)

  updateState(userId, actions, prevState)

  assert(prevState.hasUpdated)
}
  • Last updated on