Spark使用to_avro序列化DataFrame至含缺失列的Avro Schema方法

问题背景 #

现有Spark DataFrame的Schema如下：

StructType(
    StructField(id,StringType,true),
    StructField(type,StringType,true),
)

需要通过spark-avro的to_avro(spark_df, jsonFormatSchema)方法，将其序列化为指定Avro Schema格式，目标Avro Schema为：

{
  "type": "record",
  "name": "Value",
  "fields": [
    {
      "name": "id",
      "type": "string"
    },
    {
      "name": "type",
      "type": "string"
    },
    {
      "name": "x",
      "type": [
        "null",
        "string"
      ],
      "default": null
    },
    {
      "name": "y",
      "type": [
        {
          "type": "boolean",
          "connect.default": false
        },
        "null"
      ],
      "default": false
    }
  ]
}

由于DataFrame缺少x和y字段，直接使用上述Avro Schema会抛出IncompatibleSchemaException，需要调整配置或处理逻辑，让序列化后的Avro二进制自动填充这两个字段的默认值。

解决方案 #

方法一：补全DataFrame缺失字段（推荐） #

在序列化前，给DataFrame添加缺失的x、y列并设置对应默认值，让DataFrame Schema与Avro Schema完全匹配：

import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.types.{StringType, BooleanType}

// 补全x（默认null）和y（默认false）字段
val updatedDf = spark_df
  .withColumn("x", lit(null).cast(StringType))
  .withColumn("y", lit(false).cast(BooleanType))

之后直接使用原Avro Schema调用to_avro即可：

import org.apache.spark.sql.avro.functions.to_avro
import org.apache.spark.sql.functions.col

val avroBinaryCol = to_avro(col("*"), jsonFormatSchema)
val resultDf = updatedDf.select(avroBinaryCol as "avro_binary")

方法二：开启自动填充默认值配置（Spark 3.1+） #

Spark 3.1及以上版本提供了spark.sql.avro.write.setDefaultForMissingField配置，开启后，序列化时会自动为DataFrame中缺失的、带有默认值的Avro字段填充默认值，无需修改DataFrame：

// 开启自动填充缺失字段默认值的配置
spark.conf.set("spark.sql.avro.write.setDefaultForMissingField", "true")

// 直接使用原Avro Schema序列化
val avroBinaryCol = to_avro(col("*"), jsonFormatSchema)
val resultDf = spark_df.select(avroBinaryCol as "avro_binary")

问题原因说明 #

Avro Schema中的null类型仅表示字段值允许为null，并不允许字段本身不存在。spark-avro默认要求DataFrame字段与Avro Schema字段完全匹配，因此缺失字段会触发兼容性异常。只有通过补全字段或开启上述配置，才能让序列化器自动填充缺失字段的默认值。

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