PySpark实现cost列滞后1位后4窗口右对齐滚动求和求助

你的代码有两个核心问题导致运行异常：

• Spark不支持在sum()函数里直接嵌套lag()窗口函数，必须分步计算
• 所有窗口定义都缺少orderBy，不管是lag还是滚动求和，都依赖确定的行顺序，不加排序的话结果会随机出错

正确实现步骤 #

1. 先定义带排序的分区窗口，用于计算cost列的滞后1位值（必须指定排序字段，比如数据中的日期/月份列，这里假设是date，请替换成你的实际排序字段）
2. 新增cost_lag1列存储滞后1位的结果
3. 再定义滚动求和的窗口（同样需要分区+排序），窗口范围设为rowsBetween(-3, 0)，对应右对齐的窗口大小4
4. 基于cost_lag1列计算滚动求和

完整代码示例 #

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

# 1. 定义计算lag的窗口：按year分区，按排序字段（比如date）升序排列
lag_window = Window.partitionBy("year").orderBy("date")

# 2. 新增滞后1位的列
cost_df = cost_df.withColumn("cost_lag1", F.lag(F.col("cost"), 1).over(lag_window))

# 3. 定义滚动求和的窗口：同样分区+排序，窗口范围是当前行及前3行（共4行）
rolling_window = Window.partitionBy("year").orderBy("date").rowsBetween(-3, 0)

# 4. 计算滚动求和
cost_df = cost_df.withColumn("cost_calc", F.sum(F.col("cost_lag1")).over(rolling_window))

注意事项 #

• 务必替换orderBy("date")中的date为你数据里实际的排序字段（比如月份、季度等），否则行顺序不确定，结果会异常
• 如果滞后1位的行不存在（比如每个year的第一行），cost_lag1会是null，求和时会自动忽略null值；如果需要填充默认值，可以用F.coalesce(F.lag(...), F.lit(0))处理

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