PySpark如何合并DataFrame中的重复同名数据列？

核心逻辑为按列名分组后对同组列做聚合处理，以下为可直接运行的完整实现： #

前置依赖导入 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import coalesce, col
from collections import defaultdict

步骤1：初始化Spark并构造测试样例（可跳过该步骤直接使用你自己的DataFrame） #

spark = SparkSession.builder.appName("merge_duplicate_columns").getOrCreate()

# 测试数据和示例输入结构一致，包含两个重名的age列
data = [("Alice", 25, None), ("Bob", None, 30), ("Charlie", 35, None)]
df = spark.createDataFrame(data, schema=["name", "age", "age"])

步骤2：对同名列进行分组 #

col_groups = defaultdict(list)
for idx, col_name in enumerate(df.columns):
    col_groups[col_name].append(idx)

步骤3：生成合并后的列表达式 #

默认采用取第一个非空值的规则合并，你可以根据需求替换为其他聚合函数：

merged_cols = []
for col_name, idxs in col_groups.items():
    if len(idxs) == 1:
        # 无重复的列直接保留
        merged_cols.append(col(col_name))
    else:
        # 重复列合并，此处用coalesce取第一个非空值
        merged_cols.append(coalesce(*[df[idx] for idx in idxs]).alias(col_name))

步骤4：执行合并得到结果 #

result_df = df.select(*merged_cols)
result_df.show()

输出结果 #

+-------+---+
|   name|age|
+-------+---+
|  Alice| 25|
|    Bob| 30|
|Charlie| 35|
+-------+---+

自定义合并规则说明 #

如果需要其他合并逻辑，只需要修改步骤3中的聚合函数即可：

• 数值型同名列求和：把coalesce(*[df[idx] for idx in idxs])替换为sum(*[df[idx] for idx in idxs])
• 字符串同名列拼接：替换为concat_ws(",", *[df[idx] for idx in idxs])即可用逗号拼接所有同名列的值

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