如何用PySpark以通用方式展开含数组的嵌套数据？

数据老陈

2026-7-14

在PySpark中自动按需展开数组以实现嵌套字段分组

要实现针对不同嵌套结构（含数组/不含数组）自动按需展开数组，再执行分组，可以通过动态解析字段路径+递归识别数组类型+批量执行explode的方式实现，无需手动判断每种结构。以下是具体实现方案：

核心思路

拆分目标字段的层级路径（比如root.x.y.z拆分为['root', 'x', 'y', 'z']）。
递归遍历DataFrame的Schema，找出路径上所有属于ArrayType的字段。
对每个识别到的数组字段依次执行explode操作，保持字段名不变以维持路径有效性。
完成所有必要的数组展开后，正常执行groupBy。

代码实现

首先导入所需依赖：

from pyspark.sql.functions import explode, col
from pyspark.sql.types import ArrayType, StructType

辅助函数：识别路径上的数组字段

这个函数会递归遍历Schema，找出目标字段路径中所有的数组类型节点：

def find_array_fields(schema, path_parts, current_path=[]):
    array_fields = []
    if not path_parts:
        return array_fields
    
    current_part = path_parts[0]
    field = schema[current_part]
    
    # 当前节点是数组，记录完整路径并继续遍历数组元素的结构
    if isinstance(field.dataType, ArrayType):
        full_path = '.'.join(current_path + [current_part])
        array_fields.append(full_path)
        element_schema = field.dataType.elementType
        array_fields.extend(find_array_fields(element_schema, path_parts[1:], current_path + [current_part]))
    
    # 当前节点是结构体，继续遍历下一层
    elif isinstance(field.dataType, StructType):
        array_fields.extend(find_array_fields(field.dataType, path_parts[1:], current_path + [current_part]))
    
    # 非数组/结构体的节点，终止遍历
    else:
        return array_fields
    
    return array_fields

主函数：自动展开数组并执行分组

这个函数会调用上面的辅助函数，自动处理所有需要展开的数组，然后返回分组后的DataFrame：

def auto_explode_for_group(df, target_field):
    # 拆分目标字段的层级路径
    path_parts = target_field.split('.')
    # 获取所有需要展开的数组字段路径
    array_fields = find_array_fields(df.schema, path_parts)
    
    # 依次展开每个数组字段
    for field in array_fields:
        df = df.withColumn(field, explode(col(field)))
    
    # 执行分组操作
    return df.groupBy(target_field)

测试不同结构的场景

场景1：无数组的嵌套结构

data1 = [{'root': {'x': {'y': {'z': 'foo'}}}}]
df1 = spark.createDataFrame(data1)
grouped1 = auto_explode_for_group(df1, 'root.x.y.z')
grouped1.count().show()
# 输出：+-----+-----+
#      |root.x.y.z|count|
#      +-----+-----+
#      |foo  |1    |
#      +-----+-----+

场景2：中间层级为数组

data2 = [{'root': {'x': {'y': [{'z': 'foo'}, {'z': 'bar'}]}}}]
df2 = spark.createDataFrame(data2)
grouped2 = auto_explode_for_group(df2, 'root.x.y.z')
grouped2.count().show()
# 输出：+-----+-----+
#      |root.x.y.z|count|
#      +-----+-----+
#      |foo  |1    |
#      |bar  |1    |
#      +-----+-----+

场景3：根层级为数组

data3 = [{'root': [{'x': {'y': {'z': 'foo'}}}, {'x': {'y': {'z': 'bar'}}}] }]
df3 = spark.createDataFrame(data3)
grouped3 = auto_explode_for_group(df3, 'root.x.y.z')
grouped3.count().show()
# 输出：+-----+-----+
#      |root.x.y.z|count|
#      +-----+-----+
#      |foo  |1    |
#      |bar  |1    |
#      +-----+-----+

场景4：多层级嵌套数组

data4 = [{'root': [{'x': {'y': [{'z': 'foo'}, {'z': 'bar'}]}}, {'x': {'y': [{'z': 'foo'}]}}]}]
df4 = spark.createDataFrame(data4)
grouped4 = auto_explode_for_group(df4, 'root.x.y.z')
grouped4.count().show()
# 输出：+-----+-----+
#      |root.x.y.z|count|
#      +-----+-----+
#      |foo  |2    |
#      |bar  |1    |
#      +-----+-----+

注意事项

确保目标字段的路径是完整的（比如从根字段开始，如root.x.y.z而非x.y.z），否则Schema遍历会出错。
该方法会保留原字段名，展开后的数组字段依然使用原路径，因此后续分组的字段名无需修改。
如果路径中存在非数组/结构体的字段（比如直接是基本类型），函数会自动终止遍历，不会产生错误。

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