Running Totals and Neighbours

The Daily Grind's owner reads the month's takings as a running total: not what each day took, but what the month had taken so far by that day. Last lesson a window gave each row a number from its order. This lesson the window gives each row a view of its neighbours, the row before, the row after, every row so far, and the one new idea is the frame: which of those rows the calculation can see.

Watch it happen

One store, four takings entries, two of them on the same day:

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

w = (
    Window.partitionBy("store")
    .orderBy("day")
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)
)

result = takings.withColumn("so_far", F.sum("amount").over(w))

sum is the same sum as before; .over(w) turns it from one number per store into one number per row. rowsBetween(unboundedPreceding, currentRow) is the frame: from the first row of the window partition up to and including this one. Each row adds itself to everything before it.

The frame you didn't write

Leave the frame off and the total still runs, until two rows share a day:

w = Window.partitionBy("store").orderBy("day")
takings.withColumn("so_far", F.sum("amount").over(w))

The first row shows 20, not 10. An ordered window with no frame of its own gets RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, and range reasons by value, not by position: every row with the current row's day is "the current row". Two entries on 1 June are one step. ROWS counts rows and takes them one at a time. When the ordering column can repeat, say rowsBetween and mean it.

The row before

lag reads a column from the previous row in the order; lead reads it from the next:

w = Window.partitionBy("store").orderBy("day")

result = takings.select(
    "day", "amount",
    F.lag("amount", 1).over(w).alias("prev"),
    (F.col("amount") - F.lag("amount", 1).over(w)).alias("change"),
)

The first row has no previous row, so lag is null and the change is null: there was no yesterday. That null is the truth, and the Null Is Not a Value rule applies: keep it, or fill it on purpose. F.lag("amount", 1, 0) fills it with 0 and the first day's change becomes 10, which reads as growth that never happened.

A window of the last three

A frame can look back a fixed number of rows. A three-day moving average is the current row and the two before it:

w = Window.partitionBy("store").orderBy("day").rowsBetween(-2, 0)
takings.withColumn("avg_3", F.avg("amount").over(w))

For the first two rows the frame is short, one row and then two, and the average is over what exists; that is correct, not an edge case to patch. rowsBetween counts rows, so a day the store was shut is simply not there and the frame reaches back to the last three entries. rangeBetween(-2, 0) on a numeric day would mean "days within two of this one" and reach a different set when a day is missing: with day 3 absent, at day 4 the rows frame averages three entries and the range frame two.

Where a session begins

Put the two ideas together and you can cut a stream of events into sessions. A session starts at a user's first event, and again whenever the gap from the previous event is more than thirty minutes:

w = Window.partitionBy("user_id").orderBy("ts")
prev = F.lag("ts").over(w)
starts = ((F.col("ts") - prev) > 30) | prev.isNull()

result = (
    events.withColumn("starts", starts.cast("int"))
    .withColumn("session", F.sum("starts").over(w.rowsBetween(Window.unboundedPreceding, Window.currentRow)))
)

lag finds the gap, a comparison marks the starts (the first row's gap is null, so it is marked by name), and a running total of the marks numbers the sessions. A gap of exactly 30 does not start one; 31 does. Every piece is from this lesson.

Four problems, in the order they build:

1. Day Over Day — lag, and the first row's null.
2. Running Revenue — the frame, with a repeated day.
3. Three-Day Moving Average — rowsBetween(-2, 0), with a missing day.
4. Sessionize the Streaks — lag, a mark, a running total.