Spark Joins and Data Shuffling

Data Engineering & Modern Data Platforms · Lesson 30 · Big Data Processing and Apache Spark

Lesson Overview

The Tuesday engagement job at TechLearningPro joined about sixty million progress events to a lesson catalog of forty thousand rows and to an enrollment table of about twenty million rows. The catalog join was correct. It was also a shuffle of both sides. The cluster moved the small table across the network so that each lesson_id could meet its events, then moved the events again to meet enrollments on (learner_key, course_id). One course, "Kubernetes for Platform Engineers," held roughly eighteen percent of the day's events. The task that received that key ran for about forty minutes. The median task in the same stage finished in about ninety seconds. The job exited zero, two hours after the 06:00 UTC bound the completion product still expects from Data Quality Engineering.

Nothing in the SQL was exotic. It was an equi-join. Spark's SQL engine had selected a sort-merge join because it did not trust the catalog to be small, and the enrollment side was genuinely large. Sort-merge is a sound default for two large inputs. It is an expensive default for a dimension that already fits in memory, and it concentrates a hot key onto one reducer. The shuffle is the part of the plan that decides which machine sees which key. Join strategy is the part that decides whether a shuffle happens at all.

This lesson treats those two decisions as one design. A shuffle is the exchange that places rows with the same join key on the same partition, by writing partitioned files in one stage and reading them in the next. A broadcast hash join avoids that exchange when one side is small enough to ship to every executor. A shuffle hash join and a sort-merge join both shuffle, and they spend the shuffle differently. Skew is the case where the key distribution makes one partition much larger than the median. Adaptive query execution, covered as a tuning system in Spark Performance Optimization, can split a skewed sort-merge partition at runtime. It does not remove the need to know which strategy you asked for, or what a broadcast hint will do to the driver.

The examples target Apache Spark 3.5 and the PySpark DataFrame API. Confirm numeric defaults with spark.conf.get on the runtime you use. Spark 4 keeps join, broadcast, and the hint names used here, and it can change a default. This lesson does not call a vendor-only optimizer. Where a screen in Databricks or EMR displays the same plan, the operators underneath are still Spark's BroadcastHashJoin, ShuffledHashJoin, and SortMergeJoin.

Learning Objectives

By the end of this lesson you will be able to:

1. Explain a shuffle as a stage boundary: map-side partitioned writes, a reduce-side read, and a new set of tasks.
2. Name the equi-join strategies Spark's physical planner actually chooses: broadcast hash, shuffled hash, and sort-merge.
3. State which side of an inner, left, right, and full outer join may be broadcast, and why a full outer join is not a broadcast hash join.
4. Force a broadcast with broadcast() or a BROADCAST hint, and describe the driver-side collect that can exhaust driver memory.
5. Read shuffle skew from task-duration spread and from per-task shuffle read, then decide among adaptive skew split, key salting, and a pre-aggregate.
6. Salt only the hot keys, with a deterministic pmod(hash(event_id), n), and drop the salt before the curated grain.
7. Keep unknown lesson_id values visible with a left join so a later quality count can see them.
8. Compare bucketing with broadcast for a repeated large-large join on a stable key.
9. Recognize a Cartesian or broadcast nested-loop plan as the cost of a non-equi condition.
10. Hand the next lesson a plan whose expensive exchange is intentional.

Prerequisites

• Apache Spark Architecture: jobs, stages, tasks, and the driver/executor split. A shuffle is a stage boundary inside that model.
• Spark DataFrames and Spark SQL: join, column expressions, and explain.
• PySpark Data Engineering: a SparkSession, Parquet reads, and a job you can submit.
• Spark RDDs and Execution Model: narrow versus wide dependencies. Every shuffle in this lesson is a wide dependency, including when you only see it inside a DataFrame plan.
• Data Formats and Serialization: events land as Parquet; the join does not change that file format, it changes how much of it moves.
• Python 3.11 and PySpark 3.5.x. The walkthrough runs in local[*] with a tiny shuffle-partition count so a laptop can execute it. Production numbers are labeled separately.

Detailed Core Concepts

What a join asks the cluster to do

An equi-join keeps rows whose key columns compare equal. For events.lesson_id = lessons.lesson_id, every event and every lesson that share an id must be on the same executor at the same time, or one of those two sets must already be present on every executor. Those are the only two distributed implementations that preserve the definition. Shipping one whole side everywhere is a broadcast. Repartitioning both sides by the key is a shuffle. A nested loop that compares every row to every row is the remaining option, and it is how non-equi predicates are executed when Spark cannot extract an equality.

The SQL text does not name the implementation. Catalyst chooses a physical operator from statistics, from thresholds, from join type, and from hints. Two plans can return the same rows and differ by an hour of runtime and by whether a hot key falls on one task. Correctness of the row set is necessary. It is not the design.

The shuffle as an exchange

Spark's default shuffle manager is the sort shuffle (SortShuffleManager). A task on the map side of the exchange computes a partition id for each row, buffers the rows, sorts them by partition, and writes a shuffle data file plus an index. The reduce side opens the index, fetches only its partition from each map output, and merges those streams. The fetch is network and disk. The next stage cannot start until the map outputs it needs are available.

