स्पार्क स्ट्रीमिंग: foreachRDD अपडेट करें मेरे मोंगो आरडीडी

Question

मैं नया mongodb RDD प्रत्येक बार जब मैं foreachRDD के अंदर प्रवेश करता हूं तो अपडेट करना चाहता हूं।

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

किसी भी विचार:

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

यह मैं एक त्रुटि दे देंगे: हालांकि मैं क्रमबद्धता मुद्दे हैं?

Answer 1

आप rdd.context है कि या तो एक SparkContext या एक SparkStreamingContext (यदि RDD एक 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) }) 

वास्तव में, ऐसा लगता है कि RDD भी एक .sparkContext विधि है। मैं ईमानदारी से अंतर नहीं जानता, शायद वे उपनाम हैं?()।

Answer 2

मेरी समझ में आपको जोड़ना होगा कि क्या आपके पास "serializable" ऑब्जेक्ट नहीं है, तो आपको 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) }})