A running revenue total per store

Description

Lesson 5 introduced Window.partitionBy("store_id").orderBy("order_ts") as syntax, over Kiosko's forty real rows. This lesson applies it to the complete business question — "what's each store's running revenue total, order by order, as they arrive in time?" — and takes it beyond the syntax: first, confirming the result by hand over the forty real rows; then, running that exact same window over fact_orders_at_scale, ten million rows, where a new phenomenon shows up that the forty rows never reveal — ties in the order column, and what Spark does with them by default.

Connection to the module. This lesson builds directly on lesson 5's syntax, with no new Window concept — what's new here is the evidence at scale, and a design decision (the tiebreak) that only becomes visible when the dataset is large enough to have repeated values in order_ts.

An analogy: the runner, and the synchronized clocks of thousands of identical runners

Pick back up lesson 5's marathon analogy. Now imagine that, instead of a single runner per category, the marathon has a strange experiment: two hundred fifty thousand runners, all with exactly the same race plan — the same pace, the same checkpoints, at exactly the same clock instants. If you ask the timing system "what's the category's total cumulative time, at the instant the first group of runners crosses kilometer 3?", the correct answer has to include every runner crossing that point at that exact instant — it can't leave some of them "ahead" of others who arrived at literally the same time. This lesson runs into exactly that phenomenon: kiosko_orders_at_scale replicates the same week 250,000 times, so, for every store, the same real order_ts shows up repeated 250,000 times — once per franchise. When that happens, what does Window.orderBy("order_ts") do with all those tied rows?

Worked example, part 1: the running total, verified by hand over the 40 rows

Step 1 — Lesson 5's window, applied to the three stores

# running_total_40.py
import glob
from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType,
)

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").getOrCreate()

orders_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("store_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("unit_price", DoubleType(), False),
    StructField("order_ts", TimestampType(), False),
])
orders_df = spark.read.csv(sorted(glob.glob("orders_2026-08-*.csv")), schema=orders_schema, header=True, enforceSchema=False)
fact_orders_df = orders_df.withColumn("revenue", F.col("quantity") * F.col("unit_price"))

store_window = Window.partitionBy("store_id").orderBy("order_ts")
with_running = fact_orders_df.withColumn(
    "running_total", F.round(F.sum("revenue").over(store_window), 2)
)

print("=== running_total per store, Kiosko's 40 real rows ===")
for store in ["S01", "S02", "S03"]:
    print(f"--- {store} ---")
    with_running.filter(F.col("store_id") == store).orderBy("order_ts").select(
        "order_id", "order_ts", F.round("revenue", 2).alias("revenue"), "running_total"
    ).show(20, truncate=False)

Step 2 — The final value, verified against the totals you already know

final_by_store = {
    r["store_id"]: r["max_running"]
    for r in with_running.groupBy("store_id").agg(F.max("running_total").alias("max_running")).collect()
}
print(f"final running_total per store = {final_by_store}")
assert final_by_store == {"S01": 38.3, "S02": 38.8, "S03": 29.05}
print("Verification: final running_total matches the known total per store -> OK")

spark.stop()

What to expect. Running python3 running_total_40.py, the output is exactly this (executed in this run):

=== running_total per store, Kiosko's 40 real rows ===
--- S01 ---
+--------+-------------------+-------+-------------+
|order_id|order_ts           |revenue|running_total|
+--------+-------------------+-------+-------------+
|ORD-1001|2026-08-03 08:14:00|1.65   |1.65         |
|ORD-1002|2026-08-03 08:20:00|1.2    |2.85         |
|ORD-1004|2026-08-03 09:02:00|4.5    |7.35         |
|ORD-1008|2026-08-03 10:22:00|1.1    |8.45         |
|ORD-2001|2026-08-04 08:05:00|1.2    |9.65         |
|ORD-2004|2026-08-04 09:50:00|2.25   |11.9         |
|ORD-3002|2026-08-05 08:22:00|0.55   |12.45        |
|ORD-4001|2026-08-06 08:10:00|2.2    |14.65        |
|ORD-4004|2026-08-06 09:55:00|4.5    |19.15        |
|ORD-5001|2026-08-07 08:05:00|2.4    |21.55        |
|ORD-5004|2026-08-07 09:22:00|1.5    |23.05        |
|ORD-5007|2026-08-07 10:40:00|1.1    |24.15        |
|ORD-6001|2026-08-08 08:00:00|3.3    |27.45        |
|ORD-6004|2026-08-08 08:52:00|9.0    |36.45        |
|ORD-6007|2026-08-08 09:45:00|0.75   |37.2         |
|ORD-7001|2026-08-09 09:15:00|1.1    |38.3         |
+--------+-------------------+-------+-------------+

