本文介绍如何使用 PySpark 的窗口函数（Window）配合 count() 实现按指定列分组后对某列的去重数量统计，并以 withColumn 方式添加新列，解决多行重复导致的非预期计数问题。
本文介绍如何使用 pyspark 的窗口函数（window）配合 `count()` 实现按指定列分组后对某列的**去重数量统计**，并以 `withcolumn` 方式添加新列，解决多行重复导致的非预期计数问题。
在 PySpark 中，直接使用 count(col('user_id')).over(window_spec)
默认执行的是总行数计数，而非去重计数（distinct count）
。因此，若原始数据中存在重复 user_id（如示例中 FWE + QRS 组合下 user_id=123 出现两次），上述写法会返回该分组的总记录数（3 行 → un_count=3），而非期望的唯一用户数（2 个不同 user_id：123 和 223）。
✅ 正确做法是使用
countDistinct()
窗口聚合函数：
⚠️ 注意事项：
countDistinct() 是 PySpark 3.0+ 推荐的标准函数；旧版本可使用 approx_count_distinct()（近似去重，性能更高但有误差）或通过子查询 + agg() + join 实现精确去重（不推荐，复杂且低效）；
窗口函数无法直接在 countDistinct() 中嵌套 distinct 关键字（如 count(distinct 'user_id') 语法非法）；
若需兼容 Spark 2.x 且必须精确计数，可改用 collect_set('user_id').over(window_spec) 获取去重集合，再用 size() 计算长度（但内存开销大，不适用于海量唯一值）：
最终输出将严格匹配预期结果：FWE/QRS 组返回 un_count=2（对应 user_id 123 和 223），RWE/QRS 组返回 un_count=1（仅 user_id=223），每行均携带所属窗口的去重统计值，满足分析与报表场景需求。
from pyspark.sql import Window
from pyspark.sql.functions import countDistinct, col
# 定义窗口：按 plan 和 name 分组
window_spec = Window.partitionBy('plan', 'name')
# 添加新列 un_count：每个 (plan, name) 组内 user_id 的去重数量
df = df.withColumn('un_count', countDistinct('user_id').over(window_spec))from pyspark.sql.functions import collect_set, size
df = df.withColumn('un_count', size(collect_set('user_id')).over(window_spec))