Spark Streaming: foreachRDD aktualizuje moje mongo RDD

Question

Chcę utworzyć nowy mongodb RDD za każdym razem gdy wchodzę do środka foreachRDD. Jednak mam problemy serializacji:

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) }) 

to daje mi błąd:

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

jakiś pomysł?

Answer 1

Można spróbować użyć rdd.context zwracającą albo SparkContext lub SparkStreamingContext (jeśli RDD jest DStream).

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

Faktycznie, wydaje się, że RDD posiada również metodę .sparkContext. Szczerze mówiąc, nie znam różnicy, może są to aliasy (?).

Answer 2

W moim rozumieniu musisz dodać, jeśli masz obiekt "nie serializowalny", musisz go przekazać przez foreachPartition, abyś mógł nawiązać połączenie z bazą danych w każdym węźle przed uruchomieniem przetwarzania.

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) }})