使用Spark 2.4.0
汇合架构-注册以接收架构
消息Key在String中序列化,在Avro中Value序列化,因此我试图使用io.confluent.kafka.serializers.KafkaAvroDeserializer仅反序列化Value,但是它不起作用。任何人都可以查看我的代码以查看有什么问题
导入的库:
import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient
import io.confluent.kafka.serializers.KafkaAvroDeserializer
import org.apache.avro.generic.GenericRecord
import org.apache.kafka.common.serialization.Deserializer
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{ Encoder, SparkSession}
代码正文
val topics = "test_topic"
val spark: SparkSession = SparkSession.builder
.config("spark.streaming.stopGracefullyOnShutdown", "true")
.config("spark.streaming.backpressure.enabled", "true")
.config("spark.streaming.kafka.maxRatePerPartition", 2170)
.config("spark.streaming.kafka.maxRetries", 1)
.config("spark.streaming.kafka.consumer.poll.ms", "600000")
.appName("SparkStructuredStreamAvro")
.config("spark.sql.streaming.checkpointLocation", "/tmp/new_checkpoint/")
.enableHiveSupport()
.getOrCreate
//add settings for schema registry url, used to get deser
val schemaRegUrl = "http://xx.xx.xx.xxx:xxxx"
val client = new CachedSchemaRegistryClient(schemaRegUrl, 100)
//subscribe to kafka
val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "xx.xx.xxxx")
.option("subscribe", "test.topic")
.option("kafka.startingOffsets", "latest")
.option("group.id", "use_a_separate_group_id_for_each_stream")
.load()
//add confluent kafka avro deserializer, needed to read messages appropriately
val deser = new KafkaAvroDeserializer(client).asInstanceOf[Deserializer[GenericRecord]]
//needed to convert column select into Array[Bytes]
import spark.implicits._
val results = df.select(col("value").as[Array[Byte]]).map { rawBytes: Array[Byte] =>
//read the raw bytes from spark and then use the confluent deserializer to get the record back
val decoded = deser.deserialize(topics, rawBytes)
val recordId = decoded.get("nameId").asInstanceOf[org.apache.avro.util.Utf8].toString
recordId
}
results.writeStream
.outputMode("append")
.format("text")
.option("path", "/tmp/path_new/")
.option("truncate", "false")
.start()
.awaitTermination()
spark.stop()
无法反序列化,并且收到错误是
Caused by: java.io.NotSerializableException: io.confluent.kafka.serializers.KafkaAvroDeserializer
Serialization stack:
- object not serializable (class: io.confluent.kafka.serializers.KafkaAvroDeserializer, value: io.confluent.kafka.serializers.KafkaAvroDeserializer@591024db)
- field (class: ca.bell.wireless.ingest$$anonfun$1, name: deser$1, type: interface org.apache.kafka.common.serialization.Deserializer)
- object (class ca.bell.wireless.ingest$$anonfun$1, <function1>)
- element of array (index: 1)
[当我使用以下方法编写普通的kafka使用者(不是通过spark时),效果很好]
props.put("key.deserializer", classOf[StringDeserializer])
props.put("value.deserializer", classOf[KafkaAvroDeserializer])
使用Spark 2.4.0 Confluent schema-Registry接收模式消息Key在Avro中的String和Value中被序列化,因此我试图使用io.confluent.kafka来仅反序列化Value。
您在地图块外为KafkaAvroDeserializer定义了变量('deser')。它使该异常。
尝试像这样更改代码: