如何使用pyspark.sql.function.lag实现销售列按整周偏移

• 分区逻辑完全错位：当前把year、month、week全部放进了partitionBy的分区维度，相当于每周的所有记录被单独拆成独立窗口，窗口内只有当前周的行，不管怎么偏移都拿不到上一周的数据，本质上根本没有跨周的查询能力。
• 窗口排序缺乏匹配锚点：当前排序仅按week列排序，既没有标记每条记录在所属周内的位置序号，排序维度也无法支撑跨周同位置匹配，最终只会在同周窗口内逐行平移相邻行的销售值，完全达不到取上周同位置值的效果。

要实现整周同序位偏移，需要两个核心前置动作，再搭配正确的窗口参数：

1. 为每条日度记录生成周内序位标记：标记当前记录是所属周的第几天（从1开始计数），这是跨周匹配同位置数据的核心参照。
2. 调整窗口分区维度：把month、week从分区列里去掉，只保留固定业务维度和年份（保留年份是为了避免跨年时误取上一年同周的数据），让同一商品、同一门店、同一年份的所有周度记录落在同一个跨周窗口里。
3. 调整窗口排序规则：窗口内先按周内序位升序排序，再按周次升序排序，保证同序位的记录在窗口内按时间顺序连续排列。

对应的可直接运行的代码参考：

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

# 第一步：计算每条记录的周内序位（同周内按实际交易日期排序生成从1开始的序号）
seq_window = (
    Window
    .partitionBy('store', 'state_prov_cd', 'department', 'item', 'year', 'week')
    .orderBy('sales_date') # 替换为你实际存储交易日期的列名
)
df_base = data.withColumn('week_seq', f.row_number().over(seq_window))

# 第二步：构建跨周偏移窗口（不把周/月纳入分区，让同业务维度、同一年份的所有记录落在同一窗口）
# 排序逻辑：先按周内序位升序，再按周次升序，保证同一序位的所有记录按周次排成连续序列
offset_window = (
    Window
    .partitionBy('store', 'state_prov_cd', 'department', 'item', 'year')
    .orderBy('week_seq', 'week')
)

# 第三步：取上一行的销售值，即为上一周同序位的销售数据
df_sales_shifted = df_base.withColumn(
    'sales_shifted', 
    f.lag('sales', 1).over(offset_window)
)

逻辑匹配说明：在构建好的跨周窗口内，记录会按周内序位分组排列，同一序位的所有记录按周次时间顺序连续排布：比如所有第1序位的记录排成「第1周第1天、第2周第1天、第3周第1天...」的序列，所有第2序位的记录排成「第1周第2天、第2周第2天、第3周第2天...」的序列。此时对每行做lag(1)操作，拿到的上一行必然是上一周同序位的销售值。
空值场景也完全符合需求规则：

• 第1周的所有记录是各自所属序位序列的首行，lag取值自然为空；
• 如果某一周的序位在上一周不存在（比如首周仅5天，没有第6、7序位的记录），那么这些序位的记录会成为所属序列的首行，lag取值自然为空，和描述的业务规则完全一致。

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