PySpark 2.4.3合并列名不同的表时保留全部列的方案咨询

核心思路 #

• 取两个待合并DataFrame的列名并集，生成完整的列列表
• 分别为两个DataFrame补全缺失的列，缺失列值填充null
• 将两个DataFrame的列顺序调整为和完整列列表完全一致
• 执行常规union操作即可保留所有字段

可直接复用的工具代码 #

PySpark 实现 #

from pyspark.sql.functions import lit

def union_by_name(df1, df2, keep_original_order=True):
    """
    Spark 2.4 版本替代unionByName的实现
    keep_original_order: 为True时优先保留df1的列顺序，再追加df2独有的列；为False时按字母排序列
    """
    if keep_original_order:
        all_cols = df1.columns + [col for col in df2.columns if col not in df1.columns]
    else:
        all_cols = sorted(list(set(df1.columns + df2.columns)))
    
    # 补全df1缺失列
    for col in all_cols:
        if col not in df1.columns:
            df1 = df1.withColumn(col, lit(None))
    # 补全df2缺失列
    for col in all_cols:
        if col not in df2.columns:
            df2 = df2.withColumn(col, lit(None))
    
    return df1.select(all_cols).union(df2.select(all_cols))

调用示例 #

# 直接调用工具函数即可得到保留所有列的合并结果
union_by_name(df1, df2).show()

输出结果 #

+---------+-------+----+----+----+----+----+
|month_key|account| ch1| ch2| ch3| ch4| ch5|
+---------+-------+----+----+----+----+----+
|      Aug|    abc|   0|   1|null|null|   1|
|     Sept|    bcd|null|null|   1|   1|   0|
+---------+-------+----+----+----+----+----+

注意事项 #

• 如果两个DataFrame的同名字段数据类型不一致，需要先做类型对齐，否则union操作会报错
• 如果需要合并多个DataFrame，可将函数扩展为接收DataFrame列表，批量补全缺失列后再合并
• 数百个字段的场景下该方案依然适用，不需要手动枚举任何字段

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