Ranking top products per store per day

Description

Lesson 6 used a window function to accumulate — a value that grows row by row. This lesson uses a different window function, F.row_number(), to rank within a group — the same class of question you already solved with arrays in data-modeling-for-analytics-guide: "which product did each store sell the most of, each day?". The answer needs two steps, not one: first aggregate revenue by product within each store-and-day combination (something you already know how to do with groupBy), and then rank those aggregated products within each group, keeping only the first one — the step that needs a window function, because a groupBy has no notion of "the first" within a group.

Connection to the module. This lesson closes out this module's window-function build-up: lesson 5 gave you the syntax, lesson 6 applied it to a running total, this lesson applies it to a ranking. Lesson 8 integrates all three pieces — join criterion, running total, ranking — into a single pipeline over the complete fact_orders_at_scale.

An analogy: each category's podium, at every checkpoint

Pick back up the marathon analogy. At every checkpoint, a timing system can show, for each age category, who's currently first — not each individual runner's cumulative time (lesson 6 already solved that), but their relative position within their own category. That position depends on comparing every runner against the others in their same category, sorted by their time, and keeping the order number that corresponds to each one. F.row_number() does exactly that: within each partition (Window.partitionBy("store_id", "order_day")), it sorts the rows by some criterion (orderBy(F.desc("total_revenue"))) and assigns each one a position number — 1 for the first, 2 for the second, and so on. Keeping position 1 from every group is, precisely, "the top product for each store, each day."

Worked example, part 1: aggregation and ranking, over the 40 real rows

Step 1 — Aggregate revenue by product, within each store and each day

# ranking_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"))
fact_orders_df = fact_orders_df.withColumn("order_day", F.to_date("order_ts"))

revenue_by_product_day = (
    fact_orders_df
    .groupBy("store_id", "order_day", "product_id")
    .agg(F.round(F.sum("revenue"), 2).alias("total_revenue"))
)

Notice this first step is exactly the groupBy you already know from module 3 — no window function yet. Its result no longer has forty rows: it has one row for every distinct (store_id, order_day, product_id) combination that shows up in the data — an intermediate table, smaller than fact_orders_df, but still not ranked.

Step 2 — The ranking window, with F.row_number()

rank_window = Window.partitionBy("store_id", "order_day").orderBy(F.desc("total_revenue"), "product_id")
ranked = revenue_by_product_day.withColumn("rank", F.row_number().over(rank_window))

print("=== ranked, S01, every day, sorted by day and rank ===")
ranked.filter(F.col("store_id") == "S01").orderBy("order_day", "rank").show(30, truncate=False)