--- S02 ---
+--------+-------------------+-------+-------------+
|order_id|order_ts           |revenue|running_total|
+--------+-------------------+-------+-------------+
|ORD-1003|2026-08-03 08:31:00|1.5    |1.5          |
|ORD-1006|2026-08-03 09:47:00|2.4    |3.9          |
|ORD-2002|2026-08-04 08:40:00|2.2    |6.1          |
|ORD-2005|2026-08-04 10:15:00|2.4    |8.5          |
|ORD-3001|2026-08-05 08:10:00|9.0    |17.5         |
|ORD-4002|2026-08-06 08:45:00|1.5    |19.0         |
|ORD-4005|2026-08-06 10:30:00|1.65   |20.65        |
|ORD-5003|2026-08-07 08:58:00|4.5    |25.15        |
|ORD-5006|2026-08-07 10:15:00|2.75   |27.9         |
|ORD-6002|2026-08-08 08:18:00|3.6    |31.5         |
|ORD-6005|2026-08-08 09:10:00|2.25   |33.75        |
|ORD-6008|2026-08-08 10:02:00|3.85   |37.6         |
|ORD-7002|2026-08-09 09:40:00|1.2    |38.8         |
+--------+-------------------+-------+-------------+

--- S03 ---
+--------+-------------------+-------+-------------+
|order_id|order_ts           |revenue|running_total|
+--------+-------------------+-------+-------------+
|ORD-1005|2026-08-03 09:15:00|2.75   |2.75         |
|ORD-1007|2026-08-03 10:05:00|0.75   |3.5          |
|ORD-2003|2026-08-04 09:12:00|4.5    |8.0          |
|ORD-2006|2026-08-04 10:33:00|3.3    |11.3         |
|ORD-4003|2026-08-06 09:20:00|1.2    |12.5         |
|ORD-5002|2026-08-07 08:30:00|2.2    |14.7         |
|ORD-5005|2026-08-07 09:47:00|3.6    |18.3         |
|ORD-6003|2026-08-08 08:35:00|2.2    |20.5         |
|ORD-6006|2026-08-08 09:28:00|2.4    |22.9         |
|ORD-6009|2026-08-08 10:20:00|4.5    |27.4         |
|ORD-7003|2026-08-09 10:05:00|1.65   |29.05        |
+--------+-------------------+-------+-------------+

final running_total per store = {'S01': 38.3, 'S02': 38.8, 'S03': 29.05}
Verification: final running_total matches the known total per store -> OK

Check any row by hand: for S03, ORD-1005 has revenue = 2.75 (5 × 0.55) and it's that store's earliest order by date, so its running_total is 2.75. ORD-1007 (revenue = 0.75) arrives next, and its running total is 2.75 + 0.75 = 3.5. And each store's final value — 38.3, 38.8, 29.05 — is exactly the same total you already know from module 1: the accumulated revenue, on its last row, always matches the aggregate total a groupBy would have given directly.

Worked example, part 2: the same calculation, at scale — and the tie phenomenon

Step 1 — Without a tiebreak: what happens when order_ts repeats 250,000 times

# running_total_scale.py
from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType,
)

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").getOrCreate()

scale_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("franchise_id", IntegerType(), False),
    StructField("store_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("unit_price", DoubleType(), False),
    StructField("order_ts", TimestampType(), False),
])
orders_at_scale_df = spark.read.csv(
    "kiosko_orders_at_scale.csv", schema=scale_schema, header=True, enforceSchema=False,
)
fact_orders_at_scale_df = orders_at_scale_df.withColumn("revenue", F.col("quantity") * F.col("unit_price"))

