📖 The 45-Minute Join Stage That Became 90 Seconds

A data engineering team at a retail company was running a nightly Spark job that joined their 500 GB transaction fact table against a 50 MB product dimension table. The job had been in production for six months. Whenever the business added a new data source, the join stage crept up — first 20 minutes, then 35, eventually settling into a reliable 45-minute runtime that delayed every downstream report.

A senior engineer picked up the ticket and looked at the Spark UI. The join stage had staggering metrics: 500 GB shuffle write, 500 GB shuffle read, 200 parallel sort tasks on both sides. Spark had planned a Sort-Merge Join — the default for large tables — and was faithfully shuffling all 500 GB of the fact table and all 50 MB of the dimension table across the cluster, sorting both by the join key, and then scanning them together. The dimension table was 50 MB. It was being shuffled alongside a table 10,000 times its size.

The fix was a single word added to the query: broadcast. The job dropped to 90 seconds.

That 30x improvement was not a configuration miracle. It was the result of Spark switching join strategies — from a sort-merge join that pays a full double-shuffle penalty, to a broadcast hash join that eliminates the network transfer for the small side entirely. Understanding why Spark chose the wrong strategy, what each strategy costs, and when to intervene is one of the highest-leverage skills in Spark performance engineering. A join strategy mistake can turn a two-minute job into a two-hour one. Getting it right is often the single biggest improvement available to a job without rewriting any logic.

This post walks through all five join strategies Spark knows, how Catalyst picks between them, what happens inside each strategy at the executor level, and how to read the physical plan to verify which one was actually chosen.

---

🔍 The Five Spark Join Strategies and When Catalyst Selects Each

Spark's Catalyst optimizer knows five join strategies. They differ in how much data moves across the network, how much memory they require, and what classes of join they can handle. Catalyst selects among them using a rule cascade applied to size estimates at planning time — before a single row of data is read.

Broadcast Hash Join (BHJ)

Catalyst selects BHJ when one side of the join has a size estimate below spark.sql.autoBroadcastJoinThreshold (default 10 MB). The small table is serialized by the driver and sent to every executor as a broadcast variable — a shared, read-only JVM object. Each executor builds a hash map from the broadcast table and probes it with each row of its local partition of the large table. No shuffle occurs on either side.

When Catalyst picks it: Estimated size of at least one join side ≤ autoBroadcastJoinThreshold. This estimate comes from table statistics (ANALYZE TABLE) or from Parquet/ORC file metadata. If statistics are stale or absent, Catalyst cannot confirm the table is small and falls back to Sort-Merge Join — even if the table is 20 MB in reality.

Shuffle Hash Join (SHJ)

SHJ shuffles both tables by the join key (one network pass each), then builds a hash map from the smaller partition on each executor and probes it with the larger partition. Unlike SMJ, there is no sort step. It is faster than SMJ per partition, but the hash map for the build side must fit in the executor's available memory per partition. SHJ is disabled by default in Spark 3.x (spark.sql.join.preferSortMergeJoin=true). It must be enabled explicitly and used only when you know the per-partition size of the smaller side is manageable.

When Catalyst picks it: preferSortMergeJoin=false AND estimated build-side size × partition ratio fits in executor memory.

Sort-Merge Join (SMJ)

SMJ is the default and most scalable equi-join strategy. Both tables are shuffled by the join key, producing co-partitioned datasets. Each partition is then sorted by the key, and a merge scan sweeps through both sorted lists simultaneously in a single linear pass. SMJ scales to arbitrarily large tables because it never requires the entire partition to fit in memory — it streams through data. The cost is two full shuffles plus two sorts: significant network I/O and disk I/O that dominates join latency at scale.

When Catalyst picks it: All other strategies are ruled out (table too large to broadcast, SHJ disabled or ruled out, no bucket co-partitioning).

Broadcast Nested Loop Join (BNLJ)

BNLJ handles non-equi joins — join conditions that use inequality operators (<, >, BETWEEN, LIKE) rather than equality. One side is broadcast (when small enough); the other side's partitions iterate over every row in the broadcast table. This is an O(n × m) operation and is only viable when the broadcast side is tiny.

When Catalyst picks it: No equijoin condition exists AND one side can be broadcast.

Cartesian Join (Cross Join)

