Spark基于多列lag条件识别字段变动生成row_number的实现问题

实现方案 #

你需要的是连续相同字段组内的序列生成，直接用row_number()、dense_rank()按三个字段分区无法满足需求，因为普通分区会把所有同值行归为同一组，不区分是否连续。我们可以通过「打变化标记→累加生成组ID→组内生成序号」的逻辑实现，完整代码如下：

from pyspark.sql import Window
from pyspark.sql.functions import col, lag, sum, row_number, when, monotonically_increasing_id

# 先添加自增ID作为排序依据，保证行顺序和输入一致，避免分布式环境下顺序错乱
df = df.withColumn("sort_id", monotonically_increasing_id())

# 1. 定义基础窗口：按account分区，按输入顺序排序，用于对比相邻行的字段差异
base_window = Window.partitionBy("account").orderBy("sort_id")

# 2. 标记当前行和上一行的三个字段是否发生变化，变化则打1，第一行默认打1
df = df.withColumn("is_change", 
    when(
        (lag("account", 1).over(base_window) != col("account")) |
        (lag("name", 1).over(base_window) != col("name")) |
        (lag("type", 1).over(base_window) != col("type")) |
        (lag("account", 1).over(base_window).isNull()),
        1
    ).otherwise(0)
)

# 3. 对变化标记做累加，相同连续组的累加值相同，即为组ID
group_window = Window.partitionBy("account").orderBy("sort_id").rowsBetween(Window.unboundedPreceding, 0)
df = df.withColumn("group_id", sum("is_change").over(group_window))

# 4. 按组ID分区，组内按顺序生成行号，即为预期结果
rn_window = Window.partitionBy("account", "group_id").orderBy("sort_id")
df = df.withColumn("row_number", row_number().over(rn_window))

# 输出结果，可删除中间字段
df.select("Event_date", "account", "name", "type", "row_number").show()

运行后输出结果和你给出的预期完全匹配。

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