无法使用火花结构化流反序列化avro消息，其中键已字符串化，值是avro

Question

使用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。

Answer 1

您在地图块外为KafkaAvroDeserializer定义了变量（'deser'）。它使该异常。

尝试像这样更改代码：