Project: Kiosko's scaled joins and rankings

Description

This project closes the module by pulling the previous six lessons together into a single script: you read fact_orders_at_scale and the two dimension tables, confirm with .explain() that Spark chooses BroadcastHashJoin by default, force and confirm SortMergeJoin by disabling the threshold, compute the running revenue total per store with an explicit tiebreak, and close with the top-product-per-store-per-day ranking. All of it, verified with assert, over the complete ten million rows of fact_orders_at_scale.

Connection to the module. This project introduces no new concept — it's the final integration of lessons 2 through 7, in the same order you built them, with a single script run end to end over the complete synthetic dataset.

An analogy: the full season audit

Pick back up this module's two central analogies: the photocopied (or not) directory for every JOIN, and the runner with their cumulative time and their position at every checkpoint. This project is the full audit of an entire season of deliveries and racing: what decision did every truck make in front of every directory, with evidence from its route plan? What was each runner's final cumulative total, verified checkpoint by checkpoint? Who came in first, in every category, at every checkpoint? An audit that doesn't check every one of these questions, with executed evidence, isn't a full audit — it's just someone's word that "everything went fine."

The material: everything this module built, in a single flow

You need: kiosko_orders_at_scale.csv (generated in module 4, lesson 4, with generate_orders_at_scale(250_000)), dim_store.csv, and dim_product.csv (the same two dimension tables from module 3), all in the same folder where you're going to run this script.

The verified reference solution

Part 1 and 2 — Open the session, read fact_orders_at_scale and the dimensions

# kiosko_scaled_joins_and_rankings.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,
)

print("=== Kiosko at scale, joins and windows: module 5's final delivery ===\n")

print("Part 1 -- opening the SparkSession")
spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").getOrCreate()
print(f"Spark version: {spark.version}\n")

print("Part 2 (L2-L4) -- reading fact_orders_at_scale, dim_store, dim_product")
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),
])
dim_store_schema = StructType([
    StructField("store_id", StringType(), False),
    StructField("store_name", StringType(), False),
    StructField("city", StringType(), False),
])
dim_product_schema = StructType([
    StructField("product_id", StringType(), False),
    StructField("product_name", StringType(), False),
    StructField("category", StringType(), False),
    StructField("unit_cost", DoubleType(), False),
])
orders_at_scale_df = spark.read.csv("kiosko_orders_at_scale.csv", schema=scale_schema, header=True, enforceSchema=False)
dim_store_df = spark.read.csv("dim_store.csv", schema=dim_store_schema, header=True, enforceSchema=False)
dim_product_df = spark.read.csv("dim_product.csv", schema=dim_product_schema, header=True, enforceSchema=False)

fact_orders_at_scale_df = orders_at_scale_df.withColumn("revenue", F.col("quantity") * F.col("unit_price"))
row_count = fact_orders_at_scale_df.count()
print(f"fact_orders_at_scale_df.count() = {row_count}")
assert row_count == 10_000_000
print("Verification: 10,000,000 rows -> OK\n")

Part 3 — The default JOIN: BroadcastHashJoin

print("Part 3 (L2-L3) -- default join: BroadcastHashJoin")
joined_default = fact_orders_at_scale_df.join(dim_store_df, "store_id").join(dim_product_df, "product_id")
print(f"spark.sql.autoBroadcastJoinThreshold = {spark.conf.get('spark.sql.autoBroadcastJoinThreshold')}")
print("Physical plan (note BroadcastHashJoin / BroadcastExchange, no shuffle Exchange):")
joined_default.explain()
count_default = joined_default.count()
print(f"joined_default.count() = {count_default}")
assert count_default == row_count == 10_000_000
print("Verification: default BroadcastHashJoin, no rows lost -> OK\n")

Part 4 — The same JOIN, forced to SortMergeJoin

print("Part 4 (L4) -- same join, forced to SortMergeJoin")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
joined_forced = fact_orders_at_scale_df.join(dim_store_df, "store_id")
print(f"spark.sql.autoBroadcastJoinThreshold = {spark.conf.get('spark.sql.autoBroadcastJoinThreshold')}")
print("Physical plan (note SortMergeJoin + Exchange on both sides):")
joined_forced.explain()
count_forced = joined_forced.count()
assert count_forced == row_count == 10_000_000
print(f"joined_forced.count() = {count_forced}")
print("Verification: forced SortMergeJoin, same result, different plan -> OK")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 10 * 1024 * 1024)
print(f"spark.sql.autoBroadcastJoinThreshold restored = {spark.conf.get('spark.sql.autoBroadcastJoinThreshold')}\n")

Part 5 — Running revenue total per store, with a tiebreak