A cartesian product has no join condition at all — every row from the left table is paired with every row from the right table. Two 1-million-row tables produce 1 trillion output rows. Spark requires an explicit .crossJoin() call or spark.sql.crossJoin.enabled=true to plan this. The most common accidental path to a cartesian join is a WHERE clause that Catalyst cannot push into the join condition, leaving it with no equijoin predicate.

When Catalyst picks it: No join condition is present and cross-join is explicitly enabled.

---

⚙️ How BHJ, SMJ, and SHJ Execute at the Executor Level

Broadcast Hash Join: The Driver Collects, Executors Probe

The BHJ execution flow has three phases and no shuffle:

Phase 1 — Collection and serialization. The driver (or an executor acting as coordinator) collects the entire small table from all partitions into a single dataset, serializes it using Java serialization (or Kryo if configured), and stores it as a broadcast variable in Spark's broadcast manager. The size of the serialized object is what gets checked against autoBroadcastJoinThreshold.

Phase 2 — Distribution. Spark distributes the broadcast variable to all executors. The distribution uses a BitTorrent-like protocol for large broadcasts: executors receive chunks from each other rather than all pulling from the driver, which prevents the driver from becoming a network bottleneck when hundreds of executors request the same object simultaneously.

Phase 3 — Local hash probe. Each executor deserializes the broadcast variable once per JVM (not once per task) and builds a HashedRelation — Spark's internal compact hash map optimized for join lookups. For long-type join keys, Spark uses LongToUnsafeRowMap, a flat array-based structure that avoids boxing overhead. For composite keys, it uses UnsafeHashedRelation, which stores keys and values in Tungsten off-heap memory. Each task then scans its partition of the large table and probes the hash map locally. The entire join happens without any data crossing the network for the large table.

Sort-Merge Join: Double Shuffle, Double Sort, Linear Merge

SMJ involves three phases and two full network shuffles:

Phase 1 — Shuffle (both sides). An Exchange operator is inserted into the plan for each join side. Each executor hashes every row's join key modulo the number of shuffle partitions and writes rows to per-partition shuffle files. Once all map tasks complete, reduce tasks pull the corresponding partition from every executor — this is the full network transfer.

Phase 2 — Sort (both sides). Each reduce task receives its slice of rows from both tables. It sorts each slice independently by the join key. For a 200-partition join on a 500 GB table, each partition receives ~2.5 GB of data to sort. Sorting uses Spark's TimSort implementation on Tungsten memory pages. If a partition's data exceeds memory, it spills to disk and a merge-sort is performed on the spilled chunks.

Phase 3 — Merge scan. With both sides sorted by key, the merge scan maintains two cursors — one in each sorted list — advancing through matching keys together. This linear scan is the cheapest part of the entire operation and is why SMJ scales to any dataset size despite its shuffle cost.

Shuffle Hash Join: One Shuffle, No Sort

SHJ shuffles both tables exactly once (same Exchange operator as SMJ), then builds a hash map from the designated build side (the smaller partition) and probes it with the probe side (the larger partition). The absence of a sort step makes SHJ faster per partition than SMJ when the build-side partition fits comfortably in memory. When it does not fit — due to data skew or misconfigured partition counts — SHJ spills the hash map to disk and performance degrades sharply.

---

🧠 Inside the Broadcast Variable and How Join Strategy Performance Actually Scales

The Internals of Broadcast Hash Join

The broadcast variable lifecycle has subtle memory implications that are worth understanding in detail. When the driver serializes the small table, the serialized byte array is held in driver memory until all executors have acknowledged receipt. For a 200-executor cluster broadcasting a 200 MB table, the driver allocates 200 MB and holds it for the duration of the job. The executors each deserialize their own copy — 200 MB × 200 executors = 40 GB of cluster memory consumed by the broadcast before a single join row is processed. This is why the practical safe limit for broadcast joins is often far below the theoretical threshold of autoBroadcastJoinThreshold.

The HashedRelation that executors build from the broadcast data adds overhead beyond the raw table size. A hash map requires additional memory for its bucket array, collision chains, and pointer overhead. A 200 MB serialized table may produce a 600 MB HashedRelation in memory. Spark uses Tungsten's off-heap UnsafeHashedRelation to minimize JVM GC pressure, but the memory consumption is real.

