Choosing partitioning, caching, and `JOIN` strategy, with criteria

Description

Lesson 3's pipeline made three design decisions without stopping to justify them at the time: it wrote the Parquet partitioned by store_id (not by franchise_id, not by product_id), let Spark choose BroadcastHashJoin for the two dimensions, and cached fact_orders_at_scale_df before the three queries reusing it. This lesson is Deliverable 3 from lesson 2's brief: it turns each of those three decisions into a runnable Python function, with the real evidence — measured cardinalities, real file sizes, an explicit reuse count — justifying them. No decision in this pipeline was made out of habit; this lesson proves it with code.

Connection to the module. This lesson introduces no new criterion — it codifies, in three short functions, the criteria you already built in modules 5 (JOIN strategy), 6 (caching), and 7 (partition column), and applies them with genuinely measured data over kiosko_orders_at_scale, not hypothetical examples.

An analogy: the decision's paper trail, not just the decision

Any serious engineering decision — and this isn't exclusive to Spark — should be defensible with a paper trail: not just "we chose store_id," but "we chose store_id because it has 3 distinct values, against franchise_id's 250,000, and the partitioning criterion calls for low cardinality." A paper trail like that lets someone, a year later, who never took part in the original decision, read it and understand exactly why it was made that way, without having to trust anyone's memory. This lesson builds that paper trail for this capstone's pipeline's three central decisions — not as documentation after the fact, but as runnable code that remeasures the evidence every time it runs.

Worked example: three decisions, three functions, real evidence

Part 1 — Partitioning criterion: each candidate's real cardinality

# kiosko_design_criteria.py
import os
from pyspark.sql import SparkSession
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType,
)

print("=== Module 8, lesson 5: partitioning, caching, and joins, with criteria and evidence ===\n")

spark = (
    SparkSession.builder
    .appName("kiosko-spark")
    .master("local[*]")
    .config("spark.driver.memory", "4g")
    .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)

print("Part 1 -- partitioning criterion: each candidate's real cardinality")
candidates = ["store_id", "product_id", "franchise_id"]
cardinalities = {}
for c in candidates:
    n = orders_at_scale_df.select(c).distinct().count()
    cardinalities[c] = n
    print(f"  cardinality of {c}: {n}")


def choose_partition_column(cardinalities: dict, max_folders: int = 50) -> dict:
    """Module 7 lesson 2's criterion: low cardinality (few folders),
    never a near-unique key like franchise_id."""
    candidates_ok = {c: n for c, n in cardinalities.items() if n <= max_folders}
    if not candidates_ok:
        return {"chosen": None, "reason": "no candidate has low cardinality"}
    chosen = min(candidates_ok, key=candidates_ok.get)
    discarded = {c: n for c, n in cardinalities.items() if c != chosen}
    return {
        "chosen": chosen,
        "chosen_cardinality": cardinalities[chosen],
        "discarded": discarded,
        "reason": f"{chosen} has {cardinalities[chosen]} distinct values -- few folders, "
                 f"each with reasonably sized files",
    }


partition_decision = choose_partition_column(cardinalities)
print(f"\nPartitioning decision: {partition_decision}\n")
assert partition_decision["chosen"] == "store_id"

Notice the max_folders=50 threshold: it's a deliberate choice, not an arbitrary one. With 3 folders (store_id) or 4 (product_id), a filesystem handles the organization with no trouble at all. With 250,000 folders (franchise_id), each containing, on average, 40 rows — a few hundred bytes — the filesystem would pay an enormous metadata cost for tiny files, exactly the bad decision module 7's lesson 2 already explicitly ruled out for this same dataset.

Part 2 — JOIN criterion: real size against the threshold

print("Part 2 -- JOIN criterion: each dimension table's real size, against the threshold")
threshold_bytes = int(spark.conf.get("spark.sql.autoBroadcastJoinThreshold").rstrip("b"))
dim_sizes = {
    "dim_store.csv": os.path.getsize("dim_store.csv"),
    "dim_product.csv": os.path.getsize("dim_product.csv"),
}
print(f"  spark.sql.autoBroadcastJoinThreshold = {threshold_bytes:,} bytes")
for name, size in dim_sizes.items():
    print(f"  {name}: {size:,} bytes")


def choose_join_strategy(table_size_bytes: int, threshold_bytes: int) -> str:
    """Module 5 lesson 2's criterion: if the JOIN's smaller table fits
    under the threshold, Spark photocopies it (broadcast); if not, a
    shuffle is needed (sort-merge)."""
    if table_size_bytes <= threshold_bytes:
        return "BroadcastHashJoin"
    return "SortMergeJoin (shuffle)"


for name, size in dim_sizes.items():
    strategy = choose_join_strategy(size, threshold_bytes)
    print(f"  Strategy for {name}: {strategy}")
    assert strategy == "BroadcastHashJoin"

Part 3 — Caching criterion: real reuse, not a hunch

print("\nPart 3 -- caching criterion: how many real queries reuse the same DataFrame")
REUSE_COUNT = 3  # lesson 3's three queries: total, per store, running_total