tie_window = Window.partitionBy("store_id").orderBy("order_ts")
with_ties = fact_orders_at_scale_df.withColumn(
    "running_total", F.round(F.sum("revenue").over(tie_window), 2)
)
print("S01, first 6 distinct order_ts values -- how many rows and running_total value for each:")
sample = (
    with_ties.filter(F.col("store_id") == "S01")
    .groupBy("order_ts")
    .agg(F.count("*").alias("rows_with_this_ts"), F.first("running_total").alias("running_total"))
    .orderBy("order_ts")
)
sample.show(6, truncate=False)

What to expect (executed in this run, over the complete fact_orders_at_scale):

S01, first 6 distinct order_ts values -- how many rows and running_total value for each:
+-------------------+-----------------+-------------+
|order_ts           |rows_with_this_ts|running_total|
+-------------------+-----------------+-------------+
|2026-08-03 08:14:00|250000           |412500.0     |
|2026-08-03 08:20:00|250000           |712500.0     |
|2026-08-03 09:02:00|250000           |1837500.0    |
|2026-08-03 10:22:00|250000           |2112500.0    |
|2026-08-04 08:05:00|250000           |2412500.0    |
|2026-08-04 09:50:00|250000           |2975000.0    |
+-------------------+-----------------+-------------+

There's the phenomenon: each of S01's first six distinct order_ts values has exactly 250,000 rows — one per franchise — and the 250,000 rows sharing one order_ts all share the same running_total. Check the first by hand: 250,000 × 1.65 = 412,500.0 — ORD-1001's revenue (1.65, already confirmed in part 1), multiplied by the 250,000 franchises sharing that exact instant. The second, 712,500.0, is 412,500.0 + (250,000 × 1.20) = 412,500.0 + 300,000.0. This isn't an error — it's the documented default behavior of a window with orderBy: when several rows tie on the order value, they all get the same accumulated result, computed as if they formed a single block up to that point — never an arbitrary, distinct order among them.

Step 2 — With a tiebreak: a fully deterministic order

tiebreak_window = Window.partitionBy("store_id").orderBy("order_ts", "franchise_id", "order_id")
with_tiebreak = fact_orders_at_scale_df.withColumn(
    "running_total", F.round(F.sum("revenue").over(tiebreak_window), 2)
)
print("\nS01, first 6 rows sorted by (order_ts, franchise_id, order_id):")
with_tiebreak.filter(F.col("store_id") == "S01").orderBy("order_ts", "franchise_id", "order_id").select(
    "order_id", "franchise_id", "order_ts", F.round("revenue", 2).alias("revenue"), "running_total"
).show(6, truncate=False)

final_by_store = {
    r["store_id"]: r["max_running"]
    for r in with_tiebreak.groupBy("store_id").agg(F.max("running_total").alias("max_running")).collect()
}
print(f"\nfinal running_total per store (with tiebreaker) = {final_by_store}")
assert final_by_store == {"S01": 9575000.0, "S02": 9700000.0, "S03": 7262500.0}
print("Verification: the at-scale final running_total matches the known per-store breakdown -> OK")

spark.stop()

What to expect (executed in this run):

S01, first 6 rows sorted by (order_ts, franchise_id, order_id):
+----------------+------------+-------------------+-------+-------------+
|order_id        |franchise_id|order_ts           |revenue|running_total|
+----------------+------------+-------------------+-------+-------------+
|F000000-ORD-1001|0           |2026-08-03 08:14:00|1.65   |1.65         |
|F000001-ORD-1001|1           |2026-08-03 08:14:00|1.65   |3.3          |
|F000002-ORD-1001|2           |2026-08-03 08:14:00|1.65   |4.95         |
|F000003-ORD-1001|3           |2026-08-03 08:14:00|1.65   |6.6          |
|F000004-ORD-1001|4           |2026-08-03 08:14:00|1.65   |8.25         |
|F000005-ORD-1001|5           |2026-08-03 08:14:00|1.65   |9.9          |
+----------------+------------+-------------------+-------+-------------+

final running_total per store (with tiebreaker) = {'S02': 9700000.0, 'S01': 9575000.0, 'S03': 7262500.0}
Verification: the at-scale final running_total matches the known per-store breakdown -> OK

With franchise_id and order_id added as a tiebreak, every one of the 250,000 rows sharing the same order_ts gets its own running_total, incrementing by 1.65 each time — that specific order's revenue — instead of jumping all at once in blocks of 250,000. And the final value — 9,575,000.0 for S01, 9,700,000.0 for S02, 7,262,500.0 for S03 — is exactly the same per-store breakdown module 4 already verified with assert against the raw data: the accumulated revenue, on its last row, still matches the aggregate total, whether or not the orderBy has a tiebreak — the only thing that changes is how that running total gets distributed among the intermediate rows.

