如何在Pyspark中统计近30天出现次数并将行值转置为新列

问题说明 #

需要统计每个id、name、branch组合下，status列各取值近30天的出现次数，基准日期固定为2021年7月19日。

现有代码核心问题 #

1. 日期处理不规范：输入的eventDate是dd/MM/yyyy格式的字符串，未转成日期类型直接计算日期间隔会出错；且题目要求固定基准日期，不要用current_timestamp()避免运行时间不同导致结果偏差。
2. 分组逻辑冗余：二次分组时多余携带了status字段，导致透视逻辑异常。
3. 聚合函数误用：collect_list()会把统计值收集为数组，不符合数值输出要求。

最简实现代码 #

from pyspark.sql import functions as F

# 转换日期格式，定义固定基准日期
df = df.withColumn("eventDate", F.to_date(F.col("eventDate"), "dd/MM/yyyy"))
base_date = F.lit("2021-07-19").cast("date")

# 分组+透视+统计一步完成
result_df = df.groupBy("id", "name", "branch") \
              .pivot("status") \
              .agg(
                  F.sum(F.when(F.datediff(base_date, F.col("eventDate")) <= 30, 1).otherwise(0))
              ) \
              .na.fill(0) # 无对应状态的场景默认补0

原有代码修复方案 #

如果要保留之前的中间计算逻辑，只需修改最后一段分组透视代码即可：

# 原有中间计算逻辑调整（新增日期转换、替换基准日期）
from pyspark.sql import functions as F
df = df.withColumn("eventDate", F.to_date(F.col("eventDate"), "dd/MM/yyyy"))
df = df.withColumn("eventAgeinDays", (F.datediff(F.lit("2021-07-19").cast("date"), F.col("eventDate"))))
df = df.groupBy('id', 'branch', 'name', 'status')\
       .agg(
        F.sum(
            F.when(F.col("eventAgeinDays") <= 30, 1).otherwise(0)
        ).alias("Last30dayFailure")
)

# 仅修改下方透视逻辑
df = df.groupBy('id', 'branch', 'name')\ # 不要携带status字段分组
       .pivot('status')\
       .agg(F.first('Last30dayFailure'))\ # 替换collect_list为first取唯一统计值
       .na.fill(0) # 空值补0

两种方式运行后都可以得到要求的预期输出。

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