spark mapWithState updated states出力

Question

Spark 1.6.1にアップグレードした後、updateStateByKeyをmapWithStateに置き換えるようにアプリケーションのリファクタリングを開始しました。

新しいAPIのパフォーマンス上の利点を利用するために、すべての状態をロードするstateSnapshotsに電話する必要はありません。私は更新された状態だけを必要とします。

mapWithState APIはDStreamの[key, input, state, output]を返します。各状態は、入力が取り込まれた後に部分的に更新された状態です。このDStream（つまり、対応するすべての入力がインジェスト/マップされた後の状態）から最新の状態だけを抽出するにはどうすればよいですか？

私は、MapWithStateDStream上map（入力と出力をドロップする）とreduceByKeyを行う（私は更新関数の内部で設定）より新しいタイムスタンプを持つ状態を選択することが、私はそこではないという保証を持つことはできません同じタイムスタンプを持つ2つの部分的な状態。たとえカスタム、キー、パーティショナーを使用しています。

出力MapWithStateDStreamの最新の部分状態を確認するには、mapWithStateを使用しますか？

Answer 1

mapWithStateは、現在のマイクロバッチで更新されている各状態に対してのみ呼び出されます。あなたが望むものを達成する1つの方法は、状態が更新された場合にSome[S]を返すことです。

StateSpec.functionは、次のシグネチャを持つメソッドをとります。

mappingFunction: 
    (Time, KeyType, Option[ValueType], State[StateType]) => Option[MappedType] 

私たちは何ができるか、私たちOption[MappedType]がそうでなければNone、値が更新されたとき、常にSome[MappedType]であることを確認しています。

例えば

：

def updateState(key: Int, value: Option[Int], state: State[Int]): Option[Int] = { 
    value match { 
     case Some(something) if something > 10 => 
     val updatedVal = something * something 
     state.update(updatedVal) 
     Some(updatedVal) 
     case _ => None 
    } 
} 

そして、あなたが行うことができます：

val spec = StateSpec.function(updateState _) 
ssc.mapWithState(spec).filter(!_.isEmpty).foreachRDD(/* do stuff on updated state */) 

あなたはどのなし更新された状態をフィルタリングし、あなたが探しているだけで、更新のスナップショットを維持するこの方法です。

Answer 2

更新アルゴリズムで可能な場合は、mapWithstateを呼び出す前に入力ストリームのreduceByKeyを呼び出すことができます。次に、各キーに対して1つの更新のみがあり、部分的な状態出力はありません。