print("Part 5 (L5-L6) -- running revenue total per store, with tiebreaker")
store_window = Window.partitionBy("store_id").orderBy("order_ts", "franchise_id", "order_id")
with_running = fact_orders_at_scale_df.withColumn("running_total", F.round(F.sum("revenue").over(store_window), 2))
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": 9575000.0, "S02": 9700000.0, "S03": 7262500.0}
print("Verification: final running_total matches the known breakdown -> OK\n")

Part 6 — Top product per store per day

print("Part 6 (L7) -- top product per store per day")
fact_with_day = fact_orders_at_scale_df.withColumn("order_day", F.to_date("order_ts"))
revenue_by_product_day = (
    fact_with_day.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")
)
num_top = top_products.count()
print(f"num_top (rows with rank == 1) = {num_top}")
assert num_top == 20
sample = top_products.filter((F.col("store_id") == "S01") & (F.col("order_day") == "2026-08-08")).collect()[0]
print(f"S01 2026-08-08 -- top product = {sample['product_id']}, total_revenue = {sample['total_revenue']}")
assert sample["product_id"] == "P004"
assert sample["total_revenue"] == 2_250_000.0
print("Verification: 20 store x day combinations, each with its top product, revenue scaled x250,000 -> OK\n")

Part 7 — Final summary

print("Part 7 -- final summary")
total_revenue = round(sum(final_by_store.values()), 2)
print(f"Rows processed: {row_count:,}")
print(f"Verified total revenue: {total_revenue:,}")
assert total_revenue == 26_537_500.00

spark.stop()
print("=== spark.stop() -- module 5 closed, joins and windows verified over 10M rows ===")

What to expect. Running the complete python3 kiosko_scaled_joins_and_rankings.py (all seven parts together), the output is exactly this (executed in this run, PySpark 4.2.0):

=== Kiosko at scale, joins and windows: module 5's final delivery ===

Part 1 -- opening the SparkSession
Spark version: 4.2.0

Part 2 (L2-L4) -- reading fact_orders_at_scale, dim_store, dim_product
fact_orders_at_scale_df.count() = 10000000
Verification: 10,000,000 rows -> OK

