PySpark中扁平化复杂嵌套数据类型：通用函数或可用库问询

当然有办法解决这个问题！Spark本身并没有内置一键扁平化所有嵌套结构的函数，但我们可以自己实现一个**无需指定列名、支持任意层级嵌套（struct/array）**的通用递归函数，同时社区也有不少成熟的工具可以参考。

一、自定义通用扁平化函数（Python版） #

这个函数会递归遍历DataFrame的Schema，自动处理所有struct和array嵌套，完全不需要手动指定列名：

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

def flatten_nested_df(df):
    # 递归处理单个列的嵌套结构
    def process_column(col_name, parent_prefix=""):
        # 生成列的别名，用下划线替换点号避免Spark列名冲突
        current_alias = f"{parent_prefix}_{col_name}" if parent_prefix else col_name
        col_dtype = df.schema[col_name].dataType
        
        if isinstance(col_dtype, StructType):
            # 递归展开struct的所有子字段
            return [process_column(sub_col, parent_prefix=current_alias) for sub_col in col_dtype.names]
        elif isinstance(col_dtype, ArrayType):
            # 先对array做explode_outer（保留null行），再处理元素
            exploded_col = F.explode_outer(F.col(col_name)).alias(f"{current_alias}_exploded")
            element_dtype = col_dtype.elementType
            
            if isinstance(element_dtype, StructType):
                # array元素是struct，继续递归展开子字段
                return [process_column(sub_col, parent_prefix=f"{current_alias}_exploded") for sub_col in element_dtype.names]
            else:
                # array元素是普通类型，直接返回explode后的列
                return [exploded_col.alias(f"{current_alias}_exploded")]
        else:
            # 普通数据类型，直接返回列并替换别名中的点号
            return [F.col(col_name).alias(current_alias.replace(".", "_"))]
    
    # 收集所有扁平化后的列（处理递归产生的嵌套列表）
    all_flattened_cols = []
    def flatten_col_list(lst):
        for item in lst:
            if isinstance(item, list):
                flatten_col_list(item)
            else:
                all_flattened_cols.append(item)
    
    # 遍历所有顶层列
    for col in df.columns:
        flatten_col_list(process_column(col))
    
    # 返回扁平化后的DataFrame
    return df.select(all_flattened_cols)

函数说明： #

• 自动识别任意层级的struct和array嵌套，不管嵌套多深都能处理
• 使用explode_outer而非explode，确保保留array为null的行，避免数据丢失
• 列名自动用下划线拼接（如array_exploded_col9_exploded_field1），避免Spark中列名含点号的问题

二、自定义通用扁平化函数（Scala版） #

如果用Scala开发，思路完全一致，递归处理Schema即可：

import org.apache.spark.sql.{DataFrame, Column}
import org.apache.spark.sql.types.{StructType, ArrayType}
import org.apache.spark.sql.functions._

def flattenNestedDF(df: DataFrame): DataFrame = {
    // 递归处理单个列
    def processColumn(colName: String, parentPrefix: String = ""): Seq[Column] = {
        val currentAlias = if (parentPrefix.isEmpty) colName else s"$parentPrefix_$colName"
        df.schema(colName).dataType match {
            case struct: StructType =>
                struct.fields.flatMap(field => processColumn(field.name, currentAlias))
            case array: ArrayType =>
                val explodedCol = explode_outer(col(colName)).alias(s"${currentAlias}_exploded")
                array.elementType match {
                    case struct: StructType =>
                        struct.fields.flatMap(field => processColumn(field.name, s"${currentAlias}_exploded"))
                    case _ =>
                        Seq(explodedCol.alias(s"${currentAlias}_exploded"))
                }
            case _ =>
                Seq(col(colName).alias(currentAlias.replace(".", "_")))
        }
    }
    
    // 收集所有扁平化列并生成新DataFrame
    val allColumns = df.columns.flatMap(col => processColumn(col))
    df.select(allColumns: _*)
}

三、第三方工具参考 #

如果你不想自己写代码，一些开源的Spark工具库也提供了类似的扁平化函数，但自己实现的优势是可以完全自定义逻辑（比如调整列名格式、是否保留原始嵌套列等），灵活性更高。

我自己在日常ETL工作中经常用这类递归函数处理复杂的嵌套DataFrame，不管是几层嵌套的struct套array，还是array套struct，都能轻松搞定。

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