def should_cache_df(reuse_count: int, min_reuses: int = 2) -> dict:
    """Module 6 lesson 5's criterion: only cache if the DataFrame gets
    reused more than once -- a single use never justifies the memory
    cost."""
    if reuse_count >= min_reuses:
        return {"cache": True, "reason": f"reused {reuse_count} times, >= {min_reuses}"}
    return {"cache": False, "reason": f"used {reuse_count} time(s), doesn't justify the memory cost"}


cache_decision = should_cache_df(REUSE_COUNT)
print(f"  reuse_count = {REUSE_COUNT}")
print(f"  Caching decision: {cache_decision}")
assert cache_decision["cache"] is True

print("\n=== This pipeline's three decisions, all with measured evidence, none out of habit ===")
print(f"  Write partition: store_id ({cardinalities['store_id']} values) -- NOT franchise_id ({cardinalities['franchise_id']:,} values)")
print(f"  JOIN strategy: BroadcastHashJoin for dim_store and dim_product (both < {threshold_bytes:,} bytes)")
print(f"  Cache: YES, fact_orders_at_scale_df gets reused {REUSE_COUNT} times in lesson 3")
spark.stop()

What to expect. Running python3 kiosko_design_criteria.py (executed in this run):

=== Module 8, lesson 5: partitioning, caching, and joins, with criteria and evidence ===

Part 1 -- partitioning criterion: each candidate's real cardinality
  cardinality of store_id: 3
  cardinality of product_id: 4
  cardinality of franchise_id: 250000

Partitioning decision: {'chosen': 'store_id', 'chosen_cardinality': 3, 'discarded': {'product_id': 4, 'franchise_id': 250000}, 'reason': 'store_id has 3 distinct values -- few folders, each with reasonably sized files'}

Part 2 -- JOIN criterion: each dimension table's real size, against the threshold
  spark.sql.autoBroadcastJoinThreshold = 10,485,760 bytes
  dim_store.csv: 100 bytes
  dim_product.csv: 200 bytes
  Strategy for dim_store.csv: BroadcastHashJoin
  Strategy for dim_product.csv: BroadcastHashJoin

Part 3 -- caching criterion: how many real queries reuse the same DataFrame
  reuse_count = 3
  Caching decision: {'cache': True, 'reason': 'reused 3 times, >= 2'}

=== This pipeline's three decisions, all with measured evidence, none out of habit ===
  Write partition: store_id (3 values) -- NOT franchise_id (250,000 values)
  JOIN strategy: BroadcastHashJoin for dim_store and dim_product (both < 10,485,760 bytes)
  Cache: YES, fact_orders_at_scale_df gets reused 3 times in lesson 3

Pause on the scale gap between the three cardinalities: store_id with 3 values and product_id with 4 would, either one, comfortably fit under the criterion's 50-folder threshold. franchise_id, with 250,000 values, exceeds it by more than four orders of magnitude. No subjective judgment is needed to rule out franchise_id as a partition column — the number rules it out on its own.

Diagram: three decisions, three criteria, one pipeline

flowchart TD
    subgraph Particion["Decision 1: partitioning"]
        A1["Candidates: store_id (3),\nproduct_id (4), franchise_id (250,000)"] --> A2["Criterion: cardinality <= 50"] --> A3["Chosen: store_id"]
    end
    subgraph Join["Decision 2: JOIN strategy"]
        B1["dim_store (100 bytes),\ndim_product (200 bytes)"] --> B2["Criterion: size <= 10,485,760 bytes"] --> B3["Chosen: BroadcastHashJoin"]
    end
    subgraph Cache["Decision 3: caching"]
        C1["fact_orders_at_scale_df\nreused 3 times"] --> C2["Criterion: reuse_count >= 2"] --> C3["Chosen: cache"]
    end
    A3 --> D["Lesson 3's pipeline,\nevery decision backed by evidence"]
    B3 --> D
    C3 --> D

Going deeper: what happens when the evidence changes

Coding these three criteria as functions, instead of leaving them as decisions made once, has real value: they get re-evaluated automatically if the evidence changes. Imagine Kiosko, in the future, decided to open its franchise model to the franchisees themselves, and each one could choose their own pricing structure — suddenly, dim_product could grow from 4 rows to several thousand, with one row per franchise and product. choose_join_strategy(), run again with the new dim_product.csv's real size, would automatically answer whether that table still fits under the broadcast threshold or whether, past a certain size, Spark would have to start using SortMergeJoin.

This is the real difference between a design decision made once, from memory, and a coded criterion: the criterion doesn't need anyone to remember "why we chose this" — it remeasures the evidence every time it runs, and produces the correct verdict for the data in front of it at that moment, not for the data that existed when someone first made the decision.

Common mistakes

Copying choose_partition_column()'s max_folders=50 threshold without understanding it's a reference value, not a fixed Spark rule. What happens: someone assumes 50 is a number Spark imposes internally, instead of a reasonable criterion chosen for this specific case. Why it happens: seeing a specific number inside a technically named function invites treating it as an official constant. How to spot it: if you search for 50 in Spark's official documentation as a limit related to partitionBy(), you won't find it — Spark imposes no fixed limit on the number of partition folders; the cost of having too many only shows up as small files and metadata overhead, a performance problem, not a Spark error. How to fix it: treat max_folders=50 as a reasonable engineering criterion for this specific dataset's size, not a magic value — a dataset with a different query pattern, or with far more rows per partition, could justify a different threshold.