Part 3 (L2-L3) -- default join: BroadcastHashJoin
spark.sql.autoBroadcastJoinThreshold = 10485760b
Physical plan (note BroadcastHashJoin / BroadcastExchange, no shuffle Exchange):
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [product_id#3, store_id#2, order_id#0, franchise_id#1, quantity#4, unit_price#5, order_ts#6, revenue#15, store_name#8, city#9, product_name#11, category#12, unit_cost#13]
   +- BroadcastHashJoin [product_id#3], [product_id#10], Inner, BuildRight, false, false
      :- Project [store_id#2, order_id#0, franchise_id#1, product_id#3, quantity#4, unit_price#5, order_ts#6, revenue#15, store_name#8, city#9]
      :  +- BroadcastHashJoin [store_id#2], [store_id#7], Inner, BuildRight, false, false
      :     :- Project [order_id#0, franchise_id#1, store_id#2, product_id#3, quantity#4, unit_price#5, order_ts#6, (cast(quantity#4 as double) * unit_price#5) AS revenue#15]
      :     :  +- Filter (isnotnull(store_id#2) AND isnotnull(product_id#3))
      :     :     +- FileScan csv [order_id#0,franchise_id#1,store_id#2,product_id#3,quantity#4,unit_price#5,order_ts#6] Batched: false, DataFilters: [isnotnull(store_id#2), isnotnull(product_id#3)], Format: CSV, ...
      :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=75]
      :        +- Filter isnotnull(store_id#7)
      :           +- FileScan csv [store_id#7,store_name#8,city#9] Batched: false, DataFilters: [isnotnull(store_id#7)], Format: CSV, ...
      +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=79]
         +- Filter isnotnull(product_id#10)
            +- FileScan csv [product_id#10,product_name#11,category#12,unit_cost#13] Batched: false, DataFilters: [isnotnull(product_id#10)], Format: CSV, ...

joined_default.count() = 10000000
Verification: default BroadcastHashJoin, no rows lost -> OK

Part 4 (L4) -- same join, forced to SortMergeJoin
spark.sql.autoBroadcastJoinThreshold = -1
Physical plan (note SortMergeJoin + Exchange on both sides):
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [store_id#2, order_id#0, franchise_id#1, product_id#3, quantity#4, unit_price#5, order_ts#6, revenue#15, store_name#8, city#9]
   +- SortMergeJoin [store_id#2], [store_id#7], Inner
      :- Sort [store_id#2 ASC NULLS FIRST], false, 0
      :  +- Exchange hashpartitioning(store_id#2, 200), ENSURE_REQUIREMENTS, [plan_id=331]
      :     +- Project [order_id#0, franchise_id#1, store_id#2, product_id#3, quantity#4, unit_price#5, order_ts#6, (cast(quantity#4 as double) * unit_price#5) AS revenue#15]
      :        +- Filter isnotnull(store_id#2)
      :           +- FileScan csv [order_id#0,franchise_id#1,store_id#2,product_id#3,quantity#4,unit_price#5,order_ts#6] Batched: false, DataFilters: [isnotnull(store_id#2)], Format: CSV, ...
      +- Sort [store_id#7 ASC NULLS FIRST], false, 0
         +- Exchange hashpartitioning(store_id#7, 200), ENSURE_REQUIREMENTS, [plan_id=332]
            +- Filter isnotnull(store_id#7)
               +- FileScan csv [store_id#7,store_name#8,city#9] Batched: false, DataFilters: [isnotnull(store_id#7)], Format: CSV, ...

joined_forced.count() = 10000000
Verification: forced SortMergeJoin, same result, different plan -> OK
spark.sql.autoBroadcastJoinThreshold restored = 10485760

Part 5 (L5-L6) -- running revenue total per store, with tiebreaker
final running_total per store = {'S02': 9700000.0, 'S01': 9575000.0, 'S03': 7262500.0}
Verification: final running_total matches the known breakdown -> OK

Part 6 (L7) -- top product per store per day
num_top (rows with rank == 1) = 20
S01 2026-08-08 -- top product = P004, total_revenue = 2250000.0
Verification: 20 store x day combinations, each with its top product, revenue scaled x250,000 -> OK

Part 7 -- final summary
Rows processed: 10,000,000
Verified total revenue: 26,537,500.0
=== spark.stop() -- module 5 closed, joins and windows verified over 10M rows ===

Seven parts, seven verifications, and the same 26,537,500.00 module 4 had already confirmed with assert against the raw data — now reproduced again, this time by summing lesson 6's three final running_total values, instead of a direct groupBy. Notice that restoring the threshold, at the end of Part 4 (10485760, without the b suffix this time, because it was set with an integer instead of read from Spark's default value), confirms an important discipline for any real script: any setting you change deliberately for an experiment must return to its original value before the rest of the script relies on the default behavior.

Diagram: the seven parts, closing out the complete module

flowchart TD
    A["Part 1: SparkSession opened"] --> B
    B["Part 2 (L2-L4): fact_orders_at_scale,\ndim_store, dim_product read -- 10M rows"] --> C
    C["Part 3 (L2-L3): default join --\nBroadcastHashJoin, no shuffle Exchange"] --> D
    D["Part 4 (L4): same join forced --\nSortMergeJoin, Exchange on both sides"] --> E
    E["Part 5 (L5-L6): running revenue total\nper store, with tiebreak -- 9.57M/9.7M/7.26M"] --> F
    F["Part 6 (L7): top product per\nstore and day -- 20 combinations"] --> G["Module 5 closed:\njoins and windows\nverified over 10M rows"]

Closing out this module's checklist, piece by piece

Five modules resolved out of eight — more than halfway down this guide's road. With this module closed, you have the two pieces that were missing to reason with a complete criterion about any Spark query: when a JOIN pays the cost of a shuffle and when it doesn't, and how to answer running-total and ranking questions without losing the row-level detail a groupBy sacrifices.

Common mistakes

Running this project without having generated kiosko_orders_at_scale.csv, dim_store.csv, or dim_product.csv first. What happens: someone jumps straight into this project without having the three files in the working folder, and the script fails in Part 2 with a file-not-found error. Why it happens: it's tempting to treat the closing project as an independent starting point. How to spot it: if spark.read.csv(...) fails with AnalysisException: Path does not exist, you're missing one of the three files — kiosko_orders_at_scale.csv from module 4, dim_store.csv/dim_product.csv from module 3 — in the same folder where you're running this script. How to fix it: this project, just like modules 3 and 4's, reuses artifacts generated in earlier lessons — confirm all three files exist before running the complete script.

Forgetting to restore spark.sql.autoBroadcastJoinThreshold between Part 4 and Part 5. What happens: someone removes the line that restores the threshold at the end of Part 4, and notices no immediate problem because Parts 5 and 6 of this project never use .join() again — but if the script were extended with an additional JOIN afterward, that JOIN would inherit the disabled threshold with no warning at all. Why it happens: within this specific project, skipping the restoration has no visible effect, because no later part depends on the default broadcast behavior. How to spot it: check that any script combining, within the same SparkSession, a configuration experiment (like forcing SortMergeJoin) with later work that should behave "normally," explicitly restores that setting. How to fix it: Part 4 of this project restores the threshold explicitly, with a clear comment about what it's doing and why — follow that same pattern in any script of your own that changes Spark settings midway through.

Turning in the project without Part 6's asserts, relying only on .show(). What happens: someone runs all seven parts, sees top_products.show() "looks fine," and considers the project done without checking whether sample["product_id"] == "P004"'s assert actually passed. Why it happens: twenty rows each with a rank of 1 look correct at a glance, and formally verifying a specific case feels like an extra step over a result that already looks fine. How to spot it: if your final delivery didn't run Part 6's asserts to completion without raising an AssertionError, you have no real guarantee the ranking is correct — the same trap modules 3 and 4's projects already warned about. How to fix it: this project's asserts — over the row count, the per-store breakdown, the top product for a specific combination — aren't decorative: they're proof the complete script, not just a fragment, produces the correct result end to end.

Exercises

Exercise 1 — Extend the project with a Part 8: the ranking, with dim_product joined in to show product_name instead of product_id. Add an eighth part that joins top_products against dim_product_df (using the default BroadcastHashJoin), and shows the readable product_name instead of the code.

print("Part 8 -- top product, with a readable name")
top_products_named = top_products.join(dim_product_df, "product_id").select(
    "store_id", "order_day", "product_id", "product_name", "total_revenue", "rank"
).orderBy("store_id", "order_day")
top_products_named.explain()
top_products_named.show(20, truncate=False)

num_named = top_products_named.count()
assert num_named == 20
print("Verification: the join against dim_product didn't lose any of the 20 rows -> OK")

Part 8 -- top product, with a readable name
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- ...
   +- BroadcastHashJoin [product_id#N], [product_id#M], Inner, BuildRight, false, false
      ...

+--------+----------+----------+--------------------+-------------+----+
|store_id|order_day |product_id|product_name        |total_revenue|rank|
+--------+----------+----------+--------------------+-------------+----+
|S01     |2026-08-03|P004      |Phone Charger Cable |1125000.0    |1   |
|S01     |2026-08-04|P003      |Instant Coffee Sachet|562500.0    |1   |
...

Verification: the join against dim_product didn't lose any of the 20 rows -> OK

Exercise 2 — Confirm that the order of Parts 3/4 (joins) and Parts 5/6 (windows) doesn't affect the final result. Without running anything, explain in 2-3 sentences whether reordering the script — computing the running revenue total and the ranking (Parts 5 and 6) before the JOIN experiments (Parts 3 and 4) — would change any of the final asserts.

Exercise 3 — Explain, without looking at the guide's design doc, what this pipeline is missing to become module 8's complete capstone. In a 4-6 sentence paragraph, describe which pieces kiosko_scaled_joins_and_rankings.py is missing to become module 8's complete distributed pipeline.

Summary and next step: the end of module 5

With this mini-project you close out the complete module 5. You confirmed, with .explain() run over fact_orders_at_scale, that Spark chooses BroadcastHashJoin by default against dim_store and dim_product — with no shuffle Exchange — and that forcing spark.sql.autoBroadcastJoinThreshold = -1 produces SortMergeJoin, with Exchange on both sides of the JOIN, over the same logical result. You computed the running revenue total per store with Window.partitionBy("store_id").orderBy("order_ts", "franchise_id", "order_id"), verified by hand over the forty real rows and confirmed at full scale (9,575,000.00 / 9,700,000.00 / 7,262,500.00). And you closed with the top-product-per-store-per-day ranking, with F.row_number() over already-aggregated data, confirming the same product wins each combination, regardless of whether the dataset has forty rows or ten million.

You took this module's central step: you stopped treating "JOIN strategy" as a black box Spark decides on its own, and learned the exact criterion governing it; and you stopped thinking of groupBy as the only tool for answering aggregated questions, adding window functions as the right tool when row-level detail matters as much as the aggregate.

Where you're headed. Module 6 — catalyst-explain-and-caching — takes the same kind of .explain() plan you already read in this module, and develops it in depth: the Catalyst optimizer's complete phases (parsed, analyzed, optimized, physical), Adaptive Query Execution compared active versus disabled, and the complete criterion for when .cache() genuinely helps and when it only spends memory with no benefit at all.

Resources

• Apache Spark — SQL Performance Tuning (Catalyst, JOIN strategies, spark.sql.autoBroadcastJoinThreshold — the central reference for Parts 3 and 4 of this project). spark.apache.org/docs/latest/sql-performance-tuning.html.
• PySpark — pyspark.sql.Window (the partitionBy/orderBy reference, the foundation for Parts 5 and 6 of this project). spark.apache.org/docs/latest/api/python/reference/pyspark.sql/window.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — module 5's complete objective and its place in the eight-module plan.
• data-modeling-for-analytics-guide's DESIGN doc — the source of the running-total and ranking question this module resolved with Spark's native window. src/guides/data-modeling-for-analytics-guide/DISENO.md