Spark：使用 Spark Scala 从 Kafka 读取 Avro 消息

Posted 2023-04-18

【中文标题】Spark：使用 Spark Scala 从 Kafka 读取 Avro 消息

【英文标题】：Spark: Reading Avro messages from Kafka using Spark Scala

【发布时间】：2019-11-26 23:54:10

【问题描述】：

我正在尝试spark 2.4.3 中的以下代码来读取来自 kafka 的 Avro 消息。

当数据在 kafka 上发布时，Schema 存储在 confluent schema registry 中。 我一直在尝试一些已经在这里讨论过的解决方案 (Integrating Spark Structured Streaming with the Confluent Schema Registry / Reading Avro messages from Kafka with Spark 2.0.2 (structured streaming)) 但无法使其发挥作用。 或者我找不到正确的方法来执行此操作，尤其是当架构存储在某些 Schema Registry 中时。

这是我正在尝试的当前代码，至少我能够得到一些结果 但所有记录都以null 值出现。其实这个话题有数据。 有人可以帮我解决这个问题吗？

import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient, SchemaRegistryClient
import io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer
import org.apache.avro.Schema
import org.apache.avro.generic.GenericRecord
import org.apache.spark.sql.avro.SchemaConverters

object ScalaSparkAvroConsumer 

    private val topic = "customer.v1"
    private val kafkaUrl = "localhost:9092"
    private val schemaRegistryUrl = "http://127.0.0.1:8081"

    private val schemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistryUrl, 128)
    private val kafkaAvroDeserializer = new AvroDeserializer(schemaRegistryClient)

    private val avroSchema = schemaRegistryClient.getLatestSchemaMetadata(topic + "-value").getSchema
    private var sparkSchema = SchemaConverters.toSqlType(new Schema.Parser().parse(avroSchema))

    def main(args: Array[String]): Unit = 
      val spark = getSparkSession()

      spark.sparkContext.setLogLevel("ERROR")

      spark.udf.register("deserialize", (bytes: Array[Byte]) =>
        DeserializerWrapper.deserializer.deserialize(bytes)
      )

      val df = spark
        .readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", kafkaUrl)
        .option("subscribe", topic)
        .option("startingOffsets", "earliest")
        .load()

      val valueDataFrame = df.selectExpr("""deserialize(value) AS message""")

      import org.apache.spark.sql.functions._

      val formattedDataFrame = valueDataFrame.select(
        from_json(col("message"), sparkSchema.dataType).alias("parsed_value"))
        .select("parsed_value.*")

      formattedDataFrame
        .writeStream
        .format("console")
        .option("truncate", false)
        .start()
        .awaitTermination()
    

    object DeserializerWrapper 
      val deserializer = kafkaAvroDeserializer
    

    class AvroDeserializer extends AbstractKafkaAvroDeserializer 
      def this(client: SchemaRegistryClient) 
        this()
        this.schemaRegistry = client
      

      override def deserialize(bytes: Array[Byte]): String = 
        val genericRecord = super.deserialize(bytes).asInstanceOf[GenericRecord]
        genericRecord.toString
      
    

得到如下输出：

-------------------------------------------
Batch: 0
-------------------------------------------
+------+-------+
|header|control|
+------+-------+
|null  |null   |
|null  |null   |
|null  |null   |
|null  |null   |
+------+-------+
only showing top 20 rows        

【问题讨论】：

Integrating Spark Structured Streaming with the Confluent Schema Registry的可能重复

我已经按照描述中的说明尝试了这些，但无法使其工作。你能给我建议吗？

我在那里写了答案，并且可以验证它对我有用。如果您得到 null，则生成的架构可能与记录内容不一致。那里的答案没有使用.asInstanceOf[GenericRecord]，例如

你能检查一下valueDataFrame里面有什么吗？可以valueDataFrame.writeStream.format("console")吗？为了更容易调试，使用read（Spark SQL）而不是readStream（结构化流），直到它为您提供正确的值。

是的，我使用了read，它给了我这样的实际信息。 "header": "Id": "123","control": "subject": "EOD"

【参考方案1】：

Avro 序列化、Kafka 模式服务器和 Spark Streaming 与 from_confluence_avro() 的集成将使您的生活更轻松。你可以在这里找到它：

https://github.com/AbsaOSS/ABRiS