Confusing dim_sizes (the CSV files' size on disk) with the real size Spark uses to decide the broadcast. What happens: someone assumes the 100 and 200 bytes measured in this lesson are exactly the number Spark internally compares against autoBroadcastJoinThreshold. Why it happens: this lesson measures the file's size on disk as a simple, verifiable approximation. How to spot it: if you check lesson 3's (or module 6's) .explain() plan, you're going to see statistics like sizeInBytes=8.0 MiB for these same tables — much larger than the 100/200 bytes on disk — because Spark estimates the size in memory, after deserializing the CSV, not the file's compressed size. How to fix it: the size on disk is a useful, sufficient approximation for this lesson's criterion — both numbers, on-disk and estimated-in-memory, sit far below the 10 MB threshold regardless — but for an edge case, close to the threshold, Spark's real estimate (visible in .explain(mode="cost"), module 6's topic) would be the correct reference, not the file's size.

Applying should_cache_df() to a DataFrame reused within a single action, not across separate actions. What happens: someone counts as "reuse" every time a DataFrame's column shows up in an expression, even within the same query — df.select("a", "b").filter(col("a") > 1), for example, would use column a "twice" — and concludes that DataFrame needs caching. Why it happens: the word "reuse" can be interpreted, imprecisely, as any repeated reference to a column or table. How to spot it: if your reuse_count includes references within a single chain of transformations ending in a single action, you're overestimating real reuse. How to fix it: this lesson's criterion — and module 6's lesson 5's — counts separate actions consuming the same already-materialized DataFrame (like lesson 3's Part 5's three queries, each ending in its own .collect()), not column references within a single chain of lazy transformations.

Exercises

Exercise 1 — Apply choose_partition_column() to a hypothetical scenario where Kiosko had a region column with 12 distinct values. Without running Spark, add "region": 12 to the cardinalities dictionary and call the function again. Does the result change?

hypothetical_cardinalities = {"store_id": 3, "product_id": 4, "franchise_id": 250_000, "region": 12}
print(choose_partition_column(hypothetical_cardinalities))

{'chosen': 'store_id', 'chosen_cardinality': 3, 'discarded': {'product_id': 4, 'franchise_id': 250000, 'region': 12}, 'reason': 'store_id has 3 distinct values -- few folders, each with reasonably sized files'}

Exercise 2 — Confirm choose_join_strategy() would change its verdict if dim_product.csv grew past the threshold. Without modifying the real file, call the function with a hypothetical size of 20 * 1024 * 1024 bytes (20 MB) instead of the real 200-byte size.

hypothetical_strategy = choose_join_strategy(20 * 1024 * 1024, threshold_bytes)
print(f"Strategy if dim_product weighed 20 MB: {hypothetical_strategy}")
assert hypothetical_strategy == "SortMergeJoin (shuffle)"

Strategy if dim_product weighed 20 MB: SortMergeJoin (shuffle)

Exercise 3 — Explain, without code, why this lesson measures reuse_count with a fixed value (3) instead of counting it automatically inside the script. In 2-3 sentences, argue why automatically counting how many times a DataFrame gets reused within a pipeline is harder than it looks, and why this lesson declares it explicitly instead.

Summary and next step

In this lesson you turned three of lesson 3's pipeline design decisions — which column to partition by, which JOIN strategy to expect, whether caching is worth it — into three runnable functions, each with real evidence measured over kiosko_orders_at_scale: store_id (3 values) against franchise_id (250,000 values), dim_store/dim_product (100 and 200 bytes) against the broadcast threshold (10,485,760 bytes), and a real three-query reuse justifying the cache. Not a single decision in this pipeline was left unjustified.

Before moving on you should be able to: explain this lesson's three functions' criteria from memory; explain why franchise_id was never a serious candidate for the partition column; and explain the difference between a file's size on disk and the estimated size Spark uses to decide a broadcast.

Lesson 6 — this entire guide's pedagogical heart — takes one more step back: not "how to design a good Spark pipeline" anymore, but "was Spark even needed, to begin with?" It applies module 1's same should_distribute() to the real Kiosko, to this guide's synthetic dataset, and to a much larger hypothetical Kiosko, with concrete numbers for each.

Resources

• Apache Spark — SQL Performance Tuning (autoBroadcastJoinThreshold, partitioning, caching — the unified reference for this lesson's three decisions). spark.apache.org/docs/latest/sql-performance-tuning.html.
• Apache Spark — SQL Data Sources: Parquet (folder-based partitioning and its metadata cost at high cardinality, the foundation for this lesson's Part 1 criterion). spark.apache.org/docs/latest/sql-data-sources-parquet.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — this module's brief's Deliverable 3, the exact source for the three decisions this lesson justifies. src/guides/spark-and-distributed-processing-guide/DISENO.md.