如何在PySpark rowsBetween中使用unboundedPreceding等参数

核心工作原理 #

rowsBetween是PySpark窗口规范（WindowSpec）的边界定义方法，作用是给窗口函数划定按物理行相对位置计算的计算范围，和按排序列值范围划定边界的rangeBetween是两种最常用的窗口边界定义方式。
方法接收两个位置参数：start（窗口起始边界）、end（窗口结束边界），要求start对应的行位置不能晚于end对应的位置，否则运行时会抛出边界非法错误。
三个内置边界常量的含义：

• Window.unboundedPreceding：指代当前分区内排序后的第一行，作为起始参数时代表窗口从分区最开头启动
• Window.unboundedFollowing：指代当前分区内排序后的最后一行，作为结束参数时代表窗口到分区最末尾截止
• Window.currentRow：指代当前正在执行计算的行本身
  除了三个常量，也可以直接传入整数代表相对当前行的偏移：负数代表当前行之前N行，0代表当前行，正数代表当前行之后N行。

Window对象正确使用流程 #

窗口计算必须按固定步骤定义，否则很容易得到不符合预期的结果：

• 第一步：导入依赖，从pyspark.sql引入Window类，同时按需引入要用到的聚合/分析函数（比如sum/avg/rank/lag等）
• 第二步：定义分区与排序规则，通过Window.partitionBy(分组列)指定窗口拆分的分区维度，同一分区内的数据才会进入同一个窗口计算池；通过orderBy(排序列)指定分区内的行顺序，所有涉及相对位置的边界计算都基于这个排序后的行序列——只要使用了非全分区覆盖的边界，必须显式指定orderBy，否则行顺序随机，计算结果不可复现。
• 第三步：绑定行边界，在分区排序规则后链式调用rowsBetween(start, end)，得到最终可用的WindowSpec对象。注意如果不手动指定边界，Spark默认的窗口规则是rowsBetween(Window.unboundedPreceding, Window.currentRow)，也就是从分区第一行到当前行的累计范围，这是高频踩坑点：很多开发者误以为不写边界会计算全分区数据，实际默认只计算到当前行。
• 第四步：应用窗口，把定义好的WindowSpec对象传入函数的.over()方法，通过withColumn或者select把计算结果挂到DataFrame上即可。

实战代码示例 #

先构造测试用的销售数据集，后续所有示例都基于这个DataFrame运行：

from pyspark.sql import SparkSession
from pyspark.sql import Window
from pyspark.sql.functions import sum, avg, col

spark = SparkSession.builder.appName("rowsBetween_demo").getOrCreate()

# 测试数据：分区域的月度销售额
data = [
    ("华东", "2024-01", 120),
    ("华东", "2024-02", 150),
    ("华东", "2024-03", 90),
    ("华东", "2024-04", 200),
    ("华南", "2024-01", 80),
    ("华南", "2024-02", 110),
    ("华南", "2024-03", 130),
    ("华南", "2024-04", 170)
]
df = spark.createDataFrame(data, schema=["region", "month", "sales"])

示例1：计算分区内累计销售额 #

边界设置为分区第一行到当前行，和Spark默认窗口规则效果一致：

w_cum = Window.partitionBy("region").orderBy("month").rowsBetween(
    Window.unboundedPreceding, Window.currentRow
)
df.withColumn("cum_sales", sum("sales").over(w_cum)).show()

运行后每个区域的销售额会逐行累加，比如华东区域1月累计值为120、2月为270、3月为360、4月为560。

示例2：计算分区全量聚合值 #

边界设置为分区第一行到分区最后一行，等价于按分区聚合后把结果关联回原表，性能比join写法更高：

w_total = Window.partitionBy("region").orderBy("month").rowsBetween(
    Window.unboundedPreceding, Window.unboundedFollowing
)
df.withColumn("region_total_sales", sum("sales").over(w_total)).show()

运行后同区域的每一行都会带上该区域的总销售额，比如华东区域所有行的region_total_sales都是560。

示例3：计算滑动移动平均 #

边界用整数偏移设置，比如取当前行前后各1行（共3个月）的销售额平均值，计算3月移动均值：

w_moving = Window.partitionBy("region").orderBy("month").rowsBetween(-1, 1)
df.withColumn("moving_avg_3m", avg("sales").over(w_moving)).show()

分区首尾的行如果凑不足窗口范围，会自动按实际存在的行计算：比如华东1月前面没有数据，窗口只包含1、2月两行，平均值为(120+150)/2=135；2月窗口包含1、2、3月三行，平均值为120。

示例4：计算当前行到分区末尾的累计值 #

边界设置为当前行到分区最后一行，比如计算从当月到年末的剩余总销售额：

w_remaining = Window.partitionBy("region").orderBy("month").rowsBetween(
    Window.currentRow, Window.unboundedFollowing
)
df.withColumn("remaining_sales", sum("sales").over(w_remaining)).show()

运行后华东1月的剩余销售额为全区域总和560，2月为440，3月为290，4月为200。

踩坑提醒：rowsBetween是严格按物理行位置取数，不会自动处理排序列的重复值，如果排序列有重复值，建议同时把唯一标识列加入orderBy保证行顺序唯一，避免窗口范围不符合预期。

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