排名函数应用：如何重置连续时间序列下的销售分组计数

错误根因 #

你当前的实现存在两个核心问题导致分组不符合要求：

1. 时间断层判断仅覆盖了同年场景，未处理跨年周连续的情况（如2023年第52周后接2024年第1周属于连续周，你的逻辑会误判为无断层）
2. 用dense_rank对价格字段+时间分组排序的逻辑，会将不同时间区间内出现的相同价格组合归为同一分组，刚好触发了你遇到的「序列打断后相同属性行被分到旧分组」的问题。

正确实现思路 #

将「时间断层」和「属性变更」两种场景都作为分组切换的触发条件，按时间顺序累加触发标记，累加值即为唯一分组ID，天然满足序列打断后需新建分组的要求。

完整可运行代码 #

from pyspark.sql import Window
from pyspark.sql.functions import col, lag, when, sum, to_date, concat

# 1. 先将年周转换为对应周一的日期，用于精准判断周连续（支持跨年）
# 如果你用的是非ISO周，可调整日期格式匹配你的周规则
df_with_date = initial_df.withColumn(
    "WeekStartDate", 
    to_date(concat(col("Period_Year"), col("Period_Week")), "yyyyw")
)

# 2. 定义窗口：按Product、Price_Shop_Group分区，按周起始日期排序
window_spec = Window.partitionBy("Price_Shop_Group", "Product").orderBy("WeekStartDate")

# 3. 取前一行的价格字段与周起始日期
df_with_lag = df_with_date \
    .withColumn("Prev_Base_Price", lag("Base_Price").over(window_spec)) \
    .withColumn("Prev_Unit_Price", lag("Unit_Price").over(window_spec)) \
    .withColumn("Prev_WeekStart", lag("WeekStartDate").over(window_spec))

# 4. 生成分组切换标记：满足两个条件任意一个就触发新分组
# 条件1：周不连续（当前周与上一周间隔超过7天）
# 条件2：Base_Price或Unit_Price发生变化（同分区下Product和Price_Shop_Group已固定，无需判断）
df_with_flag = df_with_lag.withColumn(
    "NewGroupFlag",
    when(
        (col("Prev_WeekStart").isNull())
        | (col("WeekStartDate") - col("Prev_WeekStart") > 7)
        | (col("Base_Price") != col("Prev_Base_Price"))
        | (col("Unit_Price") != col("Prev_Unit_Price")),
        1
    ).otherwise(0)
)

# 5. 累加标记值得到最终分组ID
window_cum = Window.partitionBy("Price_Shop_Group", "Product").orderBy("WeekStartDate").rowsBetween(Window.unboundedPreceding, 0)
final_df = df_with_flag.withColumn("ProductGroup", sum("NewGroupFlag").over(window_cum))

逻辑说明 #

该实现中分组ID是按时间顺序累加生成的，即使后续行的价格属性与历史分组完全一致，只要中间触发过一次新分组标记（时间断层或属性变更），累加值就会与历史分组不同，完全符合你的分组规则要求。

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