Module introduction: joins and window functions at scale

Why this module exists

Module 4 left two questions open on purpose, and named them as such without resolving them. Lesson 3 showed something that surprises anyone coming from pure SQL: a .join() of orders_df (forty rows) against dim_store (three rows) did not produce the Exchange node the official Spark documentation promises for join operations — it produced, instead, BroadcastHashJoin and BroadcastExchange, a completely different mechanism. Lesson 5 of that same module showed the reverse: a join of orders_at_scale_df (ten million rows) against itself, by product_id, did produce a real Exchange, with the SortMergeJoin strategy. Two behaviors, over the same .join() operation, with neither lesson explaining the exact criterion that decides which of the two Spark uses. This module exists to settle that debt: the configurable threshold (spark.sql.autoBroadcastJoinThreshold), forced and compared with evidence, over the same join, under the same conditions.

This module's second half opens a topic no earlier module touched: window functions. Up to now, every groupBy in this guide has collapsed a group of rows into a single result row — groupBy("store_id").sum("revenue") turns forty orders into three rows, one per store, losing each individual order's detail in the process. There's a class of question a groupBy can't answer without losing that detail: "what's this store's cumulative revenue, up to this exact order, without grouping anything?", or "what's each store's best-selling product, each day, without ever losing sight of each individual product?". That's exactly the class of question data-modeling-for-analytics-guide already solved, with pure SQL over DuckDB, using array columns and a cumulative table design — and it's the same question this module now solves with Spark's native window: Window.partitionBy().orderBy(), without losing a single row.

Connection to the module. Modules 1 through 4 built Spark's complete mental model — SparkSession, lazy evaluation, .join(), .groupBy(), partitions, shuffle — but none of them gave you the criterion for deciding, with data in front of you, which join strategy makes sense, nor a way to compute a running total or a ranking without losing row-level detail. This module closes the two missing pieces before module 8's capstone, and it does it over the same at-scale dataset module 4 already built: fact_orders_at_scale, ten million rows, synthetic, declared as such.

An analogy: the photocopied directory, and the runner at every checkpoint

Picture a delivery company that needs every one of its trucks to know, for each order, exactly which address corresponds to each customer code. If the customer directory is small — a few pages — the obvious solution is to photocopy it and hand a complete copy to every truck: each driver resolves the lookup on their own, needing no radio and no coordination with the other trucks. That's a broadcast join: the small table (dim_store, three rows; dim_product, four rows) gets copied whole to every partition on the large side, and each partition resolves the match alone. But if the directory were huge — millions of customers — photocopying it whole for every truck would be absurd: it would take up more space than the cargo itself. In that case, the only sensible option is for every truck to stop, reorganize its orders by customer code, and match them against an equally reorganized version of the large directory — the shuffle cost you already know from module 4, now applied to both sides of a JOIN. That's a sort-merge join.

Window functions call for a different analogy. Think about a marathon runner's cumulative time, measured at every checkpoint along the route. At kilometer 10, the runner knows their cumulative time since the start — a number that depends solely on their own run up to that point, with no need to know anything about the other runners, nor to group their time with anyone else's. And, at the same time, a scoreboard at the finish line can show the top three runners' ranking in each age category, at every checkpoint, without that requiring any runner to stop existing as an individual within that category. A running total (revenue accumulated per store, as orders come in) and a ranking within a group (the top product per store, per day) are, precisely, the two questions this module solves with Window.partitionBy().orderBy() — without a single individual row disappearing from the result, unlike what a groupBy does.

Worked example: this module's map, before touching anything

# joins_and_windows_map.py
JOINS_AND_WINDOWS_MAP = [
    (2, "Broadcast join vs shuffle join",
        "The full criterion: spark.sql.autoBroadcastJoinThreshold, and why Spark picks one or the other"),
    (3, "Reading a broadcast join in explain",
        "fact_orders_at_scale (10M) against dim_store/dim_product -- BroadcastHashJoin by default, verified"),
    (4, "Forcing and reading a sort-merge join",
        "The same join, with the threshold disabled -- SortMergeJoin with Exchange on both sides, compared"),
    (5, "Window functions: partitionBy and orderBy",
        "Window.partitionBy().orderBy() -- the syntax, verified first over the 40 real rows"),
    (6, "A running revenue total per store",
        "F.sum('revenue').over(window) -- the exact running total, without losing a single row, 40 rows and 10M"),
    (7, "Ranking top products per store per day",
        "F.row_number().over(window) -- the same top product, identical at real scale and at synthetic scale"),
    (8, "Project: Kiosko's scaled joins and rankings",
        "The previous six lessons, integrated into a single script verified over 10 million rows"),
]

print("=== Joins and window functions at scale, before touching anything ===\n")
for number, title, detail in JOINS_AND_WINDOWS_MAP:
    print(f"L{number}. {title}")
    print(f"     {detail}\n")

print("This module's goal: decide with a real criterion between two JOIN strategies,")
print("and answer running-total/ranking questions without losing row-level detail.")

What to expect. Running python3 joins_and_windows_map.py, the output is exactly this:

=== Joins and window functions at scale, before touching anything ===

