Skip to main content

Narrow, Wide, and the Broadcast Decision at 10 MB and 100 MB

Level 10: Distributed Compute & Data Operationsmedium28 minnarrow vs wide transformationsbroadcast join vs sort-merge joinautoBroadcastJoinThresholdadaptive query executionshuffle-volume triageCASE classificationwindow functions

Read narrow from wide off the shuffle metrics, pick auto-broadcast, hinted broadcast, or sort-merge with the real byte thresholds, and flag the stages whose shuffle write crosses the 10 GB line.

What you already have, and what this adds

Level 6 taught the narrow/wide split and graded one binary flag: is this dimension under 10 MB, yes or no. None of that is repeated here. What is new is the band and the bet. A broadcast is not a free win you either qualify for or do not; it is a spectrum with a judgment call in the middle, and the judgment is about the driver's memory.

Narrow and wide, read off the metrics

You can tell narrow from wide without seeing a single operator name, just from the stage metrics:

  • A stage with zero shuffle read was fed narrowly. Nothing had to move to reach it, so it pipelines with whatever came before it.
  • A stage with shuffle read above zero sits immediately downstream of a wide step. Rows moved across the network to get there.
  • The shuffle write belongs to the producing stage. That is the Level 6 attribution rule, and here it stops being a fact to memorize and becomes the thing you read: the stage that shows the shuffle bytes is the one that paid to write them, not the one that consumed them.

The broadcast band

A join has to get matching keys onto the same machine. There are two ways to do that, and the size of the smaller side decides which one you get.

Table
The broadcast band. The middle row is a judgment call, not a rule, because a hinted broadcast spends driver memory and network on every task slot.
smaller sideengine choicedata movementwhat breaks if you force it
10485760 bytes (10 MB) or lessbroadcast, automaticallysmall side copied to every executornothing: this is the default
one byte over 10 MB, up to about 100 MBbroadcast, only if you hint itcollected to the driver, then copied outdriver memory, if the driver cannot hold it
above about 100 MBsort-merge joinboth sides shuffled and sorted by keydriver OOM, the classic forced-broadcast failure
The broadcast band. The middle row is a judgment call, not a rule, because a hinted broadcast spends driver memory and network on every task slot.

spark.sql.autoBroadcastJoinThreshold defaults to exactly 10485760 bytes, which is 10 MB. At or under that line the planner broadcasts the small side without being asked: it ships a copy of the whole small table to every executor, so the join becomes narrow and no shuffle happens at all.

The threshold is a maximum, not a strict cutoff. Spark tests sizeInBytes <= threshold, so a side measured at exactly 10485760 bytes is auto-broadcast and a side one single byte larger is not. The capture below has both, one byte apart, because getting that boundary backwards is the kind of detail an interviewer uses to tell reading from doing.

Between 10 MB and roughly 100 MB you can force the same thing with a broadcast hint. That is where the bet lives. The small side is first collected to the driver, and only then copied out to every executor, so a forced broadcast spends driver memory once and network bandwidth once per task slot. If the driver cannot hold it, you get the driver OOM from the execution-model lesson: a machine that was only supposed to schedule work is now holding data.

Above roughly 100 MB the sane answer is sort-merge join. Both sides shuffle, both sides sort by the join key, and the merge streams. It is slower than a broadcast but it scales, and forcing a broadcast there is how a job that used to be slow becomes a job that does not finish.

AQE moves the line at runtime

Adaptive Query Execution has been on by default since Spark 3.2 and Glue 4.0. It re-plans between stages using the sizes it actually observed, which means a join the planner estimated at 500 MB can arrive post-filter at 8 MB, and AQE converts the sort-merge to a broadcast right then.

This is why planned_strategy and actual_strategy can disagree in the capture you are about to query, and it is why "check what actually ran, not what the plan said" is the follow-up that separates a strong answer from a memorized one.

The 10 GB line

There is a point where tuning stops being the right conversation. A join stage that writes more than 10 GB of shuffle is the physical evidence behind the design-time heuristic Level 7 states as "keep the star schema while the join side stays under roughly 10 GB". When this query returns rows, the fix is not a config flag. The fix is modeling: pre-join into one wide table, or co-locate the two sides on a shared distribution key so the join stops needing a network move at all.

Common mistake: reading the shuffle write off the stage that consumed it. The consuming stage shows shuffle read; the stage that shows shuffle write is the producer, and that is the one whose output you would shrink by changing the model.

Interview nuance: candidates who answer "broadcast the small table" get a follow-up asking how small. The strong answer names all three bands: 10 MB automatic, up to about 100 MB by hint if the driver can hold it, sort-merge past that. Adding "and AQE may convert it at runtime anyway, so I would check the actual strategy in the UI" is the sentence that makes it sound like you have done this.

