Data Engineering Module 9 – Batch processing with Apache Spark
Spark joins, shuffles and partitions
What a Spark shuffle moves, how the planner picks a broadcast, sort-merge or shuffled hash join, and where every partition count in a job comes from.
What you will learn
- Explain what a shuffle moves between tasks and why it dominates a job's cost
- Distinguish broadcast, sort-merge and shuffled hash joins by what each one moves
- Read a plan's Exchange and join operators to name the strategy Spark chose
- Work out how many partitions a shuffle declares and how many tasks really run
- Choose a number of shuffle partitions for a stage from the volume it moves
Before you start
What you’ll learn · Before you start
What you will learn
- Explain what a shuffle moves between tasks and why it dominates a job's cost
- Distinguish broadcast, sort-merge and shuffled hash joins by what each one moves
- Read a plan's Exchange and join operators to name the strategy Spark chose
- Work out how many partitions a shuffle declares and how many tasks really run
- Choose a number of shuffle partitions for a stage from the volume it moves
Before you start
On this page
When Spark joins 1,200,000 delivery events to a table of 8 city zones, it can move all 1,200,000 rows across the network, or none of them, and return the same rows either way. One comparison against a size threshold decides which, and the plan records the answer in an operator name. The job’s cost also turns on a number few people set on purpose: how many partitions each of its shuffles declares.
The numbers below come from four recorded PySpark 4.2.0 runs in local mode that report counts, not durations: rows, partitions, stages and plan operators. By the end you can read a plan and say what moves, why that strategy was chosen and how many tasks a stage runs.
What a shuffle moves, and why it sets the cost
The first recorded run asks three questions of the same events and prints what each plan holds. Keeping the events whose minutes are 40 or more returns 266,666 rows of 1,200,000 and plans no Exchange operator: The operator in a Spark physical plan that stands for one shuffle: rows leave the partitions they were computed in and are re-distributed, usually by a hash of the key.. Counting the events per zone returns 8 rows and plans one. The join to the zones table returns all 1,200,000 rows and plans none, with a BroadcastExchange beside it.
join_plan_session.py
"""Session driver: run join_plan.py on PySpark 4.2.0 in local mode, inside a private lab folder.
Spark needs a Java runtime and the pyspark package. The driver points at the pinned ones (Java 21 and PySpark
4.2.0), gives Spark a home and a temporary folder inside the lab, binds it to 127.0.0.1 and prints only what the
program prints: configuration values, row counts and operator names, never Spark's own log lines or paths.
"""
import os
import pwd
import subprocess
import sys
import tempfile
from pathlib import Path
# Where mise installs the pinned Java 21. The home folder comes from the account database, so the path holds even
# when HOME points at a scratch folder.
REAL_HOME = pwd.getpwuid(os.getuid()).pw_dir
JAVA_HOME = os.path.join(REAL_HOME, ".local", "share", "mise", "installs", "java", "temurin-21.0.12+101.0.LTS")
HERE = Path(__file__).resolve().parent
PROGRAM = "join_plan.py"
def pinned_pyspark():
"""The folder that holds the pinned pyspark 4.2.0 package, or None to use the one installed for this Python."""
for folder in [Path.cwd(), *Path.cwd().parents, *HERE.parents]:
for found in folder.glob("*/toolchains/pyspark-4.2.0/python"):
return str(found)
return None
with tempfile.TemporaryDirectory(prefix="jsp-plan-") as lab:
for name in ("home", "tmp", "out"):
os.makedirs(os.path.join(lab, name))
java_home = JAVA_HOME if os.path.isdir(JAVA_HOME) else os.environ.get("JAVA_HOME", "")
env = {
"PATH": os.pathsep.join([os.path.join(java_home, "bin"), "/usr/bin", "/bin"]),
"JAVA_HOME": java_home,
"HOME": os.path.join(lab, "home"),
"TMPDIR": os.path.join(lab, "tmp"),
"WORK": os.path.join(lab, "out"),
"SPARK_LOCAL_IP": "127.0.0.1",
"PYSPARK_PYTHON": sys.executable,
"TZ": "UTC",
"LANG": "en_US.UTF-8",
}
package = pinned_pyspark()
if package:
env["PYTHONPATH"] = package
done = subprocess.run([sys.executable, str(HERE / PROGRAM)], cwd=lab, env=env,
capture_output=True, text=True, check=False)
print(done.stdout.replace(lab, "<lab>").rstrip())
if done.returncode != 0:
last = [line for line in done.stderr.splitlines() if line.strip()][-1:] or ["no message"]
print(PROGRAM, "stopped with exit status", done.returncode, "-", last[0].replace(lab, "<lab>"))
sys.exit(done.returncode)Output
PySpark 4.2.0, three settings this session left alone: spark.sql.shuffle.partitions = 200 spark.sql.adaptive.enabled = true spark.sql.autoBroadcastJoinThreshold = 10485760b events rows = 1200000, partitions = 2 zones rows = 8, partitions = 2 filter and select: every row stays where it is rows out = 266666 Exchange = 0, BroadcastExchange = 0, Sort = 0 operators: Project | Filter | Range count per zone: equal zones must meet rows out = 8 Exchange = 1, BroadcastExchange = 0, Sort = 0 operators: HashAggregate | Exchange hashpartitioning ENSURE_REQUIREMENTS | HashAggregate | Project | Range join to the 8 zones: the planner decides how rows out = 1200000 Exchange = 0, BroadcastExchange = 1, Sort = 0 operators: Project | BroadcastHashJoin BuildRight Inner | Project | Filter | Range | BroadcastExchange | Project | Range
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 join_plan_session.py
join_plan.py
"""PySpark 4.2.0, local mode: which steps shuffle, and which join the planner chose on its own.
Three questions over the same 1,200,000 delivery events of a food-delivery company: a filter that keeps every row
where it already sits, a count per zone that must gather equal zones together, and a join to the 8-row zones table.
The program prints the three settings it did not set, so they are Spark's own defaults, then the operators of each
plan with its shuffle, broadcast and sort counts. Plans are printed through plan_ops.py, which keeps operator names
only: a raw plan carries estimated sizes and ids that may differ between runs.
"""
from pyspark.sql import SparkSession
from plan_ops import operators, shape
spark = (SparkSession.builder
# learn-lint-allow inclusive: PySpark 4.2.0 names the builder method that sets the cluster URL master()
.master("local[2]")
.appName("joins-shuffles-defaults")
.config("spark.ui.enabled", "false")
.config("spark.driver.bindAddress", "127.0.0.1")
.config("spark.driver.host", "127.0.0.1")
.config("spark.driver.memory", "1g")
.getOrCreate())
spark.sparkContext.setLogLevel("ERROR")
print(f"PySpark {spark.version}, three settings this session left alone:")
for key in ("spark.sql.shuffle.partitions", "spark.sql.adaptive.enabled", "spark.sql.autoBroadcastJoinThreshold"):
print(f" {key} = {spark.conf.get(key)}")
events = (spark.range(0, 1200000)
.selectExpr("id as event_id",
"cast(pmod(hash(id), 8) as int) as zone_id",
"cast(12 + pmod(hash(id, 7), 36) as int) as minutes"))
zones = (spark.range(0, 8)
.selectExpr("cast(id as int) as zone_id", "concat('zone-', cast(id as string)) as zone_name"))
print(f"\nevents rows = {events.count()}, partitions = {events.rdd.getNumPartitions()}")
print(f"zones rows = {zones.count()}, partitions = {zones.rdd.getNumPartitions()}")
def report(label, df, rows_out):
print(f"\n{label}")
print(f" rows out = {rows_out}")
print(f" {shape(df)}")
print(f" operators: {' | '.join(operators(df))}")
slow_deliveries = events.filter("minutes >= 40").select("event_id", "zone_id")
report("filter and select: every row stays where it is", slow_deliveries, slow_deliveries.count())
per_zone = events.groupBy("zone_id").count()
report("count per zone: equal zones must meet", per_zone, per_zone.count())
with_names = events.join(zones, "zone_id").select("event_id", "zone_name")
report("join to the 8 zones: the planner decides how", with_names, with_names.count())
spark.stop()plan_ops.py
"""Read a Spark physical plan as a list of operator names, with no sizes, ids or paths in it.
The plan string Spark prints carries estimated sizes, expression ids and plan ids, all of which may differ between
runs and between machines, so nothing here prints a plan as it stands. operators() keeps the operator names this
lesson talks about, each with the few keywords on its line that say how it works, and counts() counts them.
"""
import re
WATCHED = (
"AQEShuffleRead",
"BroadcastExchange",
"BroadcastHashJoin",
"Exchange",
"Filter",
"HashAggregate",
"Project",
"Range",
"ShuffledHashJoin",
"Sort",
"SortMergeJoin",
)
KEYWORDS = (
"BuildLeft",
"BuildRight",
"Inner",
"coalesced",
"hashpartitioning",
"local",
"SinglePartition",
"RoundRobinPartitioning",
"ENSURE_REQUIREMENTS",
"REPARTITION_BY_NUM",
)
def plan_text(df):
"""The physical plan Spark executed, or will execute, for this DataFrame.
An adaptive plan prints its final plan and then the initial plan it started from; only the first half describes
the operators that ran, so the second half is cut off here and never counted twice.
"""
printed = df._jdf.queryExecution().executedPlan().toString()
return printed.split("== Initial Plan ==")[0]
def operators(df):
"""The watched operators of the plan, root first, each followed by the keywords its line holds."""
found = []
for line in plan_text(df).splitlines():
bare = re.sub(r"^[\s+:*()\d-]*", "", line)
head = re.match(r"[A-Za-z]+", bare)
if head is None or head.group(0) not in WATCHED:
continue
words = [word for word in KEYWORDS if re.search(r"\b" + word + r"\b", line)]
found.append(" ".join([head.group(0), *words]))
return found
def counts(df):
"""How many of each watched operator the plan holds: a dict of operator name to count."""
tally = {}
for entry in operators(df):
name = entry.split(" ")[0]
tally[name] = tally.get(name, 0) + 1
return tally
def shape(df):
"""One line for a plan: the shuffles, the broadcasts and the sorts it holds."""
tally = counts(df)
return (f"Exchange = {tally.get('Exchange', 0)}, "
f"BroadcastExchange = {tally.get('BroadcastExchange', 0)}, "
f"Sort = {tally.get('Sort', 0)}")The first two differ in the one way that decides everything else. A filter decides each row where it already sits, so the task holding it finishes the work. A count per zone cannot: both tasks held rows of all 8 zones, and a count needs every row of one zone in one place. A shuffle: Spark re-distributing rows so that equal keys end up in one partition: each task of a stage writes its rows into one bucket per target partition, and the next stage's tasks fetch the buckets they need. puts them there, and Spark’s guide names its cost as disk writes, serialisation and network transfers:
The drawing shows the whole scene; its 6 steps are listed below.
Every step as text
- Two tasks, no sharing. The read gives 1,200,000 events in 2 partitions, one task each. A filter or a projection would finish here: every row is already where it needs to be, and nothing crosses between the two tasks.
- Equal zones must meet. A count per zone cannot finish locally, because both tasks hold rows of all 8 zones. Each task partly counts what it holds, then writes its rows into one bucket per target partition.
- The buckets land on local disk. Spark's own guide describes the map side: results are held in memory while they fit, then sorted by their target partition and written to a file, which the reading tasks later read as sorted blocks.
- Readers pull from every writer. Each of the 8 reducing tasks asks every writing task for its bucket, so a shuffle of W writers and R readers opens W times R transfers. This fetch is the step that costs the network and the disk.
- The shuffle starts a new stage. Spark cuts the job in two at the shuffle: the writing tasks are one stage, the reading tasks the next, and the second cannot start until the first has written every bucket.
- Eight rows out of 1,200,000. Each reading task now holds every row of its zones and finishes the count, which returns 8 rows. Almost all the work went into moving rows that the answer threw away.
Key idea
Exchange operators in a plan and you have counted the times your data crosses the cluster; everything else in the plan runs on rows where they already are.Checkpoint
Check: What a shuffle moves
Three ways to join two tables, and who chooses
Spark joins two tables by bringing equal keys together, but only two of its three strategies move the big side. A broadcast join: A join that sends the whole small side to every executor and probes it in memory against the big side's partitions, so no shuffle on the join key is needed. sends one whole side to every executor and probes it in memory, so the other side stays put. A sort-merge join: A join that shuffles both sides on the join key, sorts each partition, and merges them pairwise. The default for two large tables that will not fit a broadcast. shuffles both sides on the key, sorts each partition and merges them. A shuffled hash join: A join that shuffles both sides on the join key and then builds a hash table from one side's partition, instead of sorting both sides and merging them. shuffles both sides too, then builds a hash table from one side’s partition.
The planner chooses by size. The run printed the broadcast join threshold: The size up to which Spark sends a whole table to every executor rather than shuffling both sides of a join, set by spark.sql.autoBroadcastJoinThreshold; the value -1 switches broadcasting off., spark.sql.autoBroadcastJoinThreshold, as 10485760 bytes, the documented 10 MB default; 8 rows sit far below it, so the plan is a BroadcastHashJoin with no shuffle of the events. The threshold at -1 switches broadcasting off whatever the size, and a BROADCAST hint wins even above it.
| Strategy | What moves | What it needs |
|---|---|---|
| Broadcast hash join | one copy of the small side, to every executor | the small side below the threshold, and room for it |
| Sort-merge join | every row of both sides, once | nothing but the shuffle; it sorts and spills if it must |
| Shuffled hash join | every row of both sides, once | one task’s memory to hold its partition of a side |
Two settings and one measurement decide the rest. spark.sql.shuffle.partitions, 200 by default, sets how many shuffle partition: One of the partitions a shuffle writes its rows into, and one task of the stage that reads them. spark.sql.shuffle.partitions sets how many partitions a shuffle makes. each shuffle declares. adaptive query execution: A Spark feature, on by default, that re-plans parts of a query at runtime from the actual rows and bytes it has seen, under an AdaptiveSparkPlan root. may then join small ones together, a step called partition coalescing: Adaptive query execution joining neighbouring shuffle partitions into fewer, larger ones after it has measured a stage's output, so the next stage runs fewer and bigger tasks. with a target size of its own. Row counts per partition say where the work sits.
Checkpoint
Check: The three join strategies
Where each partition count comes from
Three rules set the three counts in a job, and mixing them up is why a tuning change often does nothing. A read’s partitions come from the source and the read settings: Spark packs files into partitions of up to spark.sql.files.maxPartitionBytes, 128 MB by default, charging every file an open cost of spark.sql.files.openCostInBytes, 4 MB, so small files share a task, and a generated range splits by the session’s cores, which is why every run here began with 2. A shuffle’s partitions come from spark.sql.shuffle.partitions, not from the stage before it. Files come from neither: a task writes one file in every partition folder it holds rows for, so a write can leave up to tasks times distinct partition values, a ceiling partitioning, clustering and file sizes measures.
The drawing shows the whole scene; its 5 steps are listed below.
Every step as text
- The source sets the first count. How many partitions a read produces comes from the source and the read settings: Spark packs a table's files into partitions of up to 128 MB, charging every file a 4 MB open cost, and splits a generated range by the session's cores.
- One task per partition. Each read partition becomes one task of the first stage. The lesson's runs read a generated range on two cores, so every one of them started with 2 partitions and 2 tasks.
- The shuffle count is a setting. The next count is not inherited. Every shuffle splits its output into as many partitions as spark.sql.shuffle.partitions says, whatever the stage before it had, and each one becomes a task.
- Adaptive execution may lower it. With adaptive query execution on, Spark can join small shuffle partitions together after measuring the stage, so the number of tasks that actually run is often well below the setting.
- The write multiplies the count. The last stage's tasks each write their own file in every partition folder they hold rows for, so the files on disk can reach tasks times distinct partition values.
Four things can still slow a stage whose counts are right, and Spark performance: AQE, skew and spills measures them: data skew, one key holding far more rows than the rest; salting, which spreads such a key over several artificial ones; a When a task's sort buffer or hash table outgrows the execution memory it can get, Spark writes part of it to the executor's local disk and reads it back later., a task writing its sort or hash table to disk; and caching a result that several actions reuse.
Checkpoint
Check: Where a partition count comes from
Practise
The self-check covers the three sections above.
Practice that adapts to you
Practise this lesson
8 questions about 6 ideas of this lesson. Each answer is explained at once, and what you get right is kept in this browser.
Broadcast, sort-merge or shuffled hash: the plan that proves the choice Pro: opens with a Pro, Premium or Ultimate pass
One join of one pair of tables can be planned four ways, and the plan says which one happened in operator names that take no interpreting. The run below records all four on the same 1,200,000 events and 8 zones, and every one of them returns the same rows.
In this part:
- The same join four ways, and the operators each one left
- What rules each strategy out
1 runnable example (join_strategies.py, join_strategies_session.py) · about 8 minutes
Opening this part with your pass…
This part opens with any MySmartCoPilot pass.
- Pro pass · 1 week₹79$2.99
- 30 days₹179$5.99
- 1 year₹1,699$45
One payment; a pass never renews. The price shown is the total you pay: no fees or charges are added at checkout.
Passes can’t be bought right now, so this part can’t be opened at the moment. The free part above stands on its own. If you have a pass, enter your licence key.
It also opens every other in-depth part in Learn and every Pro tool. Sales are final, except where the law gives you a right to withdraw, for example in the EU and the UK (refund policy).
How many partitions a stage really runs, before and after adaptive execution Pro: opens with a Pro, Premium or Ultimate pass
A shuffle declares its partition count before the query runs, and adaptive query execution decides afterwards how many tasks are worth using. Both numbers are readable, and they are usually not the same number.
In this part:
- Two hundred partitions declared, and what actually ran
- Sizing the number from the data, with the arithmetic
1 runnable example (shuffle_partitions.py, shuffle_partitions_session.py) · about 4 minutes
Opening this part with your pass…
This part opens with any MySmartCoPilot pass.
- Pro pass · 1 week₹79$2.99
- 30 days₹179$5.99
- 1 year₹1,699$45
One payment; a pass never renews. The price shown is the total you pay: no fees or charges are added at checkout.
Passes can’t be bought right now, so this part can’t be opened at the moment. The free part above stands on its own. If you have a pass, enter your licence key.
It also opens every other in-depth part in Learn and every Pro tool. Sales are final, except where the law gives you a right to withdraw, for example in the EU and the UK (refund policy).
Finding the expensive step from counts, not from guesses Pro: opens with a Pro, Premium or Ultimate pass
A stage is slow for a reason that is countable: one task holds several times the rows of its neighbours, or the query runs more stages than the work needs. Both numbers can be read off a running job, and the run below prints them for four shapes of the same 1,200,000 events.
In this part:
- Row counts per partition say where the work sits
- Stage counts follow the shuffles, and the UI fills in the rest
- Count the shuffles and files a job runs from its plan
- Results of the sample tests
- Key takeaways from these parts
1 runnable example (partition_counts.py, partition_counts_session.py) · 1 exercise with sample tests · about 10 minutes
Opening this part with your pass…
This part opens with any MySmartCoPilot pass.
- Pro pass · 1 week₹79$2.99
- 30 days₹179$5.99
- 1 year₹1,699$45
One payment; a pass never renews. The price shown is the total you pay: no fees or charges are added at checkout.
Passes can’t be bought right now, so this part can’t be opened at the moment. The free part above stands on its own. If you have a pass, enter your licence key.
It also opens every other in-depth part in Learn and every Pro tool. Sales are final, except where the law gives you a right to withdraw, for example in the EU and the UK (refund policy).
Interview questions
Warm-up: why is a shuffle the step people blame for a slow Spark job?
Because it is the only step that moves rows between tasks. Filters, projections and casts run where the rows already are. Grouping, ranking and two of the three join strategies need equal keys in one place, so every task writes its rows into buckets, those buckets reach a file once they outgrow memory, and every reading task fetches the ones addressed to it. That costs serialisation, disk and network, and it splits the job into stages that wait for each other.
Three longer questions start where this lesson ends:
-
Mid to senior
A nightly job joins 40 million delivery rows to a zones table. It used to plan a broadcast join; this week the plan shows a sort-merge join with two Exchange operators, and the job now misses its window. How do you investigate and what do you change?
-
Mid to staff
A platform team sets spark.sql.shuffle.partitions to 4000 for every job, because one large job needed it. Small jobs now run thousands of tasks that hold almost nothing. How do you decide the number, per job and as a default?
-
Senior to staff
A daily job repartitions its delivery events by city, counts them per zone and date, and writes the counts partitioned by date. Its plan holds two Exchange operators, and each run leaves about 6000 files in 30 date folders. Cut the shuffles and the files without changing a row of the output.
Flashcards · 6 cards
Recall it, then check
Say the answer to yourself before you look: recalling it is what makes it stick.
-
Why is the shuffle the step people blame for a slow Spark job?
It is the only step that moves rows between tasks so that equal keys meet. Every task writes its rows into buckets and every reader fetches its own, which costs serialisation, disk and network and splits the job into stages.
-
Why does a filter plan no Exchange while a count per zone plans one?
A filter decides each row where it already sits. A count needs every row of one zone in one place, and only a shuffle puts them there.
-
What are Spark's three join strategies, and which move the big side?
A broadcast hash join sends one whole side to every executor and leaves the other in place. A sort-merge join and a shuffled hash join both shuffle the two sides on the key; the first sorts and merges, the second builds a hash table per partition.
-
When does Spark broadcast one side of a join?
When its estimated size is under spark.sql.autoBroadcastJoinThreshold, 10485760 bytes by default. A value of -1 turns broadcasting off, and a BROADCAST hint wins even above the threshold.
-
Joining 1,200,000 delivery events to 8 city zones, how many event rows crossed the network?
None. The 8-row table sat far below the broadcast threshold, so the plan was a BroadcastHashJoin and the events never left their tasks.
-
Where do the partitions of a file read and of a shuffle come from?
A read packs files into partitions of up to spark.sql.files.maxPartitionBytes, 128 MB by default, charging each file an open cost of 4 MB. A shuffle declares spark.sql.shuffle.partitions, 200 by default, which adaptive execution may coalesce.
Key takeaways
- A shuffle re-distributes rows so that equal keys meet, and each one stands in the plan as an
Exchangeoperator. - Spark broadcasts a join side whose estimated size is under 10485760 bytes by default, and otherwise shuffles both sides.
- A file read packs files into partitions by bytes, a shuffle declares
spark.sql.shuffle.partitionspartitions, and a write leaves up to tasks times partition values in files.
Terms in this lesson
- broadcast join threshold
- The size up to which Spark sends a whole table to every executor rather than shuffling both sides of a join, set by spark.sql.autoBroadcastJoinThreshold; the value -1 switches broadcasting off.
- Exchange operator
- The operator in a Spark physical plan that stands for one shuffle: rows leave the partitions they were computed in and are re-distributed, usually by a hash of the key.
- partition coalescing
- Adaptive query execution joining neighbouring shuffle partitions into fewer, larger ones after it has measured a stage's output, so the next stage runs fewer and bigger tasks.
- shuffle
- Spark re-distributing rows so that equal keys end up in one partition: each task of a stage writes its rows into one bucket per target partition, and the next stage's tasks fetch the buckets they need.
- shuffle partition
- One of the partitions a shuffle writes its rows into, and one task of the stage that reads them. spark.sql.shuffle.partitions sets how many partitions a shuffle makes.
- shuffled hash join
- A join that shuffles both sides on the join key and then builds a hash table from one side's partition, instead of sorting both sides and merging them.
References
- Apache Spark: RDD programming guide (shuffle operations) (The Apache Software Foundation)
- Apache Spark: SQL performance tuning (The Apache Software Foundation)
- Apache Spark: cluster mode overview (The Apache Software Foundation)
- Apache Spark: monitoring and the web UI (The Apache Software Foundation)
Report a problem with this lesson
Kept only in this browser. Your Learn progress