Spark DataFrame嵌套列null值填充：na.fill无效的解决方法咨询

一、嵌套列Null值填充方案 #

DataFrameNaFunctions.fill仅支持顶层列的Null填充，嵌套列需要通过逐层重构结构或自定义UDF处理。

先修正原始代码中的错误 #

原始代码存在case class名称不匹配、未定义Data类的问题，修正后如下：

import spark.implicits._

case class Demographics(city: String)
case class Detail(age: Int, demographics: Demographics)
case class Person(name: String, details: Detail)
case class Data(person: Person)

val data = Seq(
  Data(Person("James", Detail(48, Demographics("Toronto")))),
  Data(Person("Mary", Detail(41, Demographics(null)))),
  Data(null)
).toDS

方案1：逐层重构嵌套结构 #

通过struct函数重新构造嵌套字段，配合when/otherwise填充Null：

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

val filledData = data
  .withColumn("person", 
    when(col("person").isNull, lit(null))
    .otherwise(
      struct(
        col("person.name"),
        struct(
          col("person.details.age"),
          struct(
            when(col("person.details.demographics.city").isNull, lit("default"))
            .otherwise(col("person.details.demographics.city"))
            .alias("city")
          ).alias("demographics")
        ).alias("details")
      )
    )
  )

filledData.show(false)

执行后，Mary的demographics.city会被替换为default，同时保留person为Null的原始数据。

方案2：自定义UDF处理复杂嵌套 #

如果嵌套层级多，用UDF递归处理更简洁：

import org.apache.spark.sql.functions.udf

// 处理Demographics层的Null填充
val fillDemoUdf = udf((demo: Demographics) => {
  if (demo == null) Demographics("default")
  else Demographics(Option(demo.city).getOrElse("default"))
})

// 处理Detail层
val fillDetailUdf = udf((detail: Detail) => {
  if (detail == null) null
  else Detail(detail.age, fillDemoUdf(detail.demographics))
})

// 处理Person层
val fillPersonUdf = udf((person: Person) => {
  if (person == null) null
  else Person(person.name, fillDetailUdf(detail = person.details))
})

val filledData = data.withColumn("person", fillPersonUdf(col("person")))
filledData.show(false)

二、Java Bean转换报错解决 #

报错原因是Scala case class使用了不可空的基本类型（如Int），而Spark无法将Null映射到基本类型，同时Java Bean若使用基本类型也会触发错误。

解决方案1：将case class类型改为可空包装类型 #

把Scala的Int替换为java.lang.Integer，确保支持Null值，同时Java Bean对应字段也使用包装类型（如Integer age而非int age）：

// 修改后的case class
case class Demographics(city: String)
case class Detail(age: java.lang.Integer, demographics: Demographics)
case class Person(name: String, details: Detail)
case class Data(person: Person)

解决方案2：彻底消除数据中的Null值 #

在转换为Java Bean前，确保所有可能为Null的字段都填充默认值，避免Null进入转换流程：

import org.apache.spark.sql.types.IntegerType

val fullyFilledData = filledData
  .withColumn("person", 
    when(col("person").isNull, 
      struct(
        lit("default_name").alias("name"),
        struct(
          lit(0).cast(IntegerType).alias("age"),
          struct(lit("default").alias("city")).alias("demographics")
        ).alias("details")
      )
    ).otherwise(col("person"))
  )

// 执行Java Bean转换
val encoder = Encoders.bean(classOf[InputBeanClass])
fullyFilledData.map(row => row)(encoder).count()

额外注意 #

确保Java BeanInputBeanClass的结构、字段名称与Scala case class完全一致，嵌套层级也必须一一对应，否则会出现字段映射错误。

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