Spark Scala Kafka Avro反序列化：将JSON转为Struct类型（流场景）

• 从Kafka消费流DataFrame，value列是Avro格式，需要反序列化为Struct类型，支持value.name这类直接访问语法
• 当前写的UDF能返回JSON字符串，但转不成Struct；而且没有对应样例类、JSON示例或Schema文件，代码得适配不同结构的Topic
• 之前参考的方案要么方法过时，要么会丢key/topic这类列，还得用foreachBatch才能处理流数据，想要基于UDF的解决方案，顺便确认下当前思路或者反序列化方式有没有问题

核心思路 #

1. 靠GenericRecord自带的Schema信息动态生成Spark StructType，不用预定义样例类或者Schema文件
2. 修改UDF逻辑，把GenericRecord转换成Spark的Row对象，同时把Schema转成StructType，让Spark识别成结构化数据
3. 全程保留原DataFrame的key、topic等列，不用额外的转换逻辑

修正后的UDF代码 #

import org.apache.avro.generic.GenericRecord
import org.apache.kafka.common.serialization.Deserializer
import org.apache.spark.sql.Row
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.types.{StructType, StructField, DataType}
import org.apache.avro.Schema

// 初始化Avro反序列化器
val kafkaAvroDeserializer: Deserializer[GenericRecord] = new KafkaAvroDeserializer()
private val kafkaAvroDeserializerConfig = Map(
  "schema.registry.url" -> "你的Schema Registry地址"
).asJava
kafkaAvroDeserializer.configure(kafkaAvroDeserializerConfig, false)

// Avro Schema转Spark StructType的工具方法
def avroSchemaToSparkStructType(avroSchema: Schema): StructType = {
  val fields = avroSchema.getFields.map { field =>
    val sparkDataType = field.schema().getType match {
      case Schema.Type.STRING => org.apache.spark.sql.types.StringType
      case Schema.Type.INT => org.apache.spark.sql.types.IntegerType
      case Schema.Type.LONG => org.apache.spark.sql.types.LongType
      case Schema.Type.BOOLEAN => org.apache.spark.sql.types.BooleanType
      case Schema.Type.FLOAT => org.apache.spark.sql.types.FloatType
      case Schema.Type.DOUBLE => org.apache.spark.sql.types.DoubleType
      case Schema.Type.RECORD => avroSchemaToSparkStructType(field.schema()) // 处理嵌套结构
      // 按需扩展其他类型，比如数组、枚举等
      case _ => org.apache.spark.sql.types.StringType // 不支持的类型默认转成String
    }
    StructField(field.name(), sparkDataType, field.schema().isNullable)
  }
  StructType(fields)
}

// GenericRecord转Spark Row的工具方法
def genericRecordToRow(record: GenericRecord): Row = {
  val values = record.getSchema.getFields.map(field => record.get(field.name()))
  Row.fromSeq(values)
}

// 生成带StructType的UDF，支持按Topic动态获取Schema
def getAvroDeserializeUDF(topic: String) = {
  // 先拿空字节获取Schema（Schema Registry会返回对应Topic的最新Schema）
  val schema = kafkaAvroDeserializer.deserialize(topic, Array.empty[Byte]).getSchema
  udf((bytes: Array[Byte]) => {
    val genericRecord = kafkaAvroDeserializer.deserialize(topic, bytes)
    if (genericRecord == null) null else genericRecordToRow(genericRecord)
  }, avroSchemaToSparkStructType(schema))
}

使用示例 #

// 假设streamingDF是从Kafka消费的原始流，包含key、value、topic等列
// 方式1：替换原value列为结构化数据，直接用value.name访问
val processedDF = streamingDF
  .withColumn("value", getAvroDeserializeUDF(col("topic").asInstanceOf[String])(col("value")))

// 方式2：保留原value列，新增结构化列并展开
val processedDF = streamingDF
  .withColumn("deserialized_value", getAvroDeserializeUDF(col("topic").asInstanceOf[String])(col("value")))
  .select("key", "topic", "deserialized_value.*")

注意点 #

• 多Topic适配：通过topic参数动态获取对应Schema，确保不同结构的Topic都能正确反序列化
• 类型扩展：可以根据业务需求，在avroSchemaToSparkStructType里补充支持数组、映射、枚举等更多Avro类型
• 空值处理：反序列化时判断bytes是否为空，返回null避免报错
• 性能优化：可以加个Map缓存topic到StructType的映射，不用每次都去Schema Registry拉取Schema

内容的提问来源于stack exchange，提问作者Mohammad Shahwez