The partition id for a hash exchange is derived from the key. Spark's hash partitioner uses a non-negative hash modulo the number of partitions. Rows with the same key land in the same partition. Rows with different keys share a partition whenever their hashes collide, which is normal: sixty million events and two hundred partitions means hundreds of thousands of rows per task even with a perfect spread. The failure mode this lesson cares about is the opposite of a hash collision. It is one key that is itself a large fraction of the input, so one partition is large no matter how many empty buckets you add.

spark.sql.shuffle.partitions (default 200) is the reduce-partition count when adaptive query execution does not coalesce. Two hundred tasks on a laptop fixture is scheduler overhead. Two hundred tasks on a multi-terabyte join can make each partition large enough to spill. The next lesson sizes that default with adaptive coalescing. Treat 200 as a number you look at.

Shuffle files sit on the local disk of the executors that wrote them, unless spark.shuffle.service.enabled is true. Dynamic allocation needs that service, or a similar tracker, because a finished map executor may otherwise be removed while its files are still required. Lineage can recompute the map stage from source files that still exist. That recompute is the batch recovery story in Distributed Processing Reliability. It does not help if the source was deleted.

Shuffle compression is on by default (spark.shuffle.compress true, spark.io.compression.codec lz4). A DataFrame SQL shuffle uses Spark's unsafe row format. Setting spark.serializer to Kryo changes RDD object shuffles. It does not change this join's shuffle bytes.

Broadcast hash join

A broadcast hash join builds a hash table from the smaller side and sends that table to every executor. The large side is then scanned locally. There is a BroadcastExchange and no shuffle of the large side. Spark will consider this when the estimated size of one side is at or below spark.sql.autoBroadcastJoinThreshold, whose default is 10 MB. Setting the threshold to -1 disables automatic broadcast. The estimate for a file-backed DataFrame is driven by the files Spark has statistics for. A compressed Parquet file of 8 MB can expand well past 10 MB in memory. A view or a DataFrame that has already been filtered may have no usable size until adaptive execution measures a shuffle. The static planner is conservative when it is unsure, and it falls through to a shuffle join.

Join type restricts which side may be broadcast:

A BROADCAST hint, or pyspark.sql.functions.broadcast, asks for a broadcast even when the estimate exceeds the threshold. Spark collects the broadcast relation on the driver in the classic execution path, then distributes it. A hint on a side that is "small" in a wiki and large on disk exhausts spark.driver.memory or spark.driver.maxResultSize. The executors can look idle while the driver dies. That is the failure mode of a generous hint. Adaptive query execution can still convert a sort-merge into a broadcast later, using the measured size of a shuffle, under spark.sql.adaptive.autoBroadcastJoinThreshold (defaulting to the same 10 MB). That conversion is a runtime decision. The hint is a planner instruction. Use the hint when you know the side is a dimension you have measured, and keep the threshold as a backstop for joins you did not hint.

spark.sql.broadcastTimeout (default 300 seconds) bounds how long executors wait for the broadcast variable. A slow driver collect looks like a timeout, not like a slow task.

Shuffled hash join and sort-merge join

Both strategies shuffle the inputs on the join keys so matching keys meet. A shuffled hash join builds a hash table per reduce partition from one side and probes it with the other. It prefers the case where each partition of the build side fits comfortably in memory. A sort-merge join sorts both sides on the key and merges them in order. The sort can spill to disk. The merge does not require one side's partition to fit in a hash table.

spark.sql.join.preferSortMergeJoin defaults to true, so Spark picks sort-merge for a large equi-join unless a hint or a size condition selects shuffled hash. The hint SHUFFLE_HASH requests the hash strategy. The hint MERGE (also written SHUFFLE_MERGE) requests sort-merge. Prefer sort-merge when partitions are large and skew handling matters: adaptive skew optimization, described below, applies to sort-merge. Prefer shuffled hash when each build partition is small, the key is selective, and you have measured that the hash table does not spill. spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold participates in that choice under adaptive execution. Leave it alone until the explain plan shows a join you disagree with; then change one setting and compare stage time.

Sort-merge requires the keys to be sortable, which ordinary string and numeric ids are. A join condition that is not an equality cannot use either strategy. Spark then uses a broadcast nested-loop join when one side can be broadcast, or a shuffled replicated nested-loop join (SHUFFLE_REPLICATE_NL) that replicates one side across partitions. A Cartesian product is the plan for a join with no condition, including a join predicate that the optimizer could not push into the join. A Python UDF in the join condition is a common way to hide an equality from the planner. The plan becomes a nested loop, and the cluster does the work the equality would have localized.

Skew

Skew means the partition sizes differ enough to dominate the stage. Hashing a uniform key into more partitions does not fix a key that is eighteen percent of the rows. That key is one partition, and the task that owns it runs until it finishes, spills, or fails. The stage's elapsed time is the slow task, not the median.

Three responses are legitimate, and they apply to different shapes.