One key implementation detail: the broadcast variable is not copied per task. All tasks running on the same executor JVM share a single reference to the deserialized HashedRelation. This is what makes BHJ so efficient for high-concurrency workloads — a 32-core executor running 32 concurrent tasks probes the same hash map 32 times in parallel, consuming no additional memory per task.

When the small table is filtered heavily before the join, Catalyst may not know the post-filter size at planning time. The broadcast() hint bypasses size estimation and forces BHJ regardless. This is the correct tool when you know a table will be small after a filter that reduces it from 5 GB to 20 MB.

Performance Analysis: Comparing Join Strategies by Real Cost

Adaptive Query Execution (AQE) changes this picture at runtime. With spark.sql.adaptive.enabled=true, Spark collects actual partition statistics after each shuffle completes and re-optimizes the downstream plan using that real data rather than estimates. The most impactful AQE behavior for joins is dynamic join strategy switching: if a sort-merge join was planned because Catalyst underestimated a table's size, but the actual shuffle output for one side is small, AQE can switch to BHJ at runtime — pulling that side into a broadcast variable and replanning the join operator without restarting the job.

AQE also handles skew join mitigation automatically. When spark.sql.adaptive.skewJoin.enabled=true (default in Spark 3.x), AQE detects partitions that are substantially larger than the median partition size and splits them into sub-tasks, then duplicates the corresponding partition from the other side to join against each sub-task. This eliminates the straggler problem that causes SMJ to appear to hang even when most of the join completes quickly.

---

📊 Broadcast Hash Join vs Sort-Merge Join: Execution Path Comparison

The two diagrams below capture the fundamental structural difference between BHJ and SMJ. In BHJ, data flows from the large table directly into a local hash probe with no network transfer. In SMJ, both sides pay a full shuffle penalty before any join work can begin. Understanding this structure makes physical plan output immediately readable.

graph TD
    subgraph BHJ[Broadcast Hash Join Path]
        A1[Small Table partitions] --> B1[Collect to Driver]
        B1 --> C1[Serialize as broadcast variable]
        C1 --> D1[Distribute to all Executors]
        D1 --> E1[Each Executor builds HashedRelation]
        F1[Large Table partition on Executor] --> G1[Probe HashedRelation locally]
        E1 --> G1
        G1 --> H1[Joined rows output - no shuffle]
    end

    subgraph SMJ[Sort-Merge Join Path]
        A2[Left Table partitions] --> B2[Exchange - shuffle by join key hash]
        A3[Right Table partitions] --> C2[Exchange - shuffle by join key hash]
        B2 --> D2[Sort left partition by key]
        C2 --> E2[Sort right partition by key]
        D2 --> F2[Merge scan - two cursors advance together]
        E2 --> F2
        F2 --> G2[Joined rows output - two shuffles paid]
    end

In the BHJ path (top subgraph), the small table travels once from the driver to all executors as a broadcast variable. The large table never moves — each executor probes the local hash map using its own partition. In the SMJ path (bottom subgraph), both tables undergo a full Exchange (shuffle) to co-partition rows by join key. Each co-located pair of partitions is then sorted, and the merge scan is the final linear pass. The two Exchange nodes in the SMJ plan are the dominant cost in any large-table join.

---

🌍 Controlling Join Strategy in Production: Hints, Bucketing, and AQE Tuning

Forcing a Broadcast When Catalyst Underestimates

The most common production scenario requiring manual intervention: a table is filtered from 2 GB to 30 MB before the join, but Catalyst planned the join before knowing the filter would reduce the table so dramatically. Because statistics were collected on the full 2 GB table, Catalyst estimates 2 GB and plans SMJ. The join runs for 40 minutes when it should take 2.

The broadcast() hint wraps a DataFrame and instructs Catalyst to force BHJ regardless of its size estimate. It is the correct tool here — not a workaround, but an explicit expression of domain knowledge the optimizer does not have. The hint is respected as long as the resulting broadcast is actually feasible (the table can be collected to the driver and the memory is available).

For Spark SQL queries, the equivalent hint is /*+ BROADCAST(tableName) */ inside the SQL string. Both forms annotate the query plan and survive Catalyst's optimization passes without being stripped out.

Disabling Broadcast for Memory-Constrained Clusters