On a real platform this differs. Redshift expresses the same decision as a table property rather than a query hint: DISTSTYLE ALL keeps a full copy of a small dimension on every node, which is a permanent broadcast decided at load time instead of a per-query one. Snowflake and BigQuery decide it for you and expose the choice only in the query profile. The reasoning is identical everywhere: how big is the small side, and who has to hold it.

Sample data for this example
CREATE TABLE join_candidates (
  join_id          TEXT,
  left_table       TEXT,
  left_size_bytes  INTEGER,   -- the planner's size estimate for the left input
  right_table      TEXT,
  right_size_bytes INTEGER,   -- the planner's size estimate for the right input
  planned_strategy TEXT,      -- what the physical plan picked before the job ran
  actual_strategy  TEXT       -- what the Spark UI shows actually ran; AQE can change it
);
INSERT INTO join_candidates (join_id, left_table, left_size_bytes, right_table, right_size_bytes, planned_strategy, actual_strategy) VALUES
  ('j01', 'fact_orders',      48000000000, 'dim_currency',       524288, 'broadcast_hash', 'broadcast_hash'),
  ('j02', 'fact_clicks',     120000000000, 'dim_country',       2097152, 'broadcast_hash', 'broadcast_hash'),
  ('j03', 'fact_sessions',    30000000000, 'dim_device',       41943040, 'sort_merge',     'broadcast_hash'),
  ('j04', 'fact_orders',      48000000000, 'dim_customer',     83886080, 'sort_merge',     'sort_merge'),
  ('j05', 'fact_events',     200000000000, 'dim_product',     314572800, 'sort_merge',     'sort_merge'),
  ('j06', 'fact_orders',      48000000000, 'fact_returns',   6291456000, 'sort_merge',     'sort_merge'),
  ('j07', 'fact_clicks',     120000000000, 'dim_campaign',    524288000, 'sort_merge',     'broadcast_hash'),
  ('j08', 'dim_experiment',       8388608, 'fact_page_views',90000000000, 'broadcast_hash', 'broadcast_hash'),
  ('j09', 'fact_orders',      48000000000, 'dim_user_agent', 2147483648, 'sort_merge',     'sort_merge'),
  ('j10', 'dim_locale',          10485760, 'fact_web_sessions',60000000000, 'sort_merge',   'sort_merge'),
  ('j11', 'fact_shipments',   80000000000, 'dim_carrier',     104857600, 'sort_merge',     'sort_merge'),
  ('j12', 'fact_impressions', 75000000000, 'dim_channel',      10485761, 'sort_merge',     'sort_merge');
Worked example (SQL)
-- The join candidates with the size of their smaller side, and what the plan
-- said versus what the UI shows actually ran.
SELECT join_id, left_table, right_table,
       ROUND(MIN(left_size_bytes, right_size_bytes) / 1048576.0, 1) AS small_side_mb,
       planned_strategy, actual_strategy
FROM join_candidates
ORDER BY small_side_mb;

Apply

Your turn

The task this lesson builds to.

Write a query that returns the join strategy each join should use, as (join_id, small_side_mb, strategy), smallest small side first, over join_candidates(join_id, left_table, left_size_bytes, right_table, right_size_bytes, planned_strategy, actual_strategy).

The smaller side is whichever of left_size_bytes and right_size_bytes is lower. Report it as small_side_mb (1 MB is 1048576 bytes) rounded to 1 decimal. strategy is 'auto_broadcast' when the smaller side is 10485760 bytes or less, since the threshold is a maximum rather than a strict cutoff, 'hint_broadcast' above that up to and including 104857600 bytes, and 'sort_merge' above that. Alias the columns exactly, ordered by small_side_mb ascending and then join_id ascending.

4 hints and 1 automated check are waiting in the workspace.

Practice

Make it stick

A second problem on the same idea, plus 3 bonus drills.

Write a query that returns the shuffle-heavy stages a modeling change would target, as (stage_id, shuffle_write_gb, pct_of_job_shuffle), biggest first, over spark_stages(stage_id, job_id, name, shuffle_read_mb, shuffle_write_mb, num_tasks).

Keep only stages writing more than 10 GB of shuffle (1 GB is 1024 MB, so more than 10240 MB). Report shuffle_write_gb rounded to 1 decimal and pct_of_job_shuffle rounded to 2 decimals, where the denominator is that stage's own job's total shuffle write across every stage in the job, including the ones this query filters out. Ordered by shuffle_write_gb descending.

2 hints and 1 automated check are waiting in the workspace.