Adaptive skew split (spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled, both default true on Spark 3.2 and later) splits a skewed partition of a sort-merge join and, when needed, replicates the matching partition of the other side. A partition is treated as skewed when its size is larger than spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5) times the median partition size and also larger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256 MB). Both tests must pass. A partition that is five times the median and only 20 MB is left alone. A partition that is 300 MB while the median is 250 MB is left alone because it is not five times the median. The split happens after the shuffle write, using measured sizes. It does not require you to name the hot key. It also does not rewrite an aggregation. A groupBy("course_id") whose hot course produces one huge aggregate partition is a different operator. Adaptive rebalance can split skewed partitions created by a repartition (spark.sql.adaptive.optimizeSkewsInRebalancePartitions.enabled, default true). A plain aggregation still needs a salt or a pre-aggregate if one key is the problem.

Salting is the manual form of that split. Use it when adaptive execution is off, when the skew is in an aggregation, or when replication must be limited to keys you measured. Add a salt on the large side, replicate only the hot keys of the smaller side, join on the key plus the salt, and drop the salt before the grain. The walkthrough uses pmod(hash(event_id), n). hash is deterministic for a Spark version. rand is not: a retry assigns new salts, and any sink or aggregate that kept the salt changes across retries. Salt only the hot keys. Replicating the entire enrollment table sixteen times costs more than the hot key saved. Refresh the hot-key list when a course launches.

Pre-aggregation removes skew that is really a grain bug. If the join only needs to know that a learner is enrolled, the enrollment side should already be unique on (learner_key, course_id). A row-per-day history multiplies events and makes the hot course hotter. Aggregate to the promised grain, then join.

Bucketing

DataFrameWriter.bucketBy writes a table already hash-partitioned on named columns. When both sides of a join use the same bucket count on the join key, and the planner can see it, that shuffle can be skipped. Bucketing pays off for a daily join you own and rewrite. It does not help a raw landing path partitioned only by ingest_date. Broadcast the lesson catalog. Consider bucketing enrollments and a curated event table only if the explain plan still shows a large shuffle after salting. Mismatched bucket counts fall back to a shuffle. The plan is the evidence.

What the join is allowed to drop

The lesson catalog carries a manifest id, the pin Data Quality Engineering requires. An inner join drops events whose lesson_id is absent from the pin, the engagement rate moves, and the stage succeeds. A left join keeps the event, leaves the dimension columns null, and lets a count of null manifest_id values become the quarantine rate. For that left join the broadcast side is the right side. This lesson chooses the left join. The quality lesson chooses the cap. Neither choice deletes raw events.

Architecture and Design Diagrams

The morning engagement build has two exchanges available to it, and a good plan uses one of them.

mermaidCopy
flowchart TD
    subgraph inputs [Inputs for one event_date]
        ev[Progress events partitioned by ingest_date]
        dim[Lesson dimension for one manifest_id]
        enr[Enrollments at grain learner_key and course_id]
    end

    subgraph plan [Physical plan]
        bcast[BroadcastExchange of the dimension]
        bhj[BroadcastHashJoin on lesson_id left outer]
        salt[Salt only hot course_id values]
        exch[ShuffleExchange on learner_key course_id salt]
        smj[SortMergeJoin]
        drop[Drop salt and project the fact grain]
    end

    dim --> bcast --> bhj
    ev --> bhj
    bhj --> salt --> exch --> smj
    enr --> salt
    smj --> drop
    drop --> fact[(fact_lesson_engagement for event_date)]

The dimension path has no shuffle of events. The enrollment path shuffles because both sides are large. The salt is an implementation column on that second join only.

A single shuffle stage, without the dimension drawn, is this exchange:

mermaidCopy
sequenceDiagram
    participant Map as Map tasks
    participant Disk as Local shuffle files
    participant Red as Reduce tasks
    Map->>Map: Hash the join key to a partition id
    Map->>Disk: Write sorted blocks and an index
    Red->>Disk: Fetch this partition from each map output
    Red->>Red: Sort-merge or build a hash table

If an executor is removed before the fetch, and no external shuffle service retains the files, the map stage is recomputed from lineage. The reduce tasks do not invent the rows.

Technical Explanation

How the planner picks a strategy

For an equi-join, Spark's join selection considers a broadcast hash join first when statistics or a hint say a side is broadcastable and the join type allows that side. If neither side is broadcast, it considers a shuffled hash join when sort-merge is not preferred or a SHUFFLE_HASH hint is present, and when the build side looks small enough per partition. Otherwise it uses sort-merge. Hints are instructions to the planner, not a separate engine. A hint that contradicts the join type, such as broadcasting both sides of a full outer join, does not create a new operator. The explain plan is the record of what was chosen.

Parquet planning statistics are file sizes and, when present, footer row-group stats. They are unknown for many non-file sources. After a stage finishes, adaptive execution knows shuffle partition bytes. A join that planned as sort-merge can then run as broadcast hash when a side is under spark.sql.adaptive.autoBroadcastJoinThreshold. spark.sql.adaptive.localShuffleReader.enabled (default true) lets the probe side read locally instead of shuffling a second time. You may still have paid for the first shuffle. Choose a defensible static plan anyway: some clusters disable adaptive execution to match an old runbook, and the next lesson is where that flag is turned back on with measurements.

