Scala-Spark如何实现含数组的嵌套Schema扁平化并保留数组数据？

要实现既能扁平化嵌套结构，又能自动展开数组生成多行的需求，你只需要修改原来的flattenSchema函数，在递归处理Schema时识别数组类型，加入explode（或explode_outer）操作，同时保留数组内部嵌套结构的扁平化逻辑。

核心思路 #

1. 递归遍历Schema的每个字段：
   • 遇到结构体（StructType）：继续递归扁平化子字段，保留字段路径前缀
   • 遇到数组（ArrayType）：
     • 如果数组元素是结构体：先explode数组，再递归扁平化结构体的子字段，并重命名列以保留原路径
     • 如果是简单类型数组：直接explode并保留原字段名
   • 基本类型字段：直接保留原字段

代码实现 #

Scala版本

import org.apache.spark.sql.types.{StructType, StructField, ArrayType}
import org.apache.spark.sql.{Column, functions => F}

def flattenSchema(schema: StructType, prefix: String = ""): Array[Column] = {
  schema.fields.flatMap { field =>
    val fieldName = if (prefix.isEmpty) field.name else s"$prefix.${field.name}"
    field.dataType match {
      case structType: StructType =>
        // 递归处理嵌套结构体
        flattenSchema(structType, fieldName)
      case arrayType: ArrayType if arrayType.elementType.isInstanceOf[StructType] =>
        // 展开数组并扁平化内部结构体
        val explodedCol = F.explode(F.col(fieldName)).alias(s"exploded_${field.name}")
        val elementSchema = arrayType.elementType.asInstanceOf[StructType]
        // 处理结构体字段并重命名，保留原数组字段的路径前缀
        flattenSchema(elementSchema, s"exploded_${field.name}").map(col => 
          col.alias(s"${fieldName}.${col.toString.split("\\.").last}")
        )
      case arrayType: ArrayType =>
        // 处理简单类型数组，直接展开并保留原字段名
        Array(F.explode(F.col(fieldName)).alias(fieldName))
      case _ =>
        // 基本类型字段直接返回
        Array(F.col(fieldName))
    }
  }
}

调用方式和你原来的一致：

val flattenedDF = df.select(flattenSchema(df.schema): _*)

Python版本

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

def flatten_schema(schema, prefix=""):
    columns = []
    for field in schema.fields:
        field_name = field.name if not prefix else f"{prefix}.{field.name}"
        if isinstance(field.dataType, StructType):
            # 递归处理嵌套结构体
            columns.extend(flatten_schema(field.dataType, field_name))
        elif isinstance(field.dataType, ArrayType):
            if isinstance(field.dataType.elementType, StructType):
                # 展开数组并扁平化内部结构体
                exploded_col = F.explode(F.col(field_name)).alias(f"exploded_{field.name}")
                # 处理结构体字段并重命名
                nested_cols = flatten_schema(field.dataType.elementType, f"exploded_{field.name}")
                renamed_cols = [col.alias(f"{field_name}.{col._jc.toString().split('.')[-1]}") for col in nested_cols]
                columns.extend(renamed_cols)
            else:
                # 处理简单类型数组
                columns.append(F.explode(F.col(field_name)).alias(field_name))
        else:
            # 基本类型字段直接添加
            columns.append(F.col(field_name))
    return columns

调用方式：

flattened_df = df.select(*flatten_schema(df.schema))

关键注意事项 #

1. 笛卡尔积问题：如果你的DataFrame中有多个数组字段，同时展开所有数组会产生笛卡尔积（比如示例中同时展开hobbies和orders会生成多行组合）。如果只需要展开特定数组，可以修改函数逻辑，只对目标数组字段执行explode，其他数组保留原样（比如判断字段名后跳过explode，直接返回原数组列）。
2. 保留Null值：如果需要保留数组为Null的行，可以把explode替换为explode_outer，这样展开后的字段会显示为Null而不是过滤掉该行。
3. 列名可读性：代码中通过重命名操作保留了原字段路径（比如orders.order_id），避免了exploded_orders.order_id这种不直观的列名。

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