DStreams

Question

我试图访问像在解决这个问题上取得过滤DStreams集合的访问收集：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.NotSerializab leException：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

如何访问数据流写入到卡夫卡在一个序列化的方式？

Answer 1

作为异常指示，所述DStream定义正在被闭合捕获。 一个简单的选择是声明此DStream瞬态：

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

@transient标志某些对象，以从某个对象的对象图的Java序列被移除。此方案的关键是，在相同的范围内（在这种情况下statusCodeStreams）的DStream宣布一些val被封闭内使用。该val从封闭中的实际参考是outer.statusCodeStreams，引起序列化过程中“拉”的outer所有上下文到关闭。使用@transient我们将DStream（以及StreamingContext）声明标记为不可序列化，并且我们避免了序列化问题。根据不同的代码结构（如果它在一个main功能（不好的做法，所有的线性，顺便说一句），它可能是必要的，以纪念ALL DSTREAM声明+中的StreamingContext实例作为@transient。

如果初始过滤的唯一目的是“路线”的内容分开卡夫卡的话题，它可能是值得移动foreachRDD内的过滤。这将使一个简单的程序结构。

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