数据流检查点已启用，但数据流及其功能不可序列化

Question

我想把DStream发送到Kafka，但它仍然不起作用。

searchWordCountsDStream.foreachRDD(rdd =>
rdd.foreachPartition(

    partitionOfRecords =>
    {
      val props = new HashMap[String, Object]()

      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, outbroker)

      props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer")
      props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer")
      val producer = new KafkaProducer[String,String](props)

      partitionOfRecords.foreach
      {
        case (x:String,y:String)=>{
          println(x)
          val message=new ProducerRecord[String, String](outtopic,null,x)
          producer.send(message)
        }
      }
      producer.close()
    })
)

以下是一些错误信息：

16/10/31 14:44:15错误StreamingContext：错误启动上下文，将其标记为停止java.io.NotSerializableException：DStream检查点已启用，但DStreams及其功能不可序列化spider.app.job.MeetMonitor序列化堆栈：-对象不可序列化（类：spider.app.job.MeetMonitor，值：spider.app.job.MeetMonitor@433c6abb）-字段（类：spider.app.job.MeetMonitor$$anonfun$createContext2美元，名称：$外，类型：类spider.app.job.MeetMonitor）-对象（类spider.app.job.MeetMonitor$anonfun$createContext2美元，）-字段（类：org.apache.spark.streaming.dstream.DStream$anonfun$foreachRDD1美元$anonfun$应用$mcV$sp3美元，名称：清洁F1美元，类型：接口scala。函数1）-对象（类org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD1美元$$anonfun$应用$mcV$sp3美元，）-写对象数据（类：org.apache.spark.streaming.dstream.DStream）-对象（类org.apache.spark.streaming.dstream.ForEachDStream，org.apache.spark.streaming.dstream.ForEachDStream@3ac3f6f）-写对象数据（类：org.apache.spark.streaming.dstream.DStreamCheckpoint Data）-对象（类org.apache.spark.streaming.dstream.DStreamCheckpoint Data，[0检查点文件

])-writeObject数据（类：org.apache.spark.streaming.dstream.dstream）-对象（类：org.apache.spark.streaming.dstream.ForEachDStream，org.apache.spark.streaming.dstream）。ForEachDStream@6f9c5048)-数组元素（索引：0）-数组（类[Ljava.lang.Object；，大小16）-字段（类：scala.collection.mutable.ArrayBuffer，名称：数组，类型：类[Ljava.lang.Object；）-对象（类scala.collection.mutable.ArrayBuffer，ArrayBuffer）（org.apache.spark.streaming.dstream）。ForEachDStream@6f9c5048，org。阿帕奇。火花流动。数据流。ForEachDStream@3ac3f6f))-writeObject数据（类：org.apache.spark.streaming.dstream.dstream checkpointdata）-对象（类org.apache.spark.stream.dstream.dstream checkpointdata，[0个检查点文件

])

Answer 1

我一直在使用Spark 2.3.0版本，遇到了同样的问题，我只是通过为它抛出错误的类实现Serializable接口解决了这个问题。

你的情况是蜘蛛。应用程序。工作MeetMonitor应该像这样实现它：。

公共类MeetMonitor实现可序列化{

// ... }

另一件事是，如果您在类中使用Logger，请注意它的实例也不可序列化，因此可能会导致相同的问题。这也可以通过将其定义为：

私有静态最终记录器记录器=Logger.get记录器（. class）；

Answer 2

我遇到了同样的问题在这里找到了答案

https://forums.databricks.com/questions/382/why-is-my-spark-streaming-application-throwing-a-n.html

似乎将检查点与foreachRDD一起使用会导致问题。删除代码中的检查点后，一切正常。

P/S.我只是想发表评论，但我没有足够的声誉这么做。