L2. Broadcast join vs shuffle join
     The full criterion: spark.sql.autoBroadcastJoinThreshold, and why Spark picks one or the other

L3. Reading a broadcast join in explain
     fact_orders_at_scale (10M) against dim_store/dim_product -- BroadcastHashJoin by default, verified

L4. Forcing and reading a sort-merge join
     The same join, with the threshold disabled -- SortMergeJoin with Exchange on both sides, compared

L5. Window functions: partitionBy and orderBy
     Window.partitionBy().orderBy() -- the syntax, verified first over the 40 real rows

L6. A running revenue total per store
     F.sum('revenue').over(window) -- the exact running total, without losing a single row, 40 rows and 10M

L7. Ranking top products per store per day
     F.row_number().over(window) -- the same top product, identical at real scale and at synthetic scale

L8. Project: Kiosko's scaled joins and rankings
     The previous six lessons, integrated into a single script verified over 10 million rows

This module's goal: decide with a real criterion between two JOIN strategies,
and answer running-total/ranking questions without losing row-level detail.

Notice the structure: lessons 2 through 4 fully close the join question module 4 left open — criterion, default evidence, forced evidence. Lessons 5 through 7 open and resolve, with real evidence, the window-function question from scratch. Lesson 8 integrates both halves into a single pipeline, run over the complete fact_orders_at_scale.

Diagram: where you were, where you're headed

flowchart LR
    subgraph M14["Modules 1-4 (already done)"]
        A["Spark installed, fact_orders rebuilt,\nkiosko_orders_at_scale generated (10M rows),\npartitions and shuffle measured"]
    end

    subgraph M5["This module (5 of 8)"]
        B["L2-L4: broadcast join vs\nsort-merge join, with threshold\nforced and explain compared"]
        C["L5-L7: Window.partitionBy/orderBy --\nrunning total per store, ranking\nof top product per store/day"]
        D["L8: project -- joins and\nwindows, integrated, 10M rows"]
    end

    subgraph Resto["Modules 6-8"]
        E["Catalyst through its phases, cache,\npartitioned Parquet, UDFs,\nfinal capstone"]
    end

    A --> B --> C --> D --> E

This module's map

Lesson    What it builds
────────  ──────────────────────────────────────────────────────────────
L1        (this one) The map: joins and windows, before touching them
L2        The full criterion: when Spark picks broadcast vs shuffle
L3        BroadcastHashJoin read in .explain(), over fact_orders_at_scale
L4        SortMergeJoin forced and read in .explain(), same join
L5        Window.partitionBy().orderBy() -- syntax, over 40 real rows
L6        Running revenue total per store -- 40 rows, then 10M
L7        Top product per store per day -- 40 rows, then 10M
L8        Project: Kiosko's scaled joins and rankings, verified

Lessons 2 through 4 are, in essence, a single question resolved in three steps: what criterion Spark uses, what it produces by default, what it produces forced. Lessons 5 through 7 build window functions from the simplest syntax up to the two complete business questions Kiosko needs answered. Lesson 8 introduces nothing new — it combines both halves of the module in the same script, over the complete ten million rows.

Going deeper: why joins and windows live in the same module

At first glance, "JOIN strategies" and "window functions" look like two unrelated topics — one decides how Spark executes a match between tables, the other computes a running total or a ranking within a group. But they share something structural worth naming before starting: both are, underneath, decisions about how to organize rows across partitions before answering a question, the same central question that ran through all of module 4. A BroadcastHashJoin avoids reorganizing the large side by copying the small side to every partition. A SortMergeJoin reorganizes both sides by the JOIN key. And a window function — you're going to see this with evidence in lesson 5 — also reorganizes rows: Window.partitionBy("store_id") physically groups a given store's rows into the same partition, exactly the way a groupBy("store_id") would, with the difference that, at the end, no row gets lost or collapsed.

This continuity matters because it lets you reason about new Spark code with the same tools you already have: any operation that needs to compare or combine rows currently living in different partitions — a join, a groupBy, a window — is going to have an answer to the question "does this trigger a shuffle?", and that answer is always readable in .explain(), never on a stopwatch. This module doesn't ask you to learn a new mental model — it asks you to apply, precisely, module 4's same mental model to two operations you hadn't seen yet.

Common mistakes

Assuming "broadcast join" is always better than "sort-merge join," unconditionally. What happens: someone, seeing that BroadcastHashJoin avoids the shuffle, concludes it's always worth forcing, even when the table meant to be broadcast isn't actually small. Why it happens: "avoids the shuffle" sounds like "cheaper," with no nuance, and it's tempting to generalize that conclusion to any table size. How to spot it: if your reasoning about which JOIN strategy to use never includes the word "size," you're missing this module's central criterion. How to fix it: a BroadcastHashJoin copies the complete table to every partition on the large side — if the "small" table actually weighs several gigabytes and the cluster has hundreds of partitions, that repeated copy can consume far more total memory than the shuffle it was meant to avoid. Lesson 2 develops the full criterion, with the exact threshold Spark uses to decide.

