Apache SparkでRAMにデータを保存できますか？

Question

例えば、Apache Sparkを使用してRAMに文字列を格納することが可能かどうかを知りたいと思います。実際、Apache Sparkが扱っている新しい入力データに応じて、これらの文字列を照会して更新したいと考えています。さらに可能であれば、ノードは他のすべてのノードにどの文字列が格納されているかを通知できますか？ プロジェクターに関する情報が必要な場合は、お気軽にお問い合わせください。

J

Answer 1

はい、あなたは、ステートフルなストリーミング機能mapWithStateが必要です。この機能を使用すると、ストリーミングバッチ全体でメモリにキャッシュされた状態を更新できます。

まだチェックポイントを設定していない場合は、チェックポイントを有効にする必要があります。

Scalaの使用例：

def stateUpdateFunction(userId: UserId, 
         newData: UserAction, 
         stateData: State[UserSession]): UserModel = { 
    val currentSession = stateData.get() // Get current session data 
    val updatedSession = ...   // Compute updated session using newData 
    stateData.update(updatedSession)   // Update session data  
    val userModel = ...     // Compute model using updatedSession 
    return userModel    // Send model downstream 
} 

// Stream of user actions, keyed by the user ID 
val userActions = ... // stream of key-value tuples of (UserId, UserAction) 
// Stream of data to commit 
val userModels = userActions.mapWithState(StateSpec.function(stateUpdateFunction)) 

https://databricks.com/blog/2016/02/01/faster-stateful-stream-processing-in-apache-spark-streaming.html

のJavaの使用例：

// Update the cumulative count function 
Function3<String, Optional<Integer>, State<Integer>, Tuple2<String, Integer>> mappingFunc = 
    new Function3<String, Optional<Integer>, State<Integer>, Tuple2<String, Integer>>() { 
     @Override 
     public Tuple2<String, Integer> call(String word, Optional<Integer> one, 
      State<Integer> state) { 
     int sum = one.orElse(0) + (state.exists() ? state.get() : 0); 
     Tuple2<String, Integer> output = new Tuple2<>(word, sum); 
     state.update(sum); 
     return output; 
     } 
    }; 

// DStream made of get cumulative counts that get updated in every batch 
JavaMapWithStateDStream<String, Integer, Integer, Tuple2<String, Integer>> stateDstream = 
    wordsDstream.mapWithState(StateSpec.function(mappingFunc).initialState(initialRDD)); 

https://github.com/apache/spark/blob/master/examples/src/main/java/org/apache/spark/examples/streaming/JavaStatefulNetworkWordCount.java行90：