What to expect. Running python3 ranking_40.py, the output is exactly this (executed in this run, over Kiosko's forty real rows):

=== ranked, S01, every day, sorted by day and rank ===
+--------+----------+----------+-------------+----+
|store_id|order_day |product_id|total_revenue|rank|
+--------+----------+----------+-------------+----+
|S01     |2026-08-03|P004      |4.5          |1   |
|S01     |2026-08-03|P001      |2.75         |2   |
|S01     |2026-08-03|P002      |1.2          |3   |
|S01     |2026-08-04|P003      |2.25         |1   |
|S01     |2026-08-04|P002      |1.2          |2   |
|S01     |2026-08-05|P001      |0.55         |1   |
|S01     |2026-08-06|P004      |4.5          |1   |
|S01     |2026-08-06|P001      |2.2          |2   |
|S01     |2026-08-07|P002      |2.4          |1   |
|S01     |2026-08-07|P003      |1.5          |2   |
|S01     |2026-08-07|P001      |1.1          |3   |
|S01     |2026-08-08|P004      |9.0          |1   |
|S01     |2026-08-08|P001      |3.3          |2   |
|S01     |2026-08-08|P003      |0.75         |3   |
|S01     |2026-08-09|P001      |1.1          |1   |
+--------+----------+----------+-------------+----+

Every group (store_id, order_day) starts its own ranking at 1 — notice S01 on 2026-08-03 has three products ranked 1, 2, 3, and 2026-08-04 also starts at 1, not continuing at 4. That reset, group by group, is exactly what partitionBy does — the same principle you already saw with lesson 6's running total, now applied to a position number instead of a sum.

Step 3 — Just the top: rank == 1

top_products = ranked.filter(F.col("rank") == 1).orderBy("store_id", "order_day")
print("\n=== top_products_per_store_per_day: rank == 1 only ===")
top_products.show(30, truncate=False)

num_groups = revenue_by_product_day.select("store_id", "order_day").distinct().count()
num_top = top_products.count()
assert num_groups == num_top == 20
print(f"\nVerification: {num_groups} store_id x order_day combinations, each with exactly one top product -> OK")

spark.stop()

What to expect (executed in this run):

=== top_products_per_store_per_day: rank == 1 only ===
+--------+----------+----------+-------------+----+
|store_id|order_day |product_id|total_revenue|rank|
+--------+----------+----------+-------------+----+
|S01     |2026-08-03|P004      |4.5          |1   |
|S01     |2026-08-04|P003      |2.25         |1   |
|S01     |2026-08-05|P001      |0.55         |1   |
|S01     |2026-08-06|P004      |4.5          |1   |
|S01     |2026-08-07|P002      |2.4          |1   |
|S01     |2026-08-08|P004      |9.0          |1   |
|S01     |2026-08-09|P001      |1.1          |1   |
|S02     |2026-08-03|P002      |2.4          |1   |
|S02     |2026-08-04|P002      |2.4          |1   |
|S02     |2026-08-05|P004      |9.0          |1   |
|S02     |2026-08-06|P001      |1.65         |1   |
|S02     |2026-08-07|P004      |4.5          |1   |
|S02     |2026-08-08|P001      |3.85         |1   |
|S02     |2026-08-09|P002      |1.2          |1   |
|S03     |2026-08-03|P001      |2.75         |1   |
|S03     |2026-08-04|P004      |4.5          |1   |
|S03     |2026-08-06|P002      |1.2          |1   |
|S03     |2026-08-07|P002      |3.6          |1   |
|S03     |2026-08-08|P004      |4.5          |1   |
|S03     |2026-08-09|P001      |1.65         |1   |
+--------+----------+----------+-------------+----+

Verification: 20 store_id x order_day combinations, each with exactly one top product -> OK

Twenty rows — one for each real store-and-day combination that shows up in Kiosko's week (S01 and S02 have activity all seven days; S03 has no orders at all on 2026-08-05, so it only contributes six combinations — 7 + 7 + 6 = 20). Each row is, precisely, the answer to "which product sold the most, at this store, on this day?" — checkable by hand against any fact_orders_df row you filter to that store and that day.

Worked example, part 2: the same ranking, at scale — the same top product

# ranking_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"))
fact_orders_at_scale_df = fact_orders_at_scale_df.withColumn("order_day", F.to_date("order_ts"))

revenue_by_product_day = (
    fact_orders_at_scale_df
    .groupBy("store_id", "order_day", "product_id")
    .agg(F.round(F.sum("revenue"), 2).alias("total_revenue"))
)

rank_window = Window.partitionBy("store_id", "order_day").orderBy(F.desc("total_revenue"), "product_id")
top_products = (
    revenue_by_product_day
    .withColumn("rank", F.row_number().over(rank_window))
    .filter(F.col("rank") == 1)
    .orderBy("store_id", "order_day")
)

print("=== top_products_per_store_per_day, at scale (250,000 franchises) ===")
top_products.show(30, truncate=False)

num_top = top_products.count()
assert num_top == 20
print(f"\nnum_top = {num_top}")

spark.stop()

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

=== top_products_per_store_per_day, at scale (250,000 franchises) ===
+--------+----------+----------+-------------+----+
|store_id|order_day |product_id|total_revenue|rank|
+--------+----------+----------+-------------+----+
|S01     |2026-08-03|P004      |1125000.0    |1   |
|S01     |2026-08-04|P003      |562500.0     |1   |
|S01     |2026-08-05|P001      |137500.0     |1   |
|S01     |2026-08-06|P004      |1125000.0    |1   |
|S01     |2026-08-07|P002      |600000.0     |1   |
|S01     |2026-08-08|P004      |2250000.0    |1   |
|S01     |2026-08-09|P001      |275000.0     |1   |
|S02     |2026-08-03|P002      |600000.0     |1   |
|S02     |2026-08-04|P002      |600000.0     |1   |
|S02     |2026-08-05|P004      |2250000.0    |1   |
|S02     |2026-08-06|P001      |412500.0     |1   |
|S02     |2026-08-07|P004      |1125000.0    |1   |
|S02     |2026-08-08|P001      |962500.0     |1   |
|S02     |2026-08-09|P002      |300000.0     |1   |
|S03     |2026-08-03|P001      |687500.0     |1   |
|S03     |2026-08-04|P004      |1125000.0    |1   |
|S03     |2026-08-06|P002      |300000.0     |1   |
|S03     |2026-08-07|P002      |900000.0     |1   |
|S03     |2026-08-08|P004      |1125000.0    |1   |
|S03     |2026-08-09|P001      |412500.0     |1   |
+--------+----------+----------+-------------+----+

num_top = 20

Compare this table, row by row, against part 1's: the winning product_id is identical across all twenty combinations, with total_revenue scaled by exactly ×250,000 (S01 on 2026-08-08: 9.0 → 2,250,000.0; S03 on 2026-08-07: 3.6 → 900,000.0). This isn't a coincidence — it's direct proof that franchise_id, exactly as module 4 declared it, is a purely technical column meant to add volume: since every product's revenue scales in exactly the same proportion across every franchise, the product that wins each group never changes, regardless of whether the dataset has forty rows or ten million.

Diagram: two steps, aggregate first, then rank

flowchart TD
    A["fact_orders_at_scale_df\n10,000,000 rows"] --> B["groupBy(store_id, order_day, product_id)\n.agg(F.sum('revenue'))"]
    B --> C["revenue_by_product_day\n~60-80 rows\n(store x day x product, aggregated)"]
    C --> D["Window.partitionBy(store_id, order_day)\n.orderBy(F.desc('total_revenue'))\nwith F.row_number().over()"]
    D --> E["ranked\nsame row count as C,\nwith a new 'rank' column"]
    E --> F["filter(rank == 1)"]
    F --> G["top_products\n20 rows -- one per\nstore_id x order_day combination"]

    style B fill:#69c,stroke:#333
    style D fill:#9c6,stroke:#333

Going deeper: why aggregate first, and rank after — not the other way around

It's worth noting this lesson resolves the ranking in two separate steps, and that separation isn't arbitrary. If you applied F.row_number() directly over fact_orders_at_scale_df's individual rows — without aggregating by product first — you'd be ranking individual orders, not products: the row with the highest revenue in a group could be a single large order for a product that, once summed with its other orders that same day, wasn't actually the best seller. Worse still, at scale, every order from every franchise would compete for its own rank within the group (store_id, order_day) — with 250,000 franchises producing the same repeated order, the "top" would basically end up being whichever of those identical copies happened to land first, with no real business meaning about which product sold the most in total.

Aggregating first — groupBy("store_id", "order_day", "product_id").agg(F.sum("revenue")) — collapses every franchise into a single number per product, exactly the right business question: "how much revenue did this product generate, at this store, on this day, summing every order?" Only after having that aggregated answer does it make sense to ask which product came in first — and that's where the window function comes in, operating over a result already reduced to a manageable size (a few dozen rows, not ten million). This order — aggregate, then rank — is the general pattern for any "top N per group, by an aggregated metric" question, and it's worth recognizing as such, beyond this specific Kiosko case.

Common mistakes

Ranking fact_orders_at_scale_df's individual rows, without aggregating first. What happens: someone, in a hurry, applies Window.partitionBy("store_id", "order_day").orderBy(F.desc("revenue")) directly over fact_orders_at_scale_df, skipping the intermediate groupBy step, and gets a "top" that doesn't correspond to any product that actually sold the most. Why it happens: the window syntax is identical in both cases — only which DataFrame it's applied to changes — so it's easy to skip the aggregation step without Spark complaining with any error. How to spot it: if your "top product" changes inconsistently between runs, or if your result's total_revenue matches a single order's revenue instead of a sum, suspect you're missing the intermediate groupBy. How to fix it: always follow this lesson's two-step pattern — aggregate first at the granularity that matters (store_id, order_day, product_id), and rank afterward over that aggregated result, never over the raw rows.

Using F.rank() instead of F.row_number(), without understanding the difference when there are ties. What happens: someone swaps F.row_number() for F.rank() — another Spark ranking function, with a similar name — and is surprised when two products with exactly the same total_revenue get the same rank number, and the next rank skips a number. Why it happens: both functions assign a position within a sorted group, and without seeing a real tie case, it's easy not to notice the difference. How to spot it: if your filter(F.col("rank") == 1) result returns more than one row for the same (store_id, order_day) combination, suspect you're using F.rank() (which does allow ties at number 1) instead of F.row_number() (which always assigns unique numbers, 1, 2, 3, ..., regardless of ties). How to fix it: this lesson deliberately uses F.row_number(), together with an explicit tiebreak in the orderBy (F.desc("total_revenue"), "product_id") to guarantee a single winner even if two products tied on revenue — the right combination when the business needs, by definition, a single "top product," never a list of ties.

Forgetting the product_id tiebreak in the window's orderBy, and getting a non-deterministic winner in case of a tie. What happens: someone writes Window.partitionBy("store_id", "order_day").orderBy(F.desc("total_revenue")), without the second product_id criterion, and in a group where two products happened to have exactly the same total_revenue, the product receiving rank == 1 could change between runs. Why it happens: Kiosko's real data has no exact total_revenue tie within a single group, so this problem never shows up visibly in the worked example — but the lack of an explicit tiebreak remains a latent fragility in the code. How to spot it: check whether your orderBy inside a Window specification has a single column that could, in principle, have repeated values — if so, and you need a single deterministic winner, you're missing a tiebreak criterion. How to fix it: always add an additional column, unique within the group (here, product_id), as the orderBy's second criterion — the same discipline you already applied with the franchise_id/order_id tiebreak in lesson 6, now applied to a ranking instead of a running total.

Exercises

Exercise 1 — Calculate each group's top-2, instead of just the top-1. Using ranked from the worked example over the forty real rows, filter by rank <= 2 instead of rank == 1, and confirm how many rows result in total.

top_2 = ranked.filter(F.col("rank") <= 2).orderBy("store_id", "order_day", "rank")
top_2.show(30, truncate=False)

num_top_2 = top_2.count()
print(f"num_top_2 = {num_top_2}")

+--------+----------+----------+-------------+----+
|store_id|order_day |product_id|total_revenue|rank|
+--------+----------+----------+-------------+----+
|S01     |2026-08-03|P004      |4.5          |1   |
|S01     |2026-08-03|P001      |2.75         |2   |
|S01     |2026-08-04|P003      |2.25         |1   |
|S01     |2026-08-04|P002      |1.2          |2   |
|S01     |2026-08-05|P001      |0.55         |1   |
...

num_top_2 = 34

Exercise 2 — Confirm S02's top product on 2026-08-05 is P004, with total_revenue = 9.0, and that it's the only product for that combination. Filter revenue_by_product_day (before ranking) to store_id == "S02" and order_day == "2026-08-05", and confirm there's only one row.

group_s02_0805 = revenue_by_product_day.filter(
    (F.col("store_id") == "S02") & (F.col("order_day") == "2026-08-05")
)
group_s02_0805.show(truncate=False)

num_rows = group_s02_0805.count()
assert num_rows == 1
row = group_s02_0805.first()
assert row["product_id"] == "P004"
assert row["total_revenue"] == 9.0
print(f"Verification: S02 on 2026-08-05 only sold {row['product_id']}, revenue={row['total_revenue']} -> OK")

+--------+----------+----------+-------------+
|store_id|order_day |product_id|total_revenue|
+--------+----------+----------+-------------+
|S02     |2026-08-05|P004      |9.0          |
+--------+----------+----------+-------------+

Verification: S02 on 2026-08-05 only sold P004, revenue=9.0 -> OK

Exercise 3 — Explain, without code, why the winning product_id is identical between the 40 rows and the 10 million, while lesson 6's running revenue did need a tiebreak to look "clean" at scale. In 2-3 sentences, explain the structural difference between this module's two business questions that makes one need an explicit tiebreak and the other not.

Summary and next step

This lesson resolved the top-product-per-store-per-day ranking in two steps: groupBy to aggregate revenue by product within each group, and F.row_number().over(Window.partitionBy("store_id", "order_day").orderBy(F.desc("total_revenue"), "product_id")) to keep each group's winner. You verified the result — twenty store-and-day combinations, each with its winning product — first over the forty real rows, and confirmed the same product_id wins each group at full scale, with revenue scaled by exactly ×250,000.

Before moving on you should be able to: explain why ranking needs an aggregation step before the window, unlike lesson 6's running total; distinguish F.row_number() from F.rank() in the case of a tie; and explain why a ranking window's orderBy tiebreak (product_id) resolves the same kind of problem as the running total's tiebreak (franchise_id, order_id), even though the two business questions are different.

Lesson 8 integrates everything you built in this module — the JOIN criterion from lessons 2 through 4, lesson 6's running total, this lesson's ranking — into a single pipeline, run end to end over the complete fact_orders_at_scale.

Resources

• PySpark — pyspark.sql.functions.row_number (the exact reference for the function used in this lesson). spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.row_number.html.
• PySpark — pyspark.sql.Window (the complete partitionBy/orderBy reference, already cited in lessons 5 and 6, the foundation for this lesson's ranking window). 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: the top-product-per-store-per-day ranking, verified over 40 rows and over fact_orders_at_scale.
• data-modeling-for-analytics-guide's DESIGN doc — the source of the "top product" question this lesson resolves with F.row_number(), instead of array columns in DuckDB. src/guides/data-modeling-for-analytics-guide/DISENO.md