rangeBetween

Frames: which rows the aggregate sees

A window function like sum(...).over(w) aggregates over a frame: a slice of the partition around the current row. You set it with rowsBetween(start, end) or rangeBetween(start, end), where negative means before, 0 means the current row, and Window.unboundedPreceding / unboundedFollowing mean the partition edges.

Step by step: a 7-day total

Store A has no sales rows on several days. We want, for each day, the total of the 7 calendar days ending that day.

On March 10 the last 7 calendar days are March 4 to 10, which contain only the 40 and the 50. The row frame grabs the previous 4 rows instead, reaching back to March 1, and over-counts.

Run the example

A range frame needs a numeric order key, so we turn the date into a day number first:

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

days = daily.withColumn("day_num", F.datediff("day", F.lit("1970-01-01")))
w = Window.partitionBy("store").orderBy("day_num")
result = days.select(
    "day", "sales",
    F.sum("sales").over(w.rowsBetween(-6, 0)).alias("last_7_rows"),
    F.sum("sales").over(w.rangeBetween(-6, 0)).alias("last_7_days"),
)

SELECT day, sales,
       SUM(sales) OVER (PARTITION BY store ORDER BY day
                        ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS last_7_rows,
       SUM(sales) OVER (PARTITION BY store ORDER BY CAST(day AS DATE)
                        RANGE BETWEEN INTERVAL 6 DAYS PRECEDING AND CURRENT ROW) AS last_7_days
FROM daily

In SQL you can order by a date and use an interval directly. In PySpark, convert to a number: days since epoch as above, or F.col("ts").cast("long") (seconds) for timestamps, with the frame in seconds, for example rangeBetween(-3600, 0) for the last hour.

The default frame trap

If a window has an orderBy but no frame, Spark uses RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: a running total. Because it is a range frame, rows that tie on the order value are all included together:

• sum("x").over(Window.partitionBy("k").orderBy("day")) gives a running total in which every row of the same day shows the same value, the total up to and including that day.
• Without orderBy, the default frame is the whole partition, so the same expression gives the group total on every row.
• last("x").over(w) with the default frame returns the current row (or its last tie), not the partition's last value. Use rowsBetween(unboundedPreceding, unboundedFollowing) for that.

Under the hood

Running frames (unbounded preceding to current row) are computed incrementally in one pass. Sliding frames are maintained by adding the row entering the frame and, where the function allows, removing the one leaving it. A frame like rangeBetween(-30, 0) on dense data still holds up to 30 days of rows per key in memory, so very wide frames on skewed keys can spill.

Common mistakes