On clusters where executors have limited heap (4 GB or less), even a 50 MB broadcast table may cause memory pressure when combined with other in-flight operations. Setting spark.sql.autoBroadcastJoinThreshold=-1 disables automatic broadcast for all joins in that session. Catalyst will plan SMJ or SHJ for all joins regardless of table size. This trades join latency for memory safety — the right call when stability under concurrent load matters more than individual query speed.

Bucketed Joins: Eliminating the Shuffle Phase from Sort-Merge Joins

Bucketing is the highest-value optimization for repeated joins on the same key. If two tables are written to disk as bucketed Hive tables using the same key and the same bucket count, Spark knows that matching rows are already co-located on disk. The Exchange operators in the SMJ plan are eliminated entirely — the shuffle never happens. A join that normally costs two full network shuffles on 500 GB tables reduces to a pure sort-and-merge on pre-partitioned data.

The requirement is strict: both tables must be bucketed on the same column(s) with the same number of buckets, and both must be read from a Hive metastore (not inferred from files directly). When these conditions hold, df.explain() will show SortMergeJoin with no Exchange operators above it — the tell-tale sign of a shuffle-free bucketed join.

AQE Skewed Join Handling: Splitting Straggler Partitions Automatically

Join stages with skewed key distributions produce straggler partitions — one or two tasks processing 100 GB while the other 198 tasks process 500 MB each. The stage cannot complete until every task finishes, so 98% of the cluster idles waiting for 2% of the work. AQE's skew join handling detects these imbalanced partitions using actual shuffle output sizes and automatically splits large partitions into multiple smaller sub-tasks. Each sub-task processes a key-range slice of the skewed partition, with the corresponding range from the other side duplicated to match. The straggler effectively disappears from the stage timeline.

---

⚖️ When Each Strategy Becomes a Liability

Broadcast Hash Join: Memory Risk Multiplied by Executor Count

BHJ's memory consumption scales with the number of executors, not the number of partitions. Broadcasting a 200 MB table to a 300-executor cluster requires 60 GB of cluster memory to be allocated simultaneously — 200 MB per executor, held for the entire duration of the join stage. On a shared cluster where multiple jobs run concurrently, each with its own broadcast joins, executor OOM errors become likely long before any individual broadcast exceeds the threshold.

The driver is also vulnerable: it must collect and hold the entire broadcast table in heap memory before distribution begins. A table that is 800 MB after collection (even if compressed on disk) will fail at the collection step if the driver's heap is smaller than 800 MB.

Sort-Merge Join: Network and Disk Amplification

For a 500 GB fact table joined to a 10 GB dimension table with 400 shuffle partitions, SMJ generates approximately 510 GB of shuffle write and 510 GB of shuffle read — 1 TB of total I/O for a join that produces 500 GB of output. Every row crosses the network twice (once written, once read), then gets sorted. On clusters with 10 Gbps network cards, 1 TB of shuffle traffic takes a minimum of 800 seconds at wire speed, assuming perfectly balanced partitions and no disk contention. Real-world numbers are typically 2–4x the theoretical minimum due to serialization overhead, disk write speed, and network scheduling.

Data skew amplifies SMJ's cost dramatically. A join key with 10% of all rows sharing the same value means one reduce partition receives 50 GB when the average is 1.25 GB. That one task runs for 40 times longer than the average, holding the entire stage open.

Shuffle Hash Join: The Sweet Spot with a Narrow Window

SHJ outperforms SMJ when the build-side partition fits comfortably in memory, because it avoids the sort step. But "fits comfortably" is a narrow window. If the build-side partition is 800 MB and executor memory is 4 GB, with multiple concurrent tasks and other cached data competing for the same heap, SHJ will spill the hash map to disk — and a spilling SHJ is slower than SMJ because its disk access pattern is random rather than sequential.

SHJ is disabled by default in Spark 3.x precisely because of this fragility. Enabling it without careful per-partition size analysis frequently causes hash map spills on skewed datasets, producing worse performance than the SMJ it was meant to replace.

Cartesian Join: The Accidental Production Killer

A cartesian join is almost never intentional. The most common path to one: a SQL query where the join condition references a column through an alias that Catalyst cannot resolve to the join predicate, leaving the planner with no equijoin condition. The query appears to run normally for small development datasets (two tables with 1000 rows each produce 1 million rows — manageable). The same query on production datasets with 1 million rows each produces 1 trillion rows — which immediately exhausts executor memory and crashes the stage.

