基于重复元素日期的分级排名：Rank/Dense_Rank遭日期干扰求解

你的需求是按ID分组，对连续相同的TORF值分配同一个LEVEL，每次TORF变化时LEVEL递增。直接用rank或dense_rank无法实现，因为这两个函数是基于字段值的排名，而非连续相同值的分组编号，日期字段的存在会让它们给每一行生成独立排名，干扰结果。

可以通过以下步骤实现：

1. 按ID分组，按DT排序（保证数据按日期顺序处理）
2. 判断当前行TORF与上一行是否不同，生成增量标记
3. 对增量标记做累加求和，得到最终的LEVEL

PySpark 代码示例 #

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

# 定义窗口：按ID分组，按日期排序
window_spec = Window.partitionBy("ID").orderBy("DT")

# 生成变化标记：首行标记为1，TORF与上一行不同时标记为1，否则为0
df_with_change = df.withColumn(
    "is_torf_change",
    F.when(F.lag("TORF").over(window_spec).isNull(), 1)
    .otherwise(F.when(F.col("TORF") != F.lag("TORF").over(window_spec), 1).otherwise(0))
)

# 累加变化标记得到LEVEL
result_df = df_with_change.withColumn(
    "LEVEL",
    F.sum("is_torf_change").over(window_spec.rangeBetween(Window.unboundedPreceding, Window.currentRow))
).drop("is_torf_change")

# 查看结果
result_df.show()

代码说明 #

• lag("TORF").over(window_spec)：获取当前行的上一行TORF值，首行没有上一行，返回null
• is_torf_change列：标记TORF是否发生变化，发生变化时为1，否则为0
• sum("is_torf_change").over(...)：对每个ID内的变化标记从首行到当前行累加，相同连续TORF的行累加结果一致，变化时累加值+1，正好对应需求的LEVEL

运行后得到的结果和你期望的完全一致。

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