如何将PySpark DataFrame中的嵌套字段提升至顶层？

一、核心思路 #

嵌套结构分两类针对性处理：

• Struct类型字段：直接通过字段路径提取至顶层，不会改变原DataFrame的行数
• Array类型字段：需用explode/explode_outer展开数组，展开后行数会对应数组元素数量增加

二、分步操作代码示例 #

假设你的临时表名为nested_table，先加载为DataFrame：

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

df = spark.table("nested_table")

1. 提取顶层Struct嵌套字段（无行数变化） #

直接用.withColumn提取meta.view下的非数组Struct字段，用下划线拼接路径作为新列名，避免字段名冲突：

# 提取meta.view下的基础Struct字段
df = df.withColumn("meta_view_assetType", F.col("meta.view.assetType")) \
       .withColumn("meta_view_attribution", F.col("meta.view.attribution")) \
       .withColumn("meta_view_averageRating", F.col("meta.view.averageRating")) \
       .withColumn("meta_view_category", F.col("meta.view.category")) \
       .withColumn("meta_view_createdAt", F.col("meta.view.createdAt")) \
       .withColumn("meta_view_description", F.col("meta.view.description")) \
       # 提取嵌套更深的Struct字段
       .withColumn("meta_view_owner_displayName", F.col("meta.view.owner.displayName")) \
       .withColumn("meta_view_owner_id", F.col("meta.view.owner.id"))

# 带空格的字段必须用反引号包裹路径
df = df.withColumn("meta_view_metadata_agency", F.col("meta.view.metadata.custom_fields.`Dataset Information`.Agency")) \
       .withColumn("meta_view_metadata_see_also", F.col("meta.view.metadata.custom_fields.`Additional Resources`.`See Also`"))

2. 处理Array类型字段（行数会增加） #

（1）处理一维数组（如meta.view.approvals）

先用explode_outer展开数组（保留含null的原数据行），再提取数组内嵌套Struct的字段：

# 展开approvals数组，生成临时列存储数组元素
df_approvals = df.withColumn("approvals_element", F.explode_outer(F.col("meta.view.approvals")))

# 提取数组元素内的嵌套字段
df_approvals = df_approvals.withColumn("approval_reviewedAt", F.col("approvals_element.reviewedAt")) \
                          .withColumn("approval_state", F.col("approvals_element.state")) \
                          .withColumn("approval_submissionId", F.col("approvals_element.submissionId")) \
                          .withColumn("approval_submitter_id", F.col("approvals_element.submitter.id"))

# 删除临时数组元素列
df_approvals = df_approvals.drop("approvals_element")

（2）处理二维数组（如data字段）

需逐层展开数组，最终将每个字符串元素转为单独行：

# 展开data的第一层数组
df_data = df.withColumn("data_level1", F.explode_outer(F.col("data")))
# 展开第二层数组，得到最终的字符串元素
df_data = df_data.withColumn("data_value", F.explode_outer(F.col("data_level1")))

# 删除中间过渡列
df_data = df_data.drop("data", "data_level1")

3. 批量提取Struct字段的自动化技巧 #

如果嵌套Struct字段过多，手动编写.withColumn效率低，可以用递归遍历Schema的方式自动生成提取逻辑：

def flatten_struct(schema, prefix=""):
    fields = []
    for field in schema.fields:
        full_path = prefix + "." + field.name if prefix else field.name
        if isinstance(field.dataType, StructType):
            # 递归遍历子Struct
            fields += flatten_struct(field.dataType, full_path)
        else:
            # 跳过Array类型字段，单独处理
            if not isinstance(field.dataType, ArrayType):
                # 用下划线替换路径中的点号作为新列名
                new_col_name = full_path.replace(".", "_")
                fields.append(F.col(full_path).alias(new_col_name))
    return fields

# 自动生成所有Struct字段的提取逻辑
struct_extract_fields = flatten_struct(df.schema)
# 生成扁平化后的DataFrame
df_flattened = df.select(*struct_extract_fields)

三、关键注意事项 #

• 数组展开会导致行数激增，若需保留原数据行与数组元素的关联，可使用posexplode替代explode，同时获取数组元素的索引位置。
• 带空格或特殊字符的字段名，必须用反引号包裹完整路径。
• 若存在可能为null的数组，务必使用explode_outer，避免丢失原数据行。
• 操作完成后，可通过df.printSchema()查看新Schema，确认所有目标字段已提取至顶层。

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