Spark窗口范围聚合：按窗口起始行间隔取数的聚合函数实现

我明白你要解决的问题了——你想在Spark的窗口聚合里，对窗口内每隔N行的数据执行sum这类聚合操作，而且这个“隔行规则”必须以窗口的起始行为基准（窗口首行始终要包含在内）。你之前尝试用datetime取模的方式近似实现，但这个逻辑是基于全局时间周期的，并非以窗口的起始点为参照，所以没法满足需求。

下面是针对这个需求的完整解决方案，我们通过计算窗口内每行相对于起始行的偏移量来精准筛选目标数据：

实现步骤 #

核心思路是先给每个分区内的行分配排序后的行号，然后在窗口范围内确定起始行的行号，计算当前行与起始行的偏移量，最后只对偏移量符合offset % N == 0的行做聚合。

1. 定义基础窗口并添加行号 #

首先为每个symbol分区按datetime排序，生成每行的行号：

val df = // 你的输入DataFrame {symbol, datetime, metric}
val baseWin = Window.partitionBy("symbol").orderBy("datetime")
val dfWithRowNum = df.withColumn("row_num", row_number().over(baseWin))

2. 计算窗口起始行的行号 #

在目标窗口范围内（这里以rowsBetween(-12, 0)为例），用first_value获取窗口首行的行号：

val targetWin = baseWin.rowsBetween(-12, 0)
val dfWithWinStart = dfWithRowNum.withColumn("win_start_row", first_value("row_num").over(targetWin))

3. 计算相对于窗口起始行的偏移量 #

用当前行的行号减去窗口起始行的行号，得到偏移量：

val dfWithOffset = dfWithWinStart.withColumn("offset_from_start", col("row_num") - col("win_start_row"))

4. 执行隔行聚合 #

假设我们要每隔3行取一个数据（包含起始行），筛选offset_from_start % 3 == 0的行，然后对这些行的metric求和：

val N = 3 // 你可以根据需求修改这个值
val result = dfWithOffset.withColumn(
  "sum_every_N_rows",
  sum(when(col("offset_from_start") % N === 0, col("metric")).otherwise(lit(null))).over(targetWin)
)

为什么这个方案可行？ #

• offset_from_start是当前行与窗口首行的行号差，窗口首行的偏移量为0，必然满足offset % N == 0，确保首行始终被包含。
• 后续符合条件的行偏移量为N、2N...，严格以窗口起始行为基准每隔N行选取，完全符合你的需求。
• 不管窗口是固定行数范围（rowsBetween）还是时间范围（rangeBetween），这个逻辑都适用，因为行号是基于分区内排序生成的，偏移量计算不受窗口范围类型影响。

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