Python/Pyspark中实现月度对应天数阈值去重ID计数的高效方法

实现思路 #

核心是仅对数据集做1次全量扫描，一次性计算所有阈值对应的去重ID数，避免多次遍历造成的性能损耗，最终直接转换为你需要的结果格式。

完整PySpark实现代码 #

from pyspark.sql import functions as F

# 1. 自定义你需要的天数阈值列表，步长30，最大到3600
threshold_list = list(range(30, 3601, 30))

# 2. 构造聚合表达式，一次性计算所有阈值对应的去重ID数
agg_expr = [
    F.countDistinct(F.when(F.col("Days") >= t, F.col("ID"))).alias(f"threshold_{t}") 
    for t in threshold_list
]
agg_result = df.agg(*agg_expr)

# 3. 把宽表转换为你需要的长表格式
final_result = agg_result.select(
    F.explode(F.array(*[
        F.struct(
            F.lit(f"{t} days and greater").alias("month"),
            F.col(f"threshold_{t}").alias("count")
        ) for t in threshold_list
    ])).alias("tmp")
).select("tmp.*")

# 查看结果
final_result.show(truncate=False)

# 直接导出到Excel，无需手动复制
final_result.toPandas().to_excel("days_stat_result.xlsx", index=False)

优化提示 #

如果你的原始DF中每个ID唯一、仅对应一个Days值，可把代码中的countDistinct替换为count，性能会进一步提升。

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