scala – spark kafka生产者可序列化
|
我想出了一个例外：
// @transient
val sparkConf = new SparkConf()
sparkConf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
// @transient
val sc = new SparkContext(sparkConf)
val requestSet: RDD[String] = sc.textFile(s"hdfs:/user/bigdata/ADVERTISE-IMPRESSION-STAT*/*")
// @transient
val props = new HashMap[String,Object]()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,NearLineConfig.kafka_brokers)
// props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.ByteArraySerializer");
// props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.ByteArraySerializer");
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")
props.put("producer.type","async")
props.put(ProducerConfig.BATCH_SIZE_CONFIG,"49152")
// @transient
val producer: KafkaProducer[String,String] = new KafkaProducer[String,String](props)
requestSet.foreachPartition((partisions: Iterator[String]) => {
partisions.foreach((line: String) => {
try {
producer.send(new ProducerRecord[String,String]("testtopic",line))
} catch {
case ex: Exception => {
log.warn(ex.getMessage,ex)
}
}
})
})
producer.close()
在这个程序中,我尝试从hdfs路径读取记录并将它们保存到kafka中. 解决方法
KafkaProducer不可序列化.您需要将实例的创建移动到foreachPartition中：
requestSet.foreachPartition((partitions: Iterator[String]) => {
val producer: KafkaProducer[String,String](props)
partitions.foreach((line: String) => {
try {
producer.send(new ProducerRecord[String,ex)
}
}
})
})
请注意,KafkaProducer.send返回Future [RecordMetadata],如果无法序列??化键或值,则唯一可以传播的异常是SerializationException. （编辑：李大同） 【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容! |