Always verify that df.explain() shows SortMergeJoin or BroadcastHashJoin for join operations — never CartesianProduct — before promoting a Spark job to production.

---

🧭 Choosing the Right Join Strategy: A Size and Memory Decision Guide

This table maps the key variables — small-side table size, large-side table size, and available executor memory — to the recommended join strategy. Use it as a first-pass decision before reading the physical plan.

When in doubt between BHJ (with hint) and SMJ on a medium-sized table: run df.explain(), look for BroadcastExchange in the plan to confirm the broadcast variable was actually built, and check Stage metrics in the Spark UI for shuffle read/write bytes. Zero shuffle read bytes on the probe side confirms BHJ executed as intended.

---

🧪 Reading the Physical Plan to Verify Your Join Strategy

The physical plan output from df.explain() is the ground truth for which join strategy Spark actually chose. Before running any join-heavy job in production, call explain() and confirm the plan matches your intent. The following examples show what to look for in each case.

This section demonstrates three scenarios: confirming a successful broadcast hash join, identifying an accidental sort-merge join on a small table, and spotting the most dangerous outcome — an unintended cartesian product. Each plan is a lightly simplified version of real Spark output.

Scenario 1 — Broadcast Hash Join confirmed. When BHJ was planned and executed correctly, the physical plan contains BroadcastHashJoin with a BroadcastExchange operator feeding the build side:

