如何通过循环遍历字典合并多份Spark DataFrame（指定列）

当然可以用循环替代硬编码的多次join操作，这种方式更灵活，不管CSV文件数量多少都能自动适配，下面是两种可行的实现方式：

第一步：整理DataFrame列表 #

先把字典里的所有DataFrame提取成一个列表，方便后续处理：

# 从原字典d中提取所有DataFrame
dfs = list(d.values())

方法一：使用functools.reduce批量合并 #

用reduce可以把join操作批量应用到所有DataFrame上，代码更简洁：

from functools import reduce
from pyspark.sql import DataFrame

def outer_join_on_keys(df1: DataFrame, df2: DataFrame) -> DataFrame:
    # 基于colA和colB做outer join
    return df1.join(df2, ["colA", "colB"], "outer")

# 合并所有DataFrame
df_merged = reduce(outer_join_on_keys, dfs)

方法二：普通for循环逐步合并 #

如果觉得reduce不好理解，也可以用常规循环实现：

# 处理空列表的边界情况（没有CSV文件时）
if not dfs:
    # 可根据需求创建空DataFrame，或抛出提示
    df_merged = spark.createDataFrame([], schema="colA string, colB string")
else:
    # 初始化合并结果为第一个DataFrame
    df_merged = dfs[0]
    # 依次和剩下的DataFrame做join
    for df in dfs[1:]:
        df_merged = df_merged.join(df, ["colA", "colB"], "outer")

注意事项 #

• 如果不同CSV文件里除了colA和colB之外有重名列，Spark会自动给重复列添加后缀（比如_col1），如果需要自定义处理重名列，可以在join前对列重命名，或者通过select过滤掉不需要的列。
• 确保所有DataFrame的colA和colB字段类型一致，否则join可能会出现类型不匹配的问题。

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