访问 DStream 集合

Question

我正在尝试访问在此问题的解决方案中获得的过滤 DStream 集合：Spark Streaming - Best way to Split Input Stream based on filter Param

我按如下方式创建集合：

val statuCodes = Set("200","500", "404")
    spanTagStream.cache()
    val statusCodeStreams = statuCodes.map(key => key -> spanTagStream.filter(x => x._3.get("http.status_code").getOrElse("").asInstanceOf[String].equals(key)))

我尝试通过以下方式访问statusCodeStreams：

for(streamTuple <- statusCodeStreams){
      streamTuple._2.foreachRDD(rdd =>
  rdd.foreachPartition(
      partitionOfRecords =>
        {
            val props = new HashMap[String, Object]()
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers)
            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
            {
                 x=>{ 
                 /* Code Writing to Kafka using streamTuple._1 as the topic-String */
                 }
            }
      })
   )
}

执行此操作时，我收到以下错误： java.io.NotSerializableException: Object of org.apache.spark.streaming.kafka010.DirectKafkaInputDStream is being serialized possibly as a part of closure of an RDD operation. This is because the DStream object is being referred to from within the closure. Please rewrite the RDD operation inside this DStream to avoid this. This has been enforced to avoid bloating of Spark tasks with unnecessary objects

如何访问 Streams 以以可序列化的方式写入 Kafka？

Answer 1

正如异常所示，DStream 定义正在被闭包捕获。 一个简单的选择是声明这个DStream 瞬态：

@transient val spamTagStream = //KafkaUtils.create...

@transient 标记要从某个对象的对象图的 Java 序列化中删除的某些对象。这种情况的关键是在闭包中使用了一些在与DStream（本例中为statusCodeStreams）相同范围内声明的val。闭包中val 的实际引用是outer.statusCodeStreams，导致序列化过程将outer 的所有上下文“拉”到闭包中。使用@transient，我们将 DStream（以及 StreamingContext）声明标记为不可序列化，并且避免了序列化问题。根据代码结构（如果它在一个 main 函数中都是线性的（不好的做法，顺便说一句），则可能需要将 ALL DStream 声明 + StreamingContext 实例标记为 @transient。

如果初始过滤的唯一目的是将内容“路由”到单独的 Kafka 主题，则可能值得在 foreachRDD 中移动过滤。这将使程序结构更简单。

spamTagStream.foreachRDD{ rdd => 
    rdd.cache()
    statuCodes.map{code =>
        val matchingCodes = rdd.filter(...)
        matchingCodes.foreachPartition{write to kafka}
    }
    rdd.unpersist(true)
}