如何递归删除嵌套Struct类型DataFrame中的空值字段？

我有两个包含嵌套Struct类型列value的DataFrame：DataFrame 1和DataFrame 2。已基于DataFrame 1的Schema创建Delta表，尝试将DataFrame 2的数据追加到该表时，因DataFrame 2中部分嵌套字段（如Economic_Activities）为null，导致字段类型与Delta表Schema不匹配，抛出如下PySpark错误：

AnalysisException: Failed to merge fields 'value' and 'value'. Failed to merge fields 'Operations' and 'Operations'. Failed to merge fields 'KafkaTriggerCoreApprovedApps' and 'KafkaTriggerCoreApprovedApps'. Failed to merge fields 'Request' and 'Request'. Failed to merge fields 'Data' and 'Data'. Failed to merge fields 'Applicant' and 'Applicant'. Failed to merge fields 'Economic_Activities' and 'Economic_Activities'. Failed to merge incompatible data types ArrayType(StructType(StructField(Employer,StringType,true),StructField(Income_Amount,StructType(StructField(Amount,LongType,true)),true),StructField(Occupation,StringType,true),StructField(Ownership_Percentage,DoubleType,true),StructField(Supporting_Documents,ArrayType(StructType(StructField(Issuing_State,StringType,true)),true),true)),true) and StringType

目前我能通过手动指定字段路径删除问题字段，代码如下：

df = df.withColumn(
    "value",
    col("value").dropFields(
        "Operations.KafkaTriggerCoreApprovedApps.Request.Data.Applicant.Economic_Activities"
    )
)

但我需要实现递归处理，无需手动指定字段名，自动删除所有层级嵌套Struct中因null导致类型不匹配的字段，请问该如何实现？

可以通过递归遍历嵌套Struct字段，以Delta表的目标Schema为基准，自动识别并删除类型不匹配的字段。以下是具体实现：

1. 定义递归清理函数 #

该函数会逐层遍历Struct字段，对比目标Schema的字段类型，自动过滤因null被错误推断为简单类型（如StringType）的复杂类型字段（如ArrayType/StructType）：

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

def clean_nested_struct(current_col: Column, target_schema: StructType) -> Column:
    # 映射当前列和目标Schema的字段
    current_field_map = {f.name: f for f in current_col.dataType.fields}
    target_field_map = {f.name: f for f in target_schema.fields}
    
    cleaned_fields = []
    for field_name, target_field in target_field_map.items():
        # 跳过当前DataFrame不存在的字段
        if field_name not in current_field_map:
            continue
            
        current_field = current_field_map[field_name]
        current_col_field = current_col[field_name]
        
        # 递归处理嵌套Struct
        if isinstance(target_field.dataType, StructType):
            cleaned_substruct = clean_nested_struct(current_col_field, target_field.dataType)
            cleaned_fields.append(cleaned_substruct.alias(field_name))
        # 过滤目标为ArrayType但当前为StringType的字段（问题中的核心场景）
        elif isinstance(target_field.dataType, ArrayType) and isinstance(current_field.dataType, StringType):
            continue
        # 过滤目标为StructType但当前为StringType的字段（扩展场景）
        elif isinstance(target_field.dataType, StructType) and isinstance(current_field.dataType, StringType):
            continue
        # 保留类型匹配的字段
        else:
            cleaned_fields.append(current_col_field.alias(field_name))
    
    # 重新组合清理后的Struct
    return F.struct(*cleaned_fields)

2. 应用函数处理DataFrame #

获取Delta表的Schema，调用递归函数清理DataFrame 2的value列，之后即可正常追加数据：

# 获取Delta表的目标Schema
delta_table_schema = spark.read.table("your_delta_table_name").schema
# 提取value列的Struct类型Schema
target_value_schema = delta_table_schema["value"].dataType

# 清理DataFrame 2的嵌套字段
cleaned_df = df.withColumn(
    "value",
    clean_nested_struct(F.col("value"), target_value_schema)
)

# 追加到Delta表
cleaned_df.write.mode("append").saveAsTable("your_delta_table_name")

说明 #

• 函数以Delta表的Schema为基准，确保清理后的DataFrame结构与目标完全匹配
• 可根据实际类型不匹配的场景，扩展函数中的类型判断逻辑
• 递归逻辑会自动处理所有层级的嵌套Struct，无需手动指定字段路径

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