Spark中按时间顺序基于历史值对指定列计算排名的实现问题

你原来的实现错误是因为窗口函数rank()的排名规则完全由窗口定义中的orderBy字段决定，你的窗口按ts排序，所以最终是按时间先后排名，和value列完全无关，因此不符合需求。

我们可以先在按时间扩展的滑动窗口内收集当前分组所有历史去重的value值，排序后查找当前value的位置即可得到目标排名，完整代码如下：

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

# 窗口定义和你原有逻辑一致：按group分区，按ts升序，取截止当前行的所有历史数据
w = (
    Window
    .partitionBy("group")
    .orderBy("ts")
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)
)

# 先收集窗口内去重的value列表，排序后查找当前value的位置（Spark中array_position从1开始计数，正好符合排名规则）
df.withColumn(
    "sorted_hist_values",
    f.array_sort(f.collect_set("value").over(w))
).withColumn(
    "rank",
    f.array_position(f.col("sorted_hist_values"), f.col("value"))
).drop("sorted_hist_values").show()

运行上述代码后输出和你预期完全一致：

+-------------------+-----+-----+----+
|                 ts|value|group|rank|
+-------------------+-----+-----+----+
|2021-05-01 10:30:00| 2.15|    a|   1|
|2021-05-01 10:30:10| 2.12|    a|   1|
|2021-05-01 10:30:20| 2.13|    a|   2|
|2021-05-01 10:30:50| 2.14|    a|   3|
|2021-05-01 10:31:05| 2.13|    a|   2|
|2021-05-01 10:31:10| 2.16|    a|   5|
|2021-05-01 10:31:10| 2.16|    b|   1|
+-------------------+-----+-----+----+

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