PySpark如何递归合并嵌套Struct内字段并转换为目标数据类型

从数据源获取到的Struct类型字段存在多层嵌套，且嵌套内部包含多种数据类型，原始结构示例如下：

|-- priority: struct (nullable = true)
 |    |-- priority_a: struct (nullable = true)
 |    |    |-- union: boolean (nullable = true)
 |    |    |-- int32: integer (nullable = true)
 |    |    |-- double: double (nullable = true)
 |    |-- priority_b: integer (nullable = true)
 |    |-- priority_c: struct (nullable = true)
 |    |    |-- union: boolean (nullable = true)
 |    |    |-- double: double (nullable = true)
 |    |    |-- int32: integer (nullable = true)
 |    |-- priority_d: struct (nullable = true)
 |    |    |-- union: boolean (nullable = true)
 |    |    |-- double: double (nullable = true)
 |    |    |-- int32: integer (nullable = true)
 |    |-- priority_e: double (nullable = true)

期望将嵌套Struct的内部字段做coalesce处理后转换为最合适的目标类型，最终输出结构如下：

|-- priority: struct (nullable = true)
 |    |-- priority_a: integer (nullable = true)
 |    |-- priority_b: integer (nullable = true)
 |    |-- priority_c: double (nullable = true)
 |    |-- priority_d: double (nullable = true)
 |    |-- priority_e: double (nullable = true)

现有可处理非嵌套Struct的代码如下：

try: 
    cols = [f'{c}.{col}' for col in source.select(f'{c}.*').columns]
    if f'{struct_path}.union' in cols:
        cols.remove(f'{struct_path}.union')
    source = source.withColumn(pc, f.coalesce(*cols).cast(t)) # t是目标转换类型
except:
    source = source.withColumn(c, f.col(c).cast(t))

可以通过递归遍历Schema的方式实现任意深度嵌套Struct的处理，以下是可直接运行的PySpark实现，完全兼容你现有的处理逻辑：

第一步：导入依赖&配置映射规则 #

from pyspark.sql import functions as f
from pyspark.sql.types import StructType, StructField, IntegerType, DoubleType

# 可根据业务需求自定义字段到目标类型的映射，也可以添加自动判断类型的逻辑替换
TYPE_MAPPING = {
    "priority_a": IntegerType(),
    "priority_b": IntegerType(),
    "priority_c": DoubleType(),
    "priority_d": DoubleType(),
    "priority_e": DoubleType()
}

第二步：编写递归处理函数 #

def process_nested_struct(schema, parent_path=""):
    processed_fields = []
    for field in schema.fields:
        field_name = field.name
        current_path = f"{parent_path}.{field_name}" if parent_path else field_name
        # 判断当前字段是否为Struct类型
        if isinstance(field.dataType, StructType):
            sub_field_names = [f.name for f in field.dataType.fields]
            # 匹配需要合并的特征Struct（包含union/int32/double字段）
            if "union" in sub_field_names and ("int32" in sub_field_names or "double" in sub_field_names):
                # 执行你原有的coalesce+转类型逻辑
                sub_cols = [f.col(f"{current_path}.{sub_name}") for sub_name in sub_field_names if sub_name != "union"]
                target_type = TYPE_MAPPING.get(field_name, field.dataType)
                processed_col = f.coalesce(*sub_cols).cast(target_type).alias(field_name)
                processed_fields.append(processed_col)
            else:
                # 普通嵌套Struct，递归处理子字段后重新组装
                sub_processed = process_nested_struct(field.dataType, current_path)
                processed_col = f.struct(*sub_processed).alias(field_name)
                processed_fields.append(processed_col)
        else:
            # 普通类型字段直接转目标类型
            target_type = TYPE_MAPPING.get(field_name, field.dataType)
            processed_col = f.col(current_path).cast(target_type).alias(field_name)
            processed_fields.append(processed_col)
    return processed_fields

第三步：调用函数处理数据 #

# 假设原始DataFrame名为source，要处理的根Struct字段为priority
root_schema = source.select("priority.*").schema
processed_sub_fields = process_nested_struct(root_schema, parent_path="priority")
# 替换原有priority字段
source = source.withColumn("priority", f.struct(*processed_sub_fields))

处理完成后可执行source.printSchema()验证结构是否符合预期，代码支持任意深度的Struct嵌套，不需要手动指定层级。

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