Spark Scala：如何通用处理DataFrame数组列空值并替换默认值

核心逻辑拆解 #

• 自动识别数组列：遍历DataFrame的Schema，判断每列是否为ArrayType类型
• 双维度处理数组列：
  1. 列级null替换：用coalesce将整列null替换为指定默认数组（默认是空数组）
  2. 数组内空元素清理：用filter移除数组中的null值和空字符串（可按需调整过滤规则）
• 动态生成处理后的列：对数组列应用上述逻辑，非数组列直接保留原列，最终生成新的DataFrame

Scala 实现代码 #

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

def cleanArrayColumns(df: org.apache.spark.sql.DataFrame, defaultArray: Seq[Any] = Seq.empty): org.apache.spark.sql.DataFrame = {
  val processedColumns = df.schema.fields.map { field =>
    if (field.dataType.isInstanceOf[ArrayType]) {
      // 先替换列级null为默认数组，再过滤数组内空元素
      val cleanedArray = filter(
        coalesce(col(field.name), lit(defaultArray)), 
        x => x.isNotNull && trim(x) =!= ""
      )
      cleanedArray.as(field.name)
    } else {
      col(field.name)
    }
  }
  df.select(processedColumns: _*)
}

// 使用方式
val cleanedDf = cleanArrayColumns(originalDf)
cleanedDf.show()

Python 实现代码 #

from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType

def clean_array_columns(df, default_array=[]):
    processed_columns = []
    for field in df.schema.fields:
        if isinstance(field.dataType, ArrayType):
            # 处理数组列：替换列级null + 过滤空元素
            cleaned_array = F.filter(
                F.coalesce(F.col(field.name), F.lit(default_array)),
                lambda x: x.isNotNull() & (F.trim(x) != "")
            )
            processed_columns.append(cleaned_array.alias(field.name))
        else:
            processed_columns.append(F.col(field.name))
    return df.select(*processed_columns)

// 使用方式
cleaned_df = clean_array_columns(original_df)
cleaned_df.show()

可调整细节 #

• 默认值自定义：修改defaultArray/default_array参数，可指定列级null的替换值（比如["未知"]）
• 过滤规则调整：如果只需要移除数组中的null值，删除trim(x) =!= ""/F.trim(x) != ""即可

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