Diagram: without a tiebreak versus with a tiebreak

flowchart TD
    subgraph SinDesempate["orderBy('order_ts') -- NO tiebreak"]
        A1["250,000 rows with\norder_ts = 08:14:00"] --> A2["ALL receive\nrunning_total = 412,500.0"]
    end

    subgraph ConDesempate["orderBy('order_ts', 'franchise_id', 'order_id') -- WITH tiebreak"]
        B1["Row franchise_id=0"] --> B2["running_total = 1.65"]
        B3["Row franchise_id=1"] --> B4["running_total = 3.30"]
        B5["Row franchise_id=2"] --> B6["running_total = 4.95"]
        B2 -.-> B3
        B4 -.-> B5
    end

    style A2 fill:#f96,stroke:#333
    style B2 fill:#9c6,stroke:#333
    style B4 fill:#9c6,stroke:#333
    style B6 fill:#9c6,stroke:#333

Going deeper: why the default frame groups ties together, instead of ordering them arbitrarily

It's worth understanding the design reasoning, not just memorizing the behavior. A window's default frame with orderBy — technically, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW — defines "up to the current row" in terms of the order column's value, not in terms of a row's physical position. This means "up to the current row" actually gets interpreted as "up to every row whose order value is less than or equal to this row's" — and if two rows share exactly the same value, both are, by definition, "up to that same point" relative to each other. Spark has no basis for arbitrarily deciding that a row with order_ts = 08:14:00 comes "before" another row with that exact same order_ts — doing so would mean inventing an order the data doesn't specify. The alternative — a frame based on a row's physical position (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, which does distinguish between tied rows by their physical position within the partition) exists in Spark's API, but it produces a result that depends on the rows' internal physical order, not on any explicit business criterion — exactly the kind of non-deterministic behavior this guide avoids.

The correct solution, and the one part 2 of this lesson uses, isn't to change the frame type — it's to add additional columns to the orderBy that really are unique within the group, so ties stop existing altogether. franchise_id (250,000 distinct values) and order_id (unique even within a single franchise) together guarantee no (store_id, order_ts, franchise_id, order_id) combination repeats — the same "add a key that disambiguates" principle you already used in module 4 when building order_id with the F{franchise_id:06d}- prefix.

Common mistakes

Reading the jump from 412,500.0 to 712,500.0 as a calculation error. What happens: someone, seeing running_total jump from 412,500.0 to 712,500.0 with no intermediate values, suspects the window is summing incorrectly, or duplicating rows. Why it happens: on Kiosko's forty real rows (this lesson's part 1), running_total always grows one order at a time — never in large blocks — so a large jump at scale can look, at first glance, like a bug. How to spot it: if you suspect an error, divide the jump by the number of tied rows ((712,500.0 - 412,500.0) / 250,000 = 1.2) and compare it against the corresponding order's revenue (ORD-1002, revenue = 1.20) — if they match, the calculation is correct. How to fix it: it isn't an error — it's the documented behavior of an orderBy with ties, explained in this lesson's "going deeper" section. Before suspecting a bug, group by order_ts's value and count how many rows share that value, exactly as step 1 of the worked example does.

Using the tiebreak in only some of the pipeline's queries, and not others that also accumulate by the same column. What happens: someone adds franchise_id and order_id to a window's orderBy in one script, but forgets to do it in another window in the same pipeline that also uses order_ts to sort within store_id, producing inconsistent results between the two parts of the pipeline. Why it happens: the tiebreak is easy to remember the first time the problem gets discovered, but easy to forget the second or third place where the same discipline is needed. How to spot it: if your pipeline has more than one window sorted by order_ts over fact_orders_at_scale, check that all of them use exactly the same tiebreak criterion — inconsistencies between windows in the same pipeline are hard to catch afterward, because each one, on its own, still looks "correct" in its own context. How to fix it: define the window specification once, as a reusable variable (the way store_window does in this guide), and reuse it across every query in the same pipeline that needs the same order — avoid rewriting the orderBy from scratch every time.