== Physical Plan ==
*(2) BroadcastHashJoin [transaction_id#10], [txn_id#42], Inner, BuildRight
:- *(2) Project [transaction_id#10, merchant_id#11, amount#12]
:  +- *(2) Scan parquet fact_transactions [...]
+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]))
   +- *(1) Project [txn_id#42, product_name#43, category#44]
      +- *(1) Scan parquet dim_products [...]

The BroadcastExchange on the right side confirms the dimension table was collected and broadcast. The absence of any Exchange operator above the fact table scan confirms no shuffle occurred on the probe side.

Scenario 2 — Sort-Merge Join on a small table (Catalyst missed the size). When statistics are stale or absent and Catalyst cannot confirm the dimension table is small, you see:

== Physical Plan ==
*(5) SortMergeJoin [transaction_id#10], [txn_id#42], Inner
:- *(2) Sort [transaction_id#10 ASC NULLS FIRST], false, 0
:  +- Exchange hashpartitioning(transaction_id#10, 200)
:     +- *(1) Scan parquet fact_transactions [...]
+- *(4) Sort [txn_id#42 ASC NULLS FIRST], false, 0
   +- Exchange hashpartitioning(txn_id#42, 200)
      +- *(3) Scan parquet dim_products [...]

Two Exchange operators and two Sort operators confirm both sides are being shuffled and sorted. To fix this, add broadcast() around the dimension table reference, or run ANALYZE TABLE dim_products COMPUTE STATISTICS to give Catalyst accurate size data.

Scenario 3 — Accidental CartesianProduct. The most dangerous plan output, and the one to catch before production:

== Physical Plan ==
CartesianProduct
:- *(1) Scan parquet fact_transactions [...]
+- *(2) Scan parquet dim_products [...]

No join key. No Exchange. No Sort. Just every row paired with every row. This plan will OOM the cluster on any production-scale table. Trace back to the query and verify the join condition — check for aliasing issues, ambiguous column references, or a missing .on() / WHERE predicate.

---

🛠️ Apache Spark Join Configuration Reference

The following properties directly control join strategy selection. All can be set per-session in SparkConf, passed via --conf at spark-submit, or applied dynamically using spark.conf.set().

# ─────────────────────────────────────────────────────────────────
# Join Strategy Thresholds
# ─────────────────────────────────────────────────────────────────

# Maximum size (bytes) for auto-broadcast. Default: 10MB (10485760).
# Set to -1 to disable all automatic broadcasting.
spark.sql.autoBroadcastJoinThreshold=10485760

# When true, Catalyst prefers Sort-Merge Join over Shuffle Hash Join.
# Default: true (Spark 3.x). Set to false to allow SHJ when applicable.
spark.sql.join.preferSortMergeJoin=true

# ─────────────────────────────────────────────────────────────────
# Adaptive Query Execution (AQE)
# ─────────────────────────────────────────────────────────────────

# Enables runtime plan re-optimization using actual partition statistics.
# Default: true (Spark 3.2+). Enables dynamic join strategy switching.
spark.sql.adaptive.enabled=true

# Enables automatic splitting of skewed join partitions.
# Default: true when adaptive.enabled=true.
spark.sql.adaptive.skewJoin.enabled=true

# Partitions larger than this multiple of the median size are considered skewed.
# Default: 5. Reduce to 3 for aggressively skewed keys.
spark.sql.adaptive.skewJoin.skewedPartitionFactor=5

# Minimum size (bytes) for a partition to be considered skewed.
# Default: 256MB. Prevents false-positive splits on small jobs.
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=268435456

# ─────────────────────────────────────────────────────────────────
# Shuffle Partition Tuning
# ─────────────────────────────────────────────────────────────────

# Number of shuffle partitions for joins and aggregations.
# Default: 200. Tune to (total shuffle data in bytes) / (100–200MB per partition).
spark.sql.shuffle.partitions=200

These five properties cover the majority of join performance tuning scenarios. For a full deep-dive on AQE's runtime plan rewriting, including coalescing small partitions and converting sort-merge joins to broadcast hash joins at runtime, see the Apache Spark 3.x Configuration Guide under the spark.sql.adaptive.* namespace.

---

📚 Lessons Learned From Real Spark Join Failures

Stale statistics cause more BHJ misses than anything else. The most common reason Spark plans SMJ on a small table is not a configuration problem — it is missing or stale table statistics. After any significant data load, run ANALYZE TABLE ... COMPUTE STATISTICS FOR COLUMNS ... on the dimension tables that participate in broadcast-eligible joins. Catalyst's size estimates are only as accurate as the statistics it has access to.

The broadcast() hint is not a workaround — it is an API. Many engineers treat it as a hack applied out of desperation. It is actually the correct way to encode domain knowledge that the optimizer cannot derive from statistics alone — particularly for tables that shrink dramatically under filters applied in the same query.

Executor memory planning must account for concurrent broadcasts. On a 200-executor cluster where three concurrent jobs each broadcast a 100 MB table, the cluster must hold 60 GB of broadcast data simultaneously. This is invisible in per-job monitoring but shows up as executor OOM at cluster scale. Set autoBroadcastJoinThreshold conservatively on shared clusters — 50 MB is often safer than the default 10 MB when you have many jobs running concurrently.

AQE is not optional on Spark 3.x jobs. With spark.sql.adaptive.enabled=true (the default since Spark 3.2), AQE handles a class of performance problems automatically — skewed partitions, small post-shuffle partition coalescing, and dynamic BHJ conversion — that previously required manual tuning. Disabling it to "simplify" the plan is almost always counterproductive. Leave it on and learn to read AQE-reoptimized plans.

Always validate the physical plan before production promotion. A df.explain() call before the first production run costs nothing. Catching a CartesianProduct or an unexpected SortMergeJoin on a known-small table in development costs nothing. Discovering either in production after a 4-hour stage timeout costs the entire pipeline SLA.

Bucketed joins are underused. For ETL pipelines that repeatedly join the same two large tables on the same key, bucketing the fact and dimension tables with the same bucket count eliminates the shuffle on every run. The one-time cost of rewriting the tables in bucketed form is recovered within a handful of pipeline executions. This is particularly high-value for daily jobs that run the same join 365 times a year.

---

📌 Summary: What the Join Strategy Choice Actually Costs

TLDR: Spark knows five join strategies; Catalyst picks the wrong one whenever it lacks accurate table size statistics. Broadcast Hash Join is 30× faster than Sort-Merge Join for small dimension tables because it eliminates both shuffles entirely — the small table travels once as a broadcast variable, and every executor joins locally. Sort-Merge Join scales to any dataset size but pays two full shuffles plus two sorts on every run. Use broadcast() hints when Catalyst underestimates filtered table sizes, enable AQE for dynamic strategy switching at runtime, and always call df.explain() before promoting a join-heavy job to production. The 45-minute stage is almost always a missing broadcast hint on a 50 MB table.

The three actions with the highest ROI on join performance:

1. Run ANALYZE TABLE ... COMPUTE STATISTICS on all dimension tables after data loads — gives Catalyst accurate size estimates for automatic BHJ selection.
2. Add broadcast() hints to DataFrames that are small after filtering even if they are large on disk — encodes knowledge the optimizer cannot derive.
3. Enable spark.sql.adaptive.enabled=true and spark.sql.adaptive.skewJoin.enabled=true — handles the runtime edge cases (skewed keys, post-filter size surprises) that static planning cannot anticipate.

---

📝 Practice Quiz: Test Your Join Strategy Knowledge

1. A Spark job joins a 500 GB fact table with a 15 MB dimension table. The job plans a Sort-Merge Join and runs for 45 minutes. What is the most likely cause, and what is the fix?

   Answer

   Correct Answer: The most likely cause is stale or missing table statistics. Catalyst uses size estimates to decide whether to broadcast, and if the dimension table's statistics were never collected (or collected before a recent load), Catalyst sees an unknown or overestimated size and falls back to SMJ. The fix has two parts: (1) run ANALYZE TABLE dim_table COMPUTE STATISTICS FOR COLUMNS join_key to give Catalyst accurate size data, and (2) add a broadcast() hint as an immediate override while statistics are being collected.

2. You enable spark.sql.join.preferSortMergeJoin=false to allow Shuffle Hash Join on a 10 GB × 800 GB join. After two hours, the join stage is still running and Spark UI shows heavy spill-to-disk. What went wrong?

   Answer

   Correct Answer: The build-side partitions exceeded the available executor memory per task. With 200 shuffle partitions, each partition of the 10 GB build side is approximately 50 MB. However, data skew likely caused some partitions to be much larger — potentially several GB — which exceeded executor heap space and triggered hash map spills. SHJ degrades to random-access disk I/O when spilling, which is slower than SMJ's sequential merge scan. The fix: keep preferSortMergeJoin=true, use AQE's skew join handling, or increase spark.sql.shuffle.partitions to reduce per-partition size.

3. Your physical plan shows CartesianProduct for a query joining two tables on order_id. Neither table is empty. What is the most common reason for this, and how do you diagnose it?

   Answer

   Correct Answer: The most common cause is that the join condition could not be resolved as an equijoin predicate — typically due to a column naming conflict, an aliasing issue, or the join condition being expressed in a WHERE clause that Catalyst couldn't push into the join operator. Diagnose by examining the parsed logical plan (df.explain(true) shows all plan stages), looking at the Unresolved or Filter operators for the condition you intended as the join key, and verifying column names match exactly (including case) on both sides of the join.

4. A Spark 3.2 job has spark.sql.adaptive.enabled=true. The planner chose Sort-Merge Join at plan time because the right side appeared to be 2 GB. At runtime, AQE switched it to Broadcast Hash Join. What happened, and is this behavior desirable?

   Answer

   Correct Answer: AQE collected the actual shuffle output statistics after the map stage completed and found the right side's actual data volume after filtering and projection was much smaller than the 2 GB estimate — potentially under the effective broadcast threshold. AQE re-optimized the plan at runtime and switched to BHJ, eliminating the second shuffle. This is entirely desirable behavior: it represents AQE correcting a static planning error using real data. The result is faster execution with no manual intervention.

5. You have two 500 GB tables joined daily on customer_id. The join always uses Sort-Merge Join with two full shuffles. What architectural change eliminates the shuffle entirely, and what are its requirements?

   answer

   Correct Answer: Bucketing both tables on customer_id with the same number of buckets eliminates the Exchange (shuffle) operators from the SMJ plan. Requirements: both tables must be written to a Hive metastore-backed storage location using the same bucketing specification (same column, same bucket count, compatible sort order). When Spark reads both tables and plans the join, it detects that matching rows are already co-located across partitions and removes the Exchange operators, leaving only the Sort and MergeJoin operators. The one-time cost of rewriting both tables as bucketed tables is recovered over repeated daily runs.

6. Open-ended challenge: Your team runs 50 concurrent Spark jobs on a 300-executor cluster. Each job broadcasts a 100 MB dimension table. You observe executor OOM errors on a subset of executors every afternoon when concurrent job counts peak. Propose a cluster-level strategy to prevent these OOMs without disabling broadcast joins entirely.

---

🔗 Related Posts

• Shuffles in Spark: Why groupBy Kills Performance
• Spark Partitioning: Hash, Range, and Custom Strategies
• Spark DataFrames and Spark SQL: Schema, DDL, and the Catalyst Optimizer