Expecting window functions to be just "a fancier groupBy." What happens: someone, hearing "running revenue total per store" and "ranking per store," assumes the result is going to have one row per store, like any groupBy they already know from modules 3 and 4. Why it happens: both questions sound, in everyday language, like "one answer per group" — the same shape a groupBy produces. How to spot it: if by the end of lesson 5 you can't explain why a window function's result has the same number of rows as the original table, instead of one row per group, you're missing the central difference between groupBy and Window. How to fix it: a window function computes an aggregated value — a sum, a ranking — for each row, using the group that row belongs to as context, without collapsing any row out of the result. Lesson 5 demonstrates this with evidence, side by side with a groupBy over the same data.

Jumping straight to lesson 6 or 7 without understanding lesson 5's Window.partitionBy().orderBy(). What happens: someone, impatient to see the business result (the running total, the ranking), copies lesson 6 or 7's code without understanding what each piece of the window specification does (partitionBy, orderBy, the aggregate function, .over()), and gets lost the moment something doesn't come out as expected. Why it happens: lessons 6 and 7's final result is more interesting than lesson 5's basic syntax, and it's tempting to treat the latter as a dispensable preamble. How to spot it: if you can't explain, without looking at the code, what partitionBy controls and what orderBy controls within a window specification, you're missing the foundation lesson 5 deliberately builds with a minimal example before applying it to a real business question. How to fix it: lesson 5 isn't a preamble — it's where the two pieces of any window function get separated, with evidence: partitionBy defines the group (like a groupBy), orderBy defines the order within that group (something a groupBy doesn't have).

Exercises

Exercise 1 — Recite, from memory, this module's central analogy. Without rereading this lesson, explain in your own words the two analogies: the photocopied directory (for joins) and the marathon runner at every checkpoint (for windows). What does the decision "copy the whole thing" versus "reorganize everything" represent in each?

See solution

The photocopied directory represents a BroadcastHashJoin: when the small table (the directory) is small enough to copy whole, each partition (each truck) resolves the match on its own, with no coordination among the others. When the directory is too large to photocopy, the only option is for every truck to stop and reorganize its cargo by the same key — a SortMergeJoin, with the shuffle cost you already know from module 4. The marathon runner represents a window function: their cumulative time at every checkpoint depends only on their own run (the partitionBy that groups them together with the other runners in their category, and the orderBy that defines the order of the checkpoints), with no need to group away or lose sight of any individual runner — unlike a scoreboard that only showed the whole category's average time, which would lose that detail.

Exercise 2 — Predict, before reading lesson 2, what table size makes Spark choose each strategy. Based only on the photocopied-directory analogy, predict: which characteristic of a table — row count, column count, size in bytes, or something else — do you think determines whether Spark treats it as "small enough to photocopy"?

See solution

Reasonable prediction: size in bytes, not row count or column count on their own — a table with few rows but very heavy columns (long text, for example) could weigh more than it looks, and a table with many rows but tiny columns could weigh less than expected. Lesson 2 confirms this intuition with the exact name of the setting Spark uses (spark.sql.autoBroadcastJoinThreshold) and its default value, measured in bytes.

Exercise 3 — Explain, without code, why this module verifies window functions first over the 40 real rows, and only afterward over the 10 million synthetic ones. In 2-3 sentences, justify that order, connecting it to something you already saw in module 4.

See solution

Verifying first over Kiosko's 40 real rows lets you check the result by hand — adding up revenue with a calculator, sorting by date with your own eyes — the same "verify, don't assume" discipline you already saw in module 3 when comparing fact_orders against other engines. Only after confirming the logic is correct on a small, checkable case does it make sense to run it over fact_orders_at_scale, where checking the result row by row by hand is no longer possible — there, the evidence becomes the same kind of assert-based verification against a known formula module 4 already used with kiosko_orders_at_scale.

Summary and next step

This module closes two questions the rest of the guide left open on purpose: the exact criterion that decides between BroadcastHashJoin and SortMergeJoin (lessons 2 through 4), and the syntax and the two business applications of Spark's window functions — running total per store, top-product ranking — without losing the row-level detail a groupBy does lose (lessons 5 through 7). Lesson 8 integrates both halves into a single pipeline, run end to end over fact_orders_at_scale.

Before moving on you should be able to: explain the difference between a BroadcastHashJoin and a SortMergeJoin with the photocopied-directory analogy; explain why a window function doesn't collapse rows, unlike a groupBy; and predict why this module verifies every result first over 40 rows, and only then over 10 million.

Lesson 2 starts with the full criterion: exactly which setting Spark uses to decide, and what happens when you deliberately disable it.

Resources

• Apache Spark — SQL Performance Tuning, JOIN strategies section and spark.sql.autoBroadcastJoinThreshold (the central reference for lessons 2 through 4 of this module). spark.apache.org/docs/latest/sql-performance-tuning.html.
• PySpark — pyspark.sql.Window (the window-function API reference, the foundation for lessons 5 through 7). 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 module's full objective within the eight-module plan.
• data-modeling-for-analytics-guide's DESIGN doc — the source of the running-total and ranking question this module revisits with Spark's native window, instead of array columns in DuckDB. src/guides/data-modeling-for-analytics-guide/DISENO.md