使用PySpark合并不同行取值 拼接class字段并对cost求和

首先明确：F.array_join()本身不负责聚合去重，它的作用是将数组类型的列按指定分隔符拼接为字符串，在解决上游数据重复问题的前提下，配合聚合函数可以得到你预期的结果。

异常结果根因 #

你当前得到重复class、cost求和翻倍的问题，和聚合函数选择无关：是执行groupBy之前的DataFrame已经出现了行膨胀，大概率是之前做多表关联时未写全关联条件，导致产生了笛卡尔积重复行，使得id=1、product=table的分组下实际存在大于2条的重复记录，才会出现收集到两个large、cost求和为10.24的结果。

正确实现方式 #

方案1：优先排查上游重复问题（推荐） #

先修复上游join/转换逻辑的重复行问题，确保每个(id, product, class)组合仅对应1条有效记录、cost值准确后，使用如下聚合代码即可得到正确结果：

import pyspark.sql.functions as F

result = df.groupBy("id", "product").agg(
    # 收集class为数组后用逗号拼接，initcap可将首字母统一转为大写匹配你要的"large, Medium"格式
    F.array_join(F.collect_list(F.initcap("class")), ", ").alias("class"),
    F.sum("cost").alias("cost")
)

执行后返回结果为：id=1、product=table、class="Large, Medium"、cost=7.32，和你的预期一致。

方案2：聚合阶段临时去重（应急使用） #

如果暂时无法定位上游重复来源，且你的业务场景下同一个(id, product, class)组合对应的cost值固定，可以使用去重聚合临时规避问题：

import pyspark.sql.functions as F

result = df.groupBy("id", "product").agg(
    # collect_set自动对分组内的class值去重
    F.array_join(F.collect_set(F.initcap("class")), ", ").alias("class"),
    # sum_distinct对分组内的cost去重后求和，避免重复累加
    F.sum_distinct("cost").alias("cost")
)

注意：该方案仅适合同一class对应唯一cost的场景，如果存在同class不同cost的情况会丢失数据，不建议作为长期方案使用。

窗口函数写法的问题 #

你之前写的窗口函数仅按id分区，如果同一个id下对应多个product，计算出的total会是该id下所有product的cost总和，不符合你按id+product聚合的粒度要求。如果需要保留原表所有行、同时展示分组聚合后的cost总和，需要把分区条件改为和聚合粒度一致：

from pyspark.sql import Window
import pyspark.sql.functions as F

df_with_total = df.withColumn(
    "total_cost", 
    F.sum("cost").over(Window.partitionBy("id", "product"))
)

注意窗口函数不会合并行，最终返回的结果和原表行数一致，不会得到你要的单条合并记录。

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