Pyspark分组排序后取列前两个不同值对应最早日志日期的实现问题

你遇到的问题核心是直接按state+case_count分组会合并不同时间周期出现的相同count，导致旧周期的记录拉低了整体的最小日期，本质属于连续相同值分组（岛屿问题），正确实现逻辑如下：

实现思路 #

1. 先对每个state下的记录按log_date降序排序，标记相邻行count发生变化的位置
2. 对变化标记做累加，生成每个连续相同count段的分组ID，把不同时间周期出现的相同count拆分为独立分组
3. 按分组聚合得到每个连续段的count值、段内最早/最晚日志日期
4. 按段的最晚日期倒序取前2个连续段，输出对应结果即可

完整可运行代码 #

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

# 重命名count避免关键字冲突
df = df.withColumnRenamed("count", "case_count")

# 定义窗口：按state分组，日志日期降序排序
win_state_order = Window.partitionBy("state").orderBy(F.col("log_date").desc())

# 标记相邻行count是否变化：变化打1，不变打0
df = df.withColumn("count_diff", 
    F.when(F.col("case_count") == F.lag("case_count").over(win_state_order), 0).otherwise(1)
)

# 对diff标记累加，生成连续段的分组ID
df = df.withColumn("group_id", F.sum("count_diff").over(win_state_order.rangeBetween(Window.unboundedPreceding, 0)))

# 按state+group_id聚合，得到每个连续段的最早/最晚日期
win_group = Window.partitionBy("state", "group_id")
df = df.withColumn("group_min_date", F.min("log_date").over(win_group)) \
       .withColumn("group_max_date", F.max("log_date").over(win_group))

# 去重后按段的最新日期倒序，取前2个最新的连续count段
df = df.select("state", "case_count", "group_min_date", "group_max_date").distinct() \
       .withColumn("rank", F.rank().over(Window.partitionBy("state").orderBy(F.col("group_max_date").desc()))) \
       .filter(F.col("rank") <= 2)

# 输出对齐预期字段
result = df.select("state", "case_count", F.col("group_min_date").alias("log_date"))
result.show()

运行结果 #

和你给出的预期输出完全一致：

+-----+----------+----------+
|state|case_count|  log_date|
+-----+----------+----------+
|   GU|      7402|2021-07-16|
|   GU|      7397|2021-07-13|
+-----+----------+----------+

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