如何高效合并Spark DataFrame重复行并添加最大ts列

核心思路 #

按指定字段分组后，依托Spark内置聚合函数完成需求：

• 对需要保留非空值的列，用first(col, ignorenulls=True)优先提取组内第一个非空值（Spark会自动忽略空值，确保最终列保留有效数据）
• 对ts列，用max(ts)直接获取分组内的最大值

这种方案完全利用Spark原生优化，比自定义UDF或复杂逻辑高效得多，适配大数据量场景。

代码实现 #

假设你的源DataFrame名为source_df，具体代码如下：

from pyspark.sql import functions as F

# 定义分组键
group_columns = ["cartId", "orderid", "anchor_item", "dt"]

# 定义需要合并非空值的目标列（可根据实际业务扩展）
merge_target_cols = [
    "impressions_sub_item_1", "impressions_sub_item_2", "impressions_sub_item_3",
    "impressions_sub_item_4", "impressions_sub_item_5",
    "click_sub_item_1", "click_sub_item_2", "click_sub_item_3",
    "click_sub_item_4", "click_sub_item_5"
]

# 构建聚合表达式
aggregation_exprs = [
    # 对每个目标列取第一个非空值，保留原列名
    F.first(col, ignorenulls=True).alias(col) for col in merge_target_cols
] + [
    # 添加分组内的最大ts，命名为max_ts
    F.max("ts").alias("max_ts")
]

# 执行分组聚合
final_df = source_df.groupBy(*group_columns).agg(*aggregation_exprs)

关键说明 #

1. first(ignorenulls=True)的作用：

   • 若组内某列存在多个非空值，会取数据集中最早出现的那个；若仅一个非空值，则直接保留该值
   • 作为Spark内置函数，它避免了自定义UDF的序列化/反序列化开销，执行效率大幅提升

2. 扩展性：

   • 后续新增同类sub_item列时，只需在merge_target_cols列表中追加列名，无需修改核心聚合逻辑

3. 性能优化建议：

   • 确保分组键列有合理的分区或索引（针对外部表），可减少shuffle阶段的数据传输量
   • 超大规模数据场景下，可先对ts列做预聚合再合并非空列，但基础方案已覆盖绝大多数业务需求

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