Assuming the tiebreak changes the running total's final value. What happens: someone, seeing how much the intermediate behavior changes between part 1 and part 2 of the worked example, expects the final value — the running total's last row — to also differ between the two versions. Why it happens: such a visible change in intermediate behavior intuitively suggests the complete result should change too. How to spot it: compare both parts of this lesson's assert — both verify the same final per-store breakdown (9,575,000.0, 9,700,000.0, 7,262,500.0), whether or not there's a tiebreak. How to fix it: the tiebreak changes how the running total gets distributed among rows in the same tie group, but it doesn't change the total sum for the whole group — any partition's last row, sorted any consistent way, always ends up summing that partition's complete total. The tiebreak matters for the intermediate detail, not for the final result.

Exercises

Exercise 1 — Calculate, by hand, the expected running_total for S01's third distinct date (2026-08-03 09:02:00), without a tiebreak. Using ORD-1004's revenue (4.50, already confirmed in part 1) and the worked example table's previous value (712,500.0), calculate by hand what running_total should be for that order_ts, and verify your calculation against the table already shown.

revenue_ord_1004 = 4.50
previous_running = 712_500.0
num_franchises = 250_000

expected_running = previous_running + (revenue_ord_1004 * num_franchises)
print(f"expected running_total for 2026-08-03 09:02:00 = {expected_running}")
assert expected_running == 1_837_500.0
print("Verification: matches the worked example's table -> OK")

expected running_total for 2026-08-03 09:02:00 = 1837500.0
Verification: matches the worked example's table -> OK

Exercise 2 — Confirm S03 also reaches its known total, with the tiebreak applied. Filter with_tiebreak to store_id == "S03", sort by order_ts, franchise_id, order_id, and confirm with assert that the last row has running_total == 7_262_500.0.

last_s03 = (
    with_tiebreak.filter(F.col("store_id") == "S03")
    .orderBy(F.desc("order_ts"), F.desc("franchise_id"), F.desc("order_id"))
    .first()
)
print(f"S03, last row: order_id={last_s03['order_id']}, running_total={last_s03['running_total']}")
assert last_s03["running_total"] == 7_262_500.0
print("Verification: S03 reaches its known total on the last row -> OK")

S03, last row: order_id=F249999-ORD-7003, running_total=7262500.0
Verification: S03 reaches its known total on the last row -> OK

Exercise 3 — Explain, without code, why adding order_id to the tiebreak, on top of franchise_id, is necessary even though franchise_id is already unique per franchise. In 2-3 sentences, explain why Window.partitionBy("store_id").orderBy("order_ts", "franchise_id") (without order_id) still wouldn't be a complete tiebreak.

Summary and next step

This lesson applied Window.partitionBy("store_id").orderBy("order_ts") to the complete running-revenue-per-store question, verified first by hand over Kiosko's forty real rows (S01=38.3, S02=38.8, S03=29.05), and then over the complete fact_orders_at_scale. At scale, you discovered with real evidence a phenomenon the small dataset never reveals: ties in order_ts — 250,000 rows per instant, one per franchise — make the running total jump in blocks, unless you add tiebreak columns (franchise_id, order_id) to the orderBy. With the tiebreak, you confirmed each store's final value — 9,575,000.0, 9,700,000.0, 7,262,500.0 — exactly matches the breakdown module 4 already verified.

Before moving on you should be able to: calculate by hand the accumulated revenue for any of Kiosko's forty real rows; explain why a window with orderBy and tied values produces the same result for every tied row, by default; and explain how an explicit tiebreak resolves that behavior without changing the final total.

Lesson 7 introduces this module's second window function, F.row_number(), to answer a different question: not a running total, but a ranking — the top product for each store, each day.

Resources

• PySpark — pyspark.sql.Window (the rangeBetween/rowsBetween reference and a window's default frame with orderBy, the foundation for this lesson's "going deeper" section). spark.apache.org/docs/latest/api/python/reference/pyspark.sql/window.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — this lesson's exact specification: revenue accumulated per store, verified over 40 rows and over the complete fact_orders_at_scale.
• data-engineering-foundations-guide's DESIGN doc — the original source of the per-store totals (S01=38.3, S02=38.8, S03=29.05) this lesson confirms as the running total's final value. src/guides/data-engineering-foundations-guide/DISENO.md