Spark from_avro函数不支持列作为Schema的问题及多Schema反序列化诉求

Spark自带的from_avro函数有个限制：它只接受字符串类型的Schema参数，函数定义如下：

def from_avro(col: Column, jsonFormatSchema: String): Column 

这就带来了问题——如果你的DataFrame里，每行的Avro记录对应不同的Schema，原生函数根本没法处理，因为只能传入一个固定的Schema字符串。我们其实更希望它能支持传入DataFrame的列作为Schema，比如这样的定义：

def from_avro(col: Column, jsonFormatSchema: Column): Column  

下面提供两种可行的解决思路：

方案一：自定义UDF处理动态Schema #

既然原生函数不支持列作为Schema参数，我们可以自己实现一个UDF，在UDF内部根据每行的Schema字符串反序列化Avro二进制数据。

代码示例 #

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.types._
import org.apache.avro.Schema
import org.apache.avro.generic.GenericDatumReader
import org.apache.avro.io.DecoderFactory
import java.io.ByteArrayInputStream
import scala.collection.JavaConverters._

object DynamicAvroParser {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("DynamicAvroParser")
      .master("local[*]")
      .getOrCreate()
    import spark.implicits._

    // 定义两个不同的Avro Schema（这里模拟实际业务中不同结构的Schema）
    val avroSchema1 = """{"type":"record","name":"myrecord","fields":[{"name":"str1","type":"string"},{"name":"num1","type":"double"}]}""" 
    val avroSchema2 = """{"type":"record","name":"myrecord","fields":[{"name":"str1","type":"string"},{"name":"bool1","type":"boolean"}]}"""

    // 构造测试DataFrame：每行包含Avro二进制数据和对应的Schema字符串
    val df = Seq(
      // 这里的二进制数据是对应schema1的序列化结果（str1=apple1, num1=1.0）
      (Array[Byte](10, 97, 112, 112, 108, 101, 49, 0, 64, 0, 0, 0, 0, 0, 0, 0), avroSchema1),
      // 这里的二进制数据是对应schema2的序列化结果（str1=apple2, bool1=true）
      (Array[Byte](10, 97, 112, 112, 108, 101, 50, 0, 1), avroSchema2)
    ).toDF("binaryData", "schema")

    // 实现自定义UDF：接收二进制数据和Schema字符串，返回反序列化后的Map
    val parseAvroDynamic = udf((binary: Array[Byte], schemaStr: String) => {
      val schema = new Schema.Parser().parse(schemaStr)
      val reader = new GenericDatumReader[Any](schema)
      val decoder = DecoderFactory.get().binaryDecoder(new ByteArrayInputStream(binary), null)
      val record = reader.read(null, decoder).asInstanceOf[org.apache.avro.generic.GenericRecord]
      // 将GenericRecord转为Map，方便后续展开字段
      record.getSchema.getFields.asScala
        .map(field => field.name() -> record.get(field.pos()))
        .toMap
    })

    // 使用UDF解析数据
    val parsedDF = df.select(parseAvroDynamic($"binaryData", $"schema").as("parsedData"))
    parsedDF.show(false)
  }
}

关键说明 #

• 可以根据需求调整UDF的返回类型：如果需要固定的StructType，可以提前定义StructType，再把Map转换成对应的Row对象，后续就能直接用.访问字段。
• 若Schema重复率高，建议在UDF内部缓存已解析的Schema对象，避免重复解析字符串，提升性能。

方案二：分组解析后合并 #

如果DataFrame中不同Schema的分组比较明确，可以按Schema字符串分组，每组用对应的Schema调用原生from_avro解析，最后合并结果。

代码示例 #

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.from_avro

object GroupedAvroParser {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("GroupedAvroParser")
      .master("local[*]")
      .getOrCreate()
    import spark.implicits._

    val avroSchema1 = """{"type":"record","name":"myrecord","fields":[{"name":"str1","type":"string"},{"name":"num1","type":"double"}]}""" 
    val avroSchema2 = """{"type":"record","name":"myrecord","fields":[{"name":"str1","type":"string"},{"name":"bool1","type":"boolean"}]}"""

    val df = Seq(
      (Array[Byte](10, 97, 112, 112, 108, 101, 49, 0, 64, 0, 0, 0, 0, 0, 0, 0), avroSchema1),
      (Array[Byte](10, 97, 112, 112, 108, 101, 50, 0, 1), avroSchema2)
    ).toDF("binaryData", "schema")

    // 获取所有唯一的Schema字符串
    val uniqueSchemas = df.select($"schema").distinct().as[String].collect()

    // 对每个Schema分组解析
    val parsedDfs = uniqueSchemas.map { schema =>
      df.filter($"schema" === schema)
        .select(from_avro($"binaryData", schema).as("parsedData"), $"schema")
    }

    // 合并所有解析后的DataFrame
    val finalDf = parsedDfs.reduce(_ unionByName _)
    finalDf.show(false)
  }
}

适用场景 #

这种方法适合Schema数量较少的情况，能利用原生from_avro的性能优势；但如果Schema数量极多，会生成大量小DataFrame，反而影响性能，此时更适合用方案一。

内容的提问来源于stack exchange，提问作者Philip K. Adetiloye