SPARK - 2つのデータストリームの結合 - キャッシュのメンテナンス

Question

スパークストリーミングのアウトオブボックス結合機能は、実際の使用の多くのケースを保証するものではありません。その理由は、マイクロバッチRDDに含まれるデータだけを結合するからです。

2つのカフカストリームのデータを結合し、stream1の各オブジェクトをstream2の対応するオブジェクトでスパークさせて豊かにし、それをHBaseに保存します。

実装

• のようにオブジェクトを追加または交換、ストリーム2のオブジェクトからメモリ内のデータセットを維持し、それらはストリーム1内のすべての要素に対して

• を受け取っている場合、検索しキャッシュにアクセスします一致した場合はHBaseに保存し、一致しない場合はカフカストリームに戻します。

この質問は、スパークストリーミングの調査と上記の実装方法を見つけるAPIです。

Answer 1

：基本的にはあなたが何かを埋める「累計」RDDを保ちます。これはupdateStateByKeyの方が効率的です。これらはタイプKのいくつかのキーで識別されているので、ストリーム2タイプVのあなたのオブジェクトを想定し、PairDStreamFunction上で定義されている、あなたの最初のポイントは次のように行くだろう：

def stream2: DStream[(K, V)] = ??? 

def maintainStream2Objects(key: K, value: Option[V], state: State[V]): (K, V) = { 
    value.foreach(state.update(_)) 
    (key, state.get()) 
} 

val spec = StateSpec.function(maintainStream2Objects) 

val stream2State = stream2.mapWithState(spec) 

stream2Stateは今、各バッチが(K, V)を含むストリームであります各キーの最新の値とペアにします。このストリームで結合を行い、stream1を使用して、2番目のポイントのさらなるロジックを実行することができます。

Answer 2

RDDを他のRDDに参加させることができます。マイクロバッチ内のものだけでなく、RDDも参加できます。良いスタートがmapWithStateに見えることであろう

var globalRDD1: RDD[...] = sc.emptyRDD 
var globalRDD2: RDD[...] = sc.emptyRDD 

dstream1.foreachRDD(rdd => if (!rdd.isEmpty) globalRDD1 = globalRDD1.union(rdd)) 
dstream2.foreachRDD(rdd => if (!rdd.isEmpty) { 
    globalRDD2 = globalRDD2.union(rdd)) 
    globalRDD1.join(globalRDD2).foreach(...) // etc, etc 
}