如何在PySpark中结合Date_Sub与聚合函数创建新列？

要实现基于聚合后的最小/最大日期，用date_sub生成新列，需根据全局聚合或分组聚合场景选择对应方案，先纠正你之前尝试的问题：

• 第一个代码仅对每行DayDate执行日期减法（语法还存在错误，正确写法应为F.date_sub("DayDate", 364)），未使用聚合函数，无法得到全局/分组维度的最小日期结果；
• 第二个代码存在语法错误（F.date_sub(min(col("DayDate")),1)后多余364），且select会生成仅含聚合列的新DataFrame，无法在原数据中添加目标列。

方案1：全局聚合（基于整个数据集的最小/最大日期） #

若要基于全量数据的最小日期计算去年起始日期，可先通过agg获取聚合值，再用withColumn将结果添加到原DataFrame：

from pyspark.sql import functions as F

# 获取全局最小日期
global_min_day = df.agg(F.min("DayDate")).first()[0]
# 添加目标新列
df = df.withColumn("Last_Year_START_Date", F.date_sub(F.lit(global_min_day), 364))

也可直接在withColumn中嵌套聚合逻辑，简化代码：

df = df.withColumn(
    "Last_Year_START_Date",
    F.date_sub(F.lit(df.agg(F.min("DayDate")).first()[0]), 364)
)

将F.min替换为F.max，即可基于全局最大日期计算。

方案2：分组聚合（基于分组后的最小/最大日期） #

若需按指定列分组，计算每组的最小日期并生成对应新列，需使用窗口函数实现：

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

# 定义窗口规则：按目标列分组
group_window = Window.partitionBy("你的分组列名")

# 先添加分组内的最小日期列，再计算去年起始日期
df = df.withColumn(
    "Group_Min_Day",
    F.min("DayDate").over(group_window)
).withColumn(
    "Last_Year_START_Date",
    F.date_sub(F.col("Group_Min_Day"), 364)
)

同样，替换F.min为F.max即可基于分组内的最大日期计算。

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