Partition counts and empty tasks

A hash partitioner with 200 partitions and a low-cardinality key leaves some partitions nearly empty. Those tasks still start. Scheduler cost shows up on small jobs and disappears next to a 40-minute skewed task. Adding partitions does not split the hot key. Reducing partitions without splitting it makes the hot task larger. Change partition count and skew handling one at a time.

repartition(n, "course_id") is a hash shuffle on that column. repartition(n) is a round-robin shuffle, which spreads rows evenly and destroys key co-location. A later join on course_id shuffles again. coalesce(n) narrows the partition count without a full shuffle. It can only decrease the count, and it can leave the hot key's rows stacked in one of the surviving partitions. Coalesce is a file-count tool. It is a poor skew tool.

Exchange reuse

If two branches of a plan shuffle the same DataFrame on the same partitioning, Spark can reuse the exchange when spark.sql.exchange.reuse is true (the default). The physical plan shows a reused exchange. Caching is a different mechanism: it materializes rows. Reuse avoids writing the same shuffle twice inside one query. It does not survive across actions unless you cache. A job that joins events to lessons and then, in a second action, joins the same events to enrollments will shuffle events twice unless the intermediate DataFrame is cached. Cache a DataFrame that is used by more than one action. Do not cache the broadcast dimension "just in case" if the broadcast exchange already ran inside a single action. The next lesson is the memory budget for that choice.

Null keys

A join key that is null does not equal null. Those rows fall out of an inner join and stay on the outer side of an outer join, with the other side null. A salt expression on a null course_id should stay in an explicit bucket you can count, not disappear into pmod of a null hash. hash of a null is a defined Spark value, so pmod(hash(course_id), n) will place nulls together. That is another hot key if missing course ids are common. Fix the missing ids in the quality gate, or isolate them, before you salt them into the enrollment join. Joining null course ids to enrollments has no business meaning in this product. Filter them to quarantine first.

Output cardinality

Broadcast versus sort-merge does not change the multiset of output rows for the same predicate. Salting does not change it either, if every salted fact row meets exactly one replica of its dimension row and you do not project the salt into a later count. A duplicated dimension row changes it, regardless of strategy. The lesson manifest must be unique on lesson_id inside one manifest_id. Enforce that before the broadcast. A broadcast of a duplicated dimension multiplies events on every executor, quietly, and the plan still says BroadcastHashJoin.

Practical Implementation or Design Walkthrough

Problem

Build the engagement join for one event_date. Keep events whose lesson is missing from the pinned manifest. Attach enrollment only when (learner_key, course_id) matches the current enrollment grain. Salt only a measured set of hot courses. Write nothing in this lesson; the physical plan and the row counts are the deliverable. The capstone writes the fact table.

Schemas

Progress events, one row per learning event already landed as Parquet:

Lesson dimension, one row per lesson inside a manifest:

Enrollments at the grain the join needs:

Sample event rows:

textCopy
evt-1, L-9, lesson-a, course-spark, 2026-03-02T01:00:00Z
evt-2, L-9, lesson-missing, course-spark, 2026-03-02T01:05:00Z
evt-3, L-2, lesson-k, course-k8s, 2026-03-02T01:06:00Z

evt-2 must survive the lesson join with null dimension columns. evt-3 stands in for the hot course.

Code

