如何按列顺序union多个DataFrame并自动转换字段为timestamp类型

• 多个列顺序完全一致的PySpark DataFrame，各DataFrame列名可能存在差异
• 其中固定2个字段应为timestamp类型，但部分DataFrame对应位置字段为date类型，直接执行union会因类型不匹配失败
• 尝试仅修改第一个DataFrame的对应字段为timestamp类型，认为后续union会自动对齐类型，方案未生效，原代码如下：

from pyspark.sql import functions as F

def change_type_timestamp(df):
    df = df.withColumn("A", F.to_timestamp(F.col("A"))) \
        .withColumn("B", F.to_timestamp(F.col("B")))
    return df

dfs = [df1, df2, df3, ...]

dfs[0] = change_type_timestamp(dfs[0])
reduce(lambda a, b: a.union(b), dfs)

PySpark的union是按列位置做严格匹配的操作，不会执行隐式类型转换，要求对应位置的字段数据类型完全一致，不会因为前一个DataFrame的字段是timestamp类型，就自动把后续DataFrame同位置的date类型做转换。

不需要手动逐个为每个DataFrame编写类型转换代码，核心思路是先定义好符合预期的基准schema，union时自动将所有后续DataFrame的字段按位置对齐到基准类型即可。
由于所有DataFrame列顺序完全一致，不需要关心列名差异，直接按位置做类型映射即可：

from functools import reduce
from pyspark.sql import functions as F

def align_to_schema(source_df, ref_schema):
    # 按列位置自动将源df字段转换为参考schema对应位置的类型
    aligned_cols = [
        F.col(source_df.columns[idx]).cast(ref_schema.fields[idx].dataType)
        for idx in range(len(source_df.columns))
    ]
    return source_df.select(*aligned_cols)

def set_base_timestamp_type(df):
    # 按字段位置指定需要转timestamp的列，此处假设需要转换的是第1、第2个位置的列
    # 可根据实际字段位置调整索引，不需要匹配列名
    col1, col2 = df.columns[0], df.columns[1]
    return df.withColumn(col1, F.to_timestamp(F.col(col1))) \
             .withColumn(col2, F.to_timestamp(F.col(col2)))

dfs = [df1, df2, df3, ...]

# 生成基准df，确定最终输出schema
base_df = set_base_timestamp_type(dfs[0])
# 剩余df自动对齐schema后完成union，无需逐个手动处理
final_df = reduce(
    lambda acc, curr_df: acc.union(align_to_schema(curr_df, acc.schema)),
    dfs[1:],
    base_df
)

方案说明 #

• 所有类型对齐逻辑在union流程中自动完成，不需要单独遍历每个DataFrame手动调用转换方法
• 完全适配列名不一致的场景，所有匹配逻辑按列位置执行，符合union的默认行为
• 除了date转timestamp的需求外，其他位置如果存在字段类型不匹配的问题，也会自动按基准schema做类型转换，不需要额外修改逻辑
• 注意：该方案依赖「所有DataFrame列顺序完全一致」的前提，如果列顺序存在变动，需要先做列顺序对齐再执行上述逻辑。

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