15

I want to create a new mongodb RDD each time I enter inside foreachRDD. However I have serialization issues:

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

This will give me an error:

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

Any idea?

2
  • SparkContext is not serializable, so you cannot use inside any transformation or action methods, you have to use only in driver class.
    – Shankar
    Commented Jan 18, 2016 at 11:37
  • are there any specific reason why you are converting list to rdd inside foreachRDD method?
    – Shankar
    Commented Jan 18, 2016 at 11:45

2 Answers 2

7
+50

You might try to use rdd.context that returns either a SparkContext or a SparkStreamingContext (if rdd is a 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) })

Actually, it seems that RDD has also a .sparkContext method. I honestly don't know the difference, maybe they are aliases (?).

2

In my understanding you have to add if you have a "not serializable" object, you need to pass it through foreachPartition so you can make a connection to database on each node before running your processing.

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) }})
1
  • You can try create your ssc inside the foreachRDD before rdd.foreachPartition val ssc = StreamingContext.getOrCreate(checkpointdirectory, functionToCreateContext _)
    – Rami
    Commented Jan 18, 2016 at 9:14

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.