The program uses only pyspark.sql methods documented for Spark 3.5. Local mode sets a small shuffle-partition count. A production submit omits master("local[*])" and lets the cluster manager set executors.

pythonCopy
"""Engagement join: broadcast the lesson manifest, salt hot courses, sort-merge enrollments.

PySpark 3.5.x. Local defaults are for a laptop fixture, not for the sixty-million-row run.
"""

from pyspark.sql import SparkSession, DataFrame
from pyspark.sql import functions as F

# Measured from the previous day's per-key counts. Review on launch day.
HOT_COURSES = ("course-k8s",)
SALT_BUCKETS = 16
MANIFEST_ID = "dim-2026-03-03"
EVENT_DATE = "2026-03-02"


def build_spark() -> SparkSession:
    return (
        SparkSession.builder.appName("tlp-engagement-join")
        .master("local[*]")
        # Laptop only. Production starts from the cluster default and uses AQE.
        .config("spark.sql.shuffle.partitions", "8")
        .config("spark.sql.adaptive.enabled", "true")
        .config("spark.sql.adaptive.skewJoin.enabled", "true")
        .config("spark.sql.autoBroadcastJoinThreshold", str(10 * 1024 * 1024))
        .getOrCreate()
    )


def salt_events(events: DataFrame) -> DataFrame:
    """Deterministic salt. Retries of a task keep the same salt for an event_id."""
    hot = F.col("course_id").isin(*HOT_COURSES)
    salt = F.when(
        hot, F.pmod(F.hash(F.col("event_id")), F.lit(SALT_BUCKETS))
    ).otherwise(F.lit(0))
    return events.withColumn("_salt", salt)


def replicate_hot_enrollments(enrollments: DataFrame) -> DataFrame:
    """Copy only hot-course enrollment rows across salt buckets. Cold keys stay at salt 0."""
    hot = enrollments.where(F.col("course_id").isin(*HOT_COURSES))
    cold = enrollments.where(~F.col("course_id").isin(*HOT_COURSES)).withColumn(
        "_salt", F.lit(0)
    )
    buckets = hot.withColumn(
        "_salt", F.explode(F.sequence(F.lit(0), F.lit(SALT_BUCKETS - 1)))
    )
    return buckets.unionByName(cold)


def engagement_join(
    events: DataFrame, lessons: DataFrame, enrollments: DataFrame
) -> DataFrame:
    # Right side is the only legal broadcast for a left outer join.
    with_lesson = events.join(F.broadcast(lessons), on="lesson_id", how="left")

    unknown = with_lesson.where(F.col("manifest_id").isNull()).count()
    print(f"events_missing_lesson={unknown}")

    salted_events = salt_events(with_lesson)
    salted_enrollments = replicate_hot_enrollments(enrollments)

    joined = salted_events.join(
        salted_enrollments,
        on=["learner_key", "course_id", "_salt"],
        how="left",
    )
    # Salt is an implementation column. The fact grain does not include it.
    return joined.drop("_salt")


def main() -> None:
    spark = build_spark()
    events = spark.read.parquet("/tmp/tlp/events").where(F.col("event_date") == EVENT_DATE)
    lessons = (
        spark.read.parquet("/tmp/tlp/lessons")
        .where(F.col("manifest_id") == MANIFEST_ID)
        .select("manifest_id", "lesson_id", "course_id", "lesson_title")
    )
    # Grain guard: one row per learner and course. A duplicate would multiply events.
    enrollments = (
        spark.read.parquet("/tmp/tlp/enrollments")
        .select("learner_key", "course_id", "enrolled_on")
        .dropDuplicates(["learner_key", "course_id"])
    )

    result = engagement_join(events, lessons, enrollments)
    result.explain(mode="formatted")
    # Action that materializes the plan. The capstone replaces this with a write.
    print(f"joined_rows={result.count()}")
    spark.stop()


if __name__ == "__main__":
    main()

The same broadcast can be written in SQL as a hint for a reviewer who lives in the SQL editor:

sqlCopy
SELECT /*+ BROADCAST(lessons) */ e.*, lessons.manifest_id, lessons.lesson_title
FROM events e
LEFT JOIN lessons
  ON e.lesson_id = lessons.lesson_id

MERGE and SHUFFLE_HASH are the other equi-join hints. SHUFFLE_REPLICATE_NL is the non-equi hint. Put a hint in the code review only when the explain without it is the wrong operator.

What you expect to see

On the fixture, events_missing_lesson is at least one if you included lesson-missing. joined_rows equals the event count when the enrollment join is a left join and enrollment duplicates have been removed: every event remains, whether or not an enrollment matched. Switch the enrollment join to inner only when the product definition says unenrolled events are out of the fact. That is a grain decision. Record it next to the plan.

A formatted explain for the dimension join should contain BroadcastHashJoin and BroadcastExchange. If it contains SortMergeJoin on lesson_id, the broadcast did not happen: the hint was dropped, the side was not the right side, or statistics plus a disabled adaptive conversion won. Do not argue from the SQL text.

For the salted enrollment join, the join keys in the plan include _salt. After drop("_salt"), the project list does not. If a downstream groupBy still lists _salt, the salt leaked into the metric.

Configuration to take to production

Unknown lessons are counted in the driver with count(), which is a separate action and will compute the broadcast join once for the count and again for the final action unless you cache with_lesson. On a sixty-million-row input, cache that intermediate if both actions stay in the job, or combine the unknown-lesson count into the same write path as an aggregation. The walkthrough keeps them separate so the count is visible in a tutorial. The capstone should not pay for the join twice.

Enterprise Real-World Scenarios

Scenario 1: The hot course on a shuffle join

A platform course is assigned to a large fraction of a new cohort in one week. Progress events for that course_id dominate the enrollment join. With two hundred shuffle partitions, one task's shuffle read is two orders of magnitude above the median. Executors elsewhere sit idle. The job misses the morning bound, and the on-call engineer raises spark.sql.shuffle.partitions to two thousand. The hot task remains one task, because the key still hashes to one partition, and the stage now schedules two thousand tasks.

Broadcast the lesson dimension, which was never the hot join. On the enrollment sort-merge, use adaptive skew split or a salt limited to course ids whose previous-day share exceeded a threshold you write down (here, five percent). Deduplicate enrollments to (learner_key, course_id) before replicating hot keys. Keep salt out of the fact grain (learner_key, lesson_id, event_date). After the hot task is gone, set the partition count from bytes per task. Verify with task-duration percentiles.

Scenario 2: A "small" lookup that fills the driver

A second pipeline joins events to a table of per-learner notification preferences that an application database exports nightly. The table was 4 MB for a year. A product change stores one row per learner per channel per course, and the export becomes 6 GB. A BROADCAST hint from the original review is still in the SQL. The driver collects 6 GB, hits spark.driver.maxResultSize, and the job fails before a single reduce task starts. The Spark UI shows a broadcast exchange and almost no shuffle.

Removing the hint lets the planner choose sort-merge. That finishes, and it is the wrong permanent answer if most runs are still small. Hint a broadcast only for a relation whose upper bound you enforce, such as one lesson manifest after a row-count check. Let the threshold and adaptive execution handle tables that grow. The preference export gets a sort-merge, a filter that drops historical channels, and no hint. A unit test that explains a preferences frame larger than the threshold, with no hint, should not contain BroadcastHashJoin.

Scenario 3: Revenue lines, briefly

The same two strategies show up on the purchase-line fact from the ingestion module, joined to a course dimension of a few thousand rows and to a refund ledger that is large. Broadcast the course dimension. Sort-merge the ledger on the order key. Do not broadcast the ledger because "finance tables are smaller than clickstreams." Measure the side. A full outer join used to reconcile "rows only in the ledger" cannot be a broadcast hash join; if someone hints it, believe the explain plan when it refuses. Payment tokens never enter the join key or the project list. The join is not the control that blocks them. It is a place they can spread if you select *.

Best Practices

• Measure side sizes before you hint. A dimension with a manifest pin and a known row count is a broadcast. A fact is a shuffle.
• Use a left join when a missing dimension row must remain visible to a quality count. Use an inner join only when the product grain says the row is out.
• Deduplicate the build side to the join grain before a broadcast or a salt replication. Duplicates multiply silently.
• Salt with pmod(hash(stable_id), n) on keys you have listed. Drop the salt in the same method that added it.
• Leave adaptive skew join on. Add manual salt when the hot key is known, when the operator is an aggregation, or when you must cap replication to those keys.
• Read explain(mode="formatted") after every strategy change. Review BroadcastHashJoin, SortMergeJoin, and the join keys, including accidental salt.
• Keep one shuffle partition setting per environment, and change skew handling separately so a before-and-after stage time means one thing.
• Project the columns the fact needs. select * through a join pulls attributes into the shuffle.
• Put the hot-key list and the manifest id in the job arguments, not in a notebook cell that drifted from production.
• When dynamic allocation is enabled, run an external shuffle service so a completed map task's files survive executor removal.

Common Mistakes and Anti-Patterns

Raising the shuffle partition count to cure a hot key. The key still maps to one partition. You pay more tasks and keep the long one. Split or salt the key, then size partitions by bytes.

A permanent broadcast hint. The hinted side grows, the driver collects it, and the failure looks like an infrastructure timeout. Hint relations with an enforced bound. Let the threshold decide the rest.

Inner join as a filter for unknown lessons. The metric changes and the stage is green. Left-join, count the nulls, and let the quality gate apply the cap.

Salting every key, or salting with rand. Full replication multiplies the smaller large table. rand makes a retry differ whenever salt is observable. Salt measured keys with a hash of a stable id, and drop the column.

Joining before the grain is unique. Event rows times enrollment history rows explode the hot course. dropDuplicates on the enrollment grain, or an aggregate, belongs before the exchange.

A Python UDF as the join condition. The planner loses the equality and schedules a nested loop. Write on=["lesson_id"] or an equality expression Spark can see.

Caching nothing and counting twice, or caching everything. Two actions recompute a shuffle. A cache of a DataFrame used once spends memory the next lesson will need for the sort. Cache the multi-action intermediate only.

Bucket counts that differ by environment. A table bucketed into 64 in production and read beside a 32-bucket partner still shuffles. The explain line is the test. A matching bucketBy in a script nobody runs is not.

Selecting every column from a raw event. Tokens and emails that landed despite the contract ride the shuffle and the broadcast. Project the named schema before the join.

Troubleshooting and Design Review

Symptom: One task in the enrollment join runs far longer than the median. Its shuffle read is a large fraction of the stage.

Likely cause: a hot course_id or a hot pair (learner_key, course_id) if a test account is misconfigured. Extra partitions did not split it.

Investigation: from the stage detail, sort tasks by shuffle read. Sample keys in the slow partition with a job that aggregates counts by course_id for that date. Check whether _salt is in the join keys. Check spark.sql.adaptive.skewJoin.enabled and whether the partition exceeded both the factor and the 256 MB threshold. A skewed partition under 256 MB is not split.

Corrective action: salt the measured hot keys or lower the byte threshold if the partitions are smaller than 256 MB and still many times the median. Deduplicate enrollments first. Do not raise the partition count as the only change. Recompute task-duration percentiles after one change.

Symptom: The driver fails during BroadcastExchange with an out-of-memory error or a max-result-size error. Executors show little work.

Likely cause: a broadcast hint or a threshold above the real size of the "small" side. The collect is on the driver.

Investigation: search the plan for BroadcastExchange and the hinted relation. Compare the input size of that relation with spark.sql.autoBroadcastJoinThreshold and driver memory.

Corrective action: remove the hint, filter the relation to one manifest, or set the threshold back to 10 MB. Broadcast only after a count or a file-size guard that fails the job when the side exceeds the bound you designed for.

Symptom: Unknown lessons vanished and the completion rate dipped. The join stage succeeded.

Likely cause: an inner join to the dimension.

Investigation: compare event counts before and after the join, grouped by course_id. Look for how="inner" or a SQL JOIN without LEFT.

Corrective action: left join, count null manifest_id, and pass that rate to the quality gate. Keep the raw events. Do not restore the dropped rows by editing the fact table in place.

Symptom: Output row count is a multiple of the event count.

Likely cause: the dimension or the enrollment side is duplicated on the join key, or the salt was joined without a matching replication and a later union duplicated rows. A salt bug usually drops matches or multiplies them if the hot side was exploded on both inputs.

Investigation: groupBy the dimension on lesson_id inside the manifest and show counts above one. Do the same for enrollments on (learner_key, course_id). Print the join keys from the plan.

Corrective action: enforce uniqueness before the exchange. Rebuild the salt so only one side is replicated.

Symptom: A non-equi "cleanup" predicate made the job many times slower, with a Cartesian or nested-loop operator in the plan.

Likely cause: the predicate is a UDF, or it uses between on a timestamp without an equality key.

Investigation: read the physical plan for BroadcastNestedLoopJoin, CartesianProduct, or ShuffledNestedLoopJoin.

Corrective action: express the equality as the join keys and the range as a post-join filter, or accept a broadcast nested loop only when one side is truly small. Do not hint SHUFFLE_REPLICATE_NL to silence the warning.

A design review asks: Which side is broadcast, and what bound makes that safe for the driver? Which join is a shuffle, and what is the key? What is the hot key, and is it salted, split by AQE, or accepted? Does the salt appear in the output? What happens to an unknown lesson_id? Is the other side unique on the join grain? Which explain mode was captured with the change? A review that says "we broadcast joins" has not answered these.

Interview Questions and Answers

Q1. What is a shuffle, and why does an equi-join of two large tables need one?

A shuffle is the exchange that repartitions rows so each join key lands in a known partition. Map tasks write partitioned, sorted shuffle files and an index. Reduce tasks fetch only their partition and then hash-join or sort-merge inside that partition. Two large tables cannot be joined by shipping either side to every executor. The rows that share a key have to meet, and the shuffle is how Spark places them together. The stage boundary is why the reduce stage waits, and why losing shuffle files means recomputing the map stage from lineage if the source still exists.

Q2. When will Spark pick a broadcast hash join, and which side may be broadcast on a left outer join?

Spark considers a broadcast when a hint says so, or when the estimated size of a side is within spark.sql.autoBroadcastJoinThreshold (default 10 MB), and the join type allows that side. For a left outer join the right side is the one that may be broadcast, because every left row must be preserved and the right side is the lookup table. Inner joins may broadcast either side. A full outer join is not executed as a broadcast hash join, because unmatched rows on both sides have to be produced. I confirm with explain, not with the hint text.

Q3. How can a broadcast hint exhaust the driver while executors still have free memory?

In the classic execution path the broadcast relation is collected to the driver and then distributed. The hint bypasses the size threshold. A side that used to be a few megabytes and is now gigabytes is collected onto the driver, which fails on heap or on spark.driver.maxResultSize. Executors never receive a complete hash table, so the UI shows a broadcast exchange and little task time. I hint only relations with an enforced bound, such as one lesson manifest, and I leave the 10 MB threshold in place for everything else. Adaptive execution can still broadcast a side that a shuffle proves is small, without a hint.

Q4. How do you choose between sort-merge and a shuffled hash join?

Both shuffle on the equi-join keys. Sort-merge sorts each side and merges. It spills instead of requiring the build partition to fit in a hash table, and it is the strategy adaptive skew splitting knows how to subdivide. Shuffled hash builds a hash table per partition and is attractive when those partitions are small. With spark.sql.join.preferSortMergeJoin left at its default true, I expect sort-merge for a large enrollment join. I request SHUFFLE_HASH only after an explain and a stage-time comparison. I do not flip the session default inside one job without recording why.

Q5. A single course id is eighteen percent of events. What does raising spark.sql.shuffle.partitions do, and what would you do instead?

It leaves that key in one partition and adds tasks around it. The stage still waits on the hot task. I deduplicate the other side to the real grain, then either let adaptive skew join split the partition when it is larger than five times the median and larger than the 256 MB threshold, or salt only that course id. The salt is pmod(hash(event_id), n) on the event side and an explode of the same buckets on the matching enrollment rows. I drop the salt before the output. I change partition count afterwards, based on bytes per task, as a separate experiment.

Q6. Why is pmod(hash(event_id), n) a better salt than rand()?

hash is deterministic for a Spark version, so a retried task assigns the same salt to the same event. rand assigns a new one. If the salt is dropped and the other side was replicated across every bucket, the business rows of the join can still match under rand, but any leak of the salt into a key, a file partition, or an intermediate aggregate makes the retry observable. Deterministic salts keep replays comparable. I still drop the column so the curated grain stays the grain the product defined.

Q7. What does a left join to the lesson dimension preserve that an inner join destroys?

Events whose lesson_id is absent from the pinned manifest. They remain, with null dimension attributes, and a count of those nulls is the rate a quality gate can fail on. An inner join deletes them from the result and returns success, so the engagement rate moves with no quarantine record. Broadcast still applies: for the left join I broadcast the dimension, the right side. The raw events stay where they were. The join is not allowed to be the only place an unknown lesson disappears.

Q8. How do you tell join skew from an under-partitioned stage in the Spark UI?

Skew is a wide gap inside one stage: a small number of tasks dominate duration and shuffle read, while the median task is short. Under-partitioning makes most tasks large together. Over-partitioning makes most tasks tiny and the scheduler busy. I sort the stage's tasks by shuffle read. If the top task is huge and the rest are not, I look at the key. If every task is huge, I look at partition count and input size. Spill on the hot task tells me the partition did not fit in execution memory. Spill on every task tells me the whole stage is short of memory or over-aggregated, which is the next lesson's problem.

Q9. When would you bucket two tables instead of broadcasting one?

When both sides are large, the join key is stable, the job runs often enough to pay for the write, and both tables are written with the same bucket count on that key through bucketBy. A later join can then skip the shuffle if the planner honors the bucketing. I would not bucket the lesson catalog. It is a broadcast. I might bucket enrollments and a curated event table if the morning join on (learner_key, course_id) is the dominant cost and adaptive execution still shows a large shuffle after salting. I treat a missing shuffle in the explain plan as the proof. Matching bucket counts in two scripts that have drifted is how the shuffle comes back.

Q10. A join predicate calls a Python function to decide whether two ids "match." What plan do you expect, and what do you change?

I expect a nested-loop or Cartesian-style operator, because the planner cannot extract an equality from the UDF. Every partition of one side is compared broadly with the other. I move the real equality into the join keys, keep range checks as filters after the join, and leave the UDF out of the condition. If the rule truly is not an equality, I broadcast the smaller side only when it fits the threshold and I accept a broadcast nested loop with my eyes open. I do not add SHUFFLE_REPLICATE_NL to hide the cost in a hint nobody measures.

Q11. What is the difference between exchange reuse and caching across two joins of the same events?

Inside one query, spark.sql.exchange.reuse can share a shuffle that two branches computed the same way, and the plan shows a reused exchange. A second action is a second job. The shuffle runs again unless the DataFrame was cached or persisted. I cache the post-dimension DataFrame if the unknown-lesson count and the enrollment join are separate actions. I do not cache a DataFrame that a single action consumes once. Cache memory is execution memory's neighbor, and the sort-merge in the same job may need it.

Hands-On Exercises

Exercise: Repair the engagement join

Problem statement. You are given three Parquet inputs for one event_date: events, a lesson manifest, and enrollments. The reference job inner-joins all three and shuffles the manifest. One course_id in the fixture is hot.

Tasks.

1. Write the row counts of each input and the uniqueness of lesson_id inside the manifest and of (learner_key, course_id) on enrollments.
2. Produce a left join of events to the manifest that broadcasts the manifest. Capture explain(mode="formatted") and mark the BroadcastHashJoin.
3. Count events that did not match the manifest. State what an inner join would have done to that count.
4. Deduplicate enrollments, salt only the hot course with pmod(hash(event_id), n), and join on the salted keys. Show that _salt is absent from the final schema.
5. Run once with adaptive skew join enabled and once with a salt, on a fixture large enough that the hot key is visible in task metrics or in a groupBy count. Describe one measurement, not both as a feeling.
6. Add a case that broadcasts a side larger than the driver's safe bound is not required. Instead, explain in writing what the driver collect would do if you hinted that side.
7. Document the production partition stance separately from the local spark.sql.shuffle.partitions you used.

Expected deliverables. The script, the two explain plans, the unknown-lesson count, the final schema, and a short note on the hot key.

Evaluation rubric.

You pass when a reviewer can point to the broadcast in the plan, the salt in the join keys, the salt's absence in the output, and the count of unmatched lessons.

Lesson Summary

A distributed equi-join either broadcasts one side or shuffles both sides onto the join key. Broadcast hash join is the right plan for the pinned lesson manifest, on the right side of a left join, and only while that side stays within a bound the driver can collect. Sort-merge is the right default for events against enrollments. Shuffle partition count sizes tasks. It does not split a hot course_id. Adaptive skew join splits a sort-merge partition that clears both the factor and the byte threshold. A deterministic salt on measured keys does the same thing explicitly, and the salt stays out of the fact grain. Unknown lessons stay in the row set so a quality gate can see them.

The next lesson, Spark Performance Optimization, takes this plan and looks at the rest of the cost: adaptive coalescing, cache levels, spill, file sizes, and how to read the explain output without guessing.

Further Reading

• Apache Spark 3.5 SQL performance tuning — broadcast threshold, join hints, and the adaptive skew-join rules used in this lesson. Confirm the page for the patch version you run.
• Apache Spark 3.5 configuration — spark.sql.shuffle.partitions, shuffle compression, and spark.driver.maxResultSize.
• Spark SQL, DataFrames and Datasets guide — join, explain, and broadcast.
• PySpark SQL API — functions.broadcast, functions.hash, functions.pmod, and DataFrame.hint.
• Holden Karau and others' Spark literature on shuffle and skew, read beside the versioned docs. The docs win when a default has moved.
• Data Quality Engineering — the cap and the manifest pin this left join is feeding.