Spark如何合并含不同Struct字段的数组列Schema？

Spark的mergeSchema选项仅能处理顶层字段的合并，对于数组内嵌套的StructType字段无法自动对齐，因此需要手动统一两个DataFrame的Schema后再执行合并。以下是通用的解决方案，支持任意层级的嵌套结构（Struct、Array嵌套Struct等）。

步骤1：递归合并两个Schema #

首先编写递归函数，将两个Schema合并为包含所有字段的目标Schema，同时处理嵌套类型的合并：

from pyspark.sql.types import StructType, StructField, ArrayType

def merge_schemas(schema1: StructType, schema2: StructType) -> StructType:
    # 获取两个Schema的所有字段名集合
    all_field_names = set(f.name for f in schema1.fields) | set(f.name for f in schema2.fields)
    merged_fields = []

    for field_name in all_field_names:
        field_from_df1 = schema1.get(field_name)
        field_from_df2 = schema2.get(field_name)

        if not field_from_df1:
            # df1无此字段，直接用df2的字段定义
            merged_fields.append(field_from_df2)
        elif not field_from_df2:
            # df2无此字段，直接用df1的字段定义
            merged_fields.append(field_from_df1)
        else:
            # 处理嵌套类型的合并
            if isinstance(field_from_df1.dataType, StructType) and isinstance(field_from_df2.dataType, StructType):
                # 递归合并StructType
                merged_data_type = merge_schemas(field_from_df1.dataType, field_from_df2.dataType)
            elif isinstance(field_from_df1.dataType, ArrayType) and isinstance(field_from_df2.dataType, ArrayType):
                # 合并数组元素类型（仅处理元素为Struct的情况）
                elem_type1 = field_from_df1.dataType.elementType
                elem_type2 = field_from_df2.dataType.elementType
                if isinstance(elem_type1, StructType) and isinstance(elem_type2, StructType):
                    merged_elem_type = merge_schemas(elem_type1, elem_type2)
                else:
                    # 非Struct数组元素，取兼容类型（此处默认取df2的类型，可根据需求调整）
                    merged_elem_type = elem_type2 if elem_type1 != elem_type2 else elem_type1
                merged_data_type = ArrayType(merged_elem_type, field_from_df1.dataType.containsNull or field_from_df2.dataType.containsNull)
            else:
                # 基本类型，取兼容类型（此处默认取df2的类型，可根据需求调整）
                merged_data_type = field_from_df2.dataType if field_from_df1.dataType != field_from_df2.dataType else field_from_df1.dataType
            
            merged_fields.append(StructField(
                field_name,
                merged_data_type,
                field_from_df1.nullable or field_from_df2.nullable
            ))
    
    return StructType(merged_fields)

步骤2：将两个DataFrame对齐到目标Schema #

编写对齐函数，将每个DataFrame的结构转换为合并后的目标Schema，缺失的字段自动填充为null：

from pyspark.sql import DataFrame
from pyspark.sql.functions import col, transform, lit, struct

def align_df_to_schema(df: DataFrame, target_schema: StructType) -> DataFrame:
    def align_field(df_column, target_field):
        current_type = df_column.dataType
        target_type = target_field.dataType

        if isinstance(target_type, StructType):
            if isinstance(current_type, StructType):
                # 递归对齐Struct字段的子字段
                struct_exprs = []
                for sub_field in target_type.fields:
                    if sub_field.name in current_type.fieldNames():
                        struct_exprs.append(align_field(df_column[sub_field.name], sub_field).alias(sub_field.name))
                    else:
                        # 缺失的子字段填充为null
                        struct_exprs.append(lit(None).cast(sub_field.dataType).alias(sub_field.name))
                return struct(*struct_exprs)
            else:
                # 类型不匹配时强制转换（根据需求调整）
                return df_column.cast(target_type)
        elif isinstance(target_type, ArrayType):
            target_elem_type = target_type.elementType
            if isinstance(current_type, ArrayType):
                current_elem_type = current_type.elementType
                if isinstance(target_elem_type, StructType):
                    # 对齐数组内的每个Struct元素
                    return transform(df_column, lambda elem: align_field(elem, StructField("elem", target_elem_type)).elem)
                else:
                    return df_column.cast(target_type)
            else:
                return df_column.cast(target_type)
        else:
            # 基本类型直接转换
            return df_column.cast(target_type)
    
    # 生成所有字段的对齐表达式
    select_exprs = []
    for target_field in target_schema.fields:
        if target_field.name in df.columns:
            select_exprs.append(align_field(col(target_field.name), target_field).alias(target_field.name))
        else:
            # 顶层缺失字段填充为null
            select_exprs.append(lit(None).cast(target_field.dataType).alias(target_field.name))
    
    return df.select(*select_exprs)

步骤3：执行合并 #

使用上述函数处理两个DataFrame后，通过unionByName完成合并：

# 假设df1、df2是需要合并的两个DataFrame
target_schema = merge_schemas(df1.schema, df2.schema)

# 对齐两个DataFrame到目标Schema
df1_aligned = align_df_to_schema(df1, target_schema)
df2_aligned = align_df_to_schema(df2, target_schema)

# 执行合并
final_merged_df = df1_aligned.unionByName(df2_aligned)

方案说明 #

• 通用性：支持任意层级的嵌套结构（Struct嵌套Struct、Array嵌套Struct等），无需针对特定字段修改代码。
• 兼容性：自动处理字段缺失问题，缺失字段填充为null；类型不一致时默认取第二个Schema的类型，可根据实际需求调整类型兼容逻辑。
• 替代mergeSchema：解决了原生mergeSchema无法处理数组内嵌套Struct的缺陷。

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