スパークストリーミング：foreachRDD私のmongo RDDを更新

Question

foreachRDDに入力するたびに mongodb RDDを作成します。しかし私は、シリアル化の問題があります。

mydstream 
    .foreachRDD(rdd => { 
     val mongoClient = MongoClient("localhost", 27017) 
     val db = mongoClient(mongoDatabase) 
     val coll = db(mongoCollection) 
     // ssc is my StreamingContext 
     val modelsRDDRaw = ssc.sparkContext.parallelize(coll.find().toList) }) 

これは私にエラーを与える：

object not serializable (class: org.apache.spark.streaming.StreamingContext, value: org.apache.spark.streaming.StreamingContext@31133b6e) 

任意のアイデア？

Answer 1

あなたはSparkContextまたはSparkStreamingContext（RDDがDSTREAMの場合）のいずれかを返すrdd.contextを使用しようとするかもしれません。

mydstream foreachRDD { rdd => { 
     val mongoClient = MongoClient("localhost", 27017) 
     val db = mongoClient(mongoDatabase) 
     val coll = db(mongoCollection) 
     val modelsRDDRaw = rdd.context.parallelize(coll.find().toList) }) 

実際には、RDDにも.sparkContextメソッドがあるようです。私は正直なところ違いを知らないのでしょうか、別名かもしれません（？）。

Answer 2

「直列化できない」オブジェクトがある場合は、foreachPartitionに渡す必要がありますので、処理を実行する前に各ノードのデータベースに接続する必要があります。

mydstream.foreachRDD(rdd => { 
     rdd.foreachPartition{ 
      val mongoClient = MongoClient("localhost", 27017) 
      val db = mongoClient(mongoDatabase) 
      val coll = db(mongoCollection) 
      // ssc is my StreamingContext 
      val modelsRDDRaw = ssc.sparkContext.parallelize(coll.find().toList) }})