我想在每次进入 foreachRDD 时创建一个new 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)

有什么想法吗？

最佳答案

您可以尝试使用 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方法。老实说，我不知道有什么区别，也许它们是别名(？)。

关于mongodb - Spark 流 : foreachRDD update my mongo RDD，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/34814615/