PySpark使用unionByName合并DataFrame时保留原Schema

当使用unionByName合并两个列类型相似但不同的DataFrame时，Spark会自动将列类型提升为更宽泛的兼容类型（比如IntegerType转为LongType、DecimalType转为DoubleType）。但如果需要强制保留指定的目标Schema而非自动升级类型，可通过以下方案实现：

核心思路 #

先将所有待合并的DataFrame统一转换为目标Schema的类型，再执行合并操作，从根源避免自动类型提升。

通用解决方案 #

1. 定义类型转换工具函数 #

编写一个通用函数，将任意DataFrame转换为指定的目标Schema：

from pyspark.sql.types import StructType
from pyspark.sql.functions import col

def cast_df_to_target_schema(df: "DataFrame", target_schema: StructType) -> "DataFrame":
    """将DataFrame的列类型转换为目标Schema指定的类型"""
    casted_columns = []
    for field in target_schema.fields:
        # 对每个列执行类型转换，保留原列名和可空性
        casted_col = col(field.name).cast(field.dataType).alias(field.name)
        casted_columns.append(casted_col)
    return df.select(*casted_columns)

2. 应用转换并合并DataFrame #

以你的示例代码为例，我们要保留IntegerType的目标Schema，执行以下步骤：

from pyspark.sql import Row
from pyspark.sql.types import StructType, StructField, LongType, IntegerType

# 定义两个原始Schema
old_schema_test = StructType(
    [StructField("numberField", LongType(), True)]
)
new_schema_test = StructType(
    [StructField("numberField", IntegerType(), True)]
)

# 创建原始DataFrame
old_df = spark.createDataFrame([Row(numberField=1)], schema=old_schema_test)
new_df = spark.createDataFrame([Row(numberField=2)], schema=new_schema_test)

# 指定要保留的目标Schema（这里选择IntegerType的Schema）
target_schema = new_schema_test

# 将两个DataFrame转换为目标Schema
casted_old_df = cast_df_to_target_schema(old_df, target_schema)
casted_new_df = cast_df_to_target_schema(new_df, target_schema)

# 执行合并
union_df = casted_new_df.unionByName(casted_old_df)
union_df.printSchema()

# 输出结果：
# root
# |-- numberField: integer (nullable = true)

Delta表场景适配 #

对于自动推断Schema导致类型差异的Delta表，可按以下流程处理：

1. 读取多个Delta表得到DataFrame
2. 定义统一的目标Schema（比如以最早的表Schema或业务指定Schema为准）
3. 将所有读取到的DataFrame转换为目标Schema
4. 合并后写入Delta表时，强制指定Schema覆盖：

# 合并后写入Delta表，强制使用目标Schema
union_df.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "false") \
    .schema(target_schema) \
    .save("/path/to/your/delta/table")

注意事项 #

• 类型转换需确保数据兼容性：比如将LongType转为IntegerType时，要提前检查数据是否在Integer的取值范围内，避免溢出报错
• 对于DecimalType，需确保目标Schema的精度和刻度能容纳原数据，避免截断或精度丢失
• 若目标Schema包含多个字段，函数会自动遍历所有字段完成转换，无需单独处理每一列

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