Understanding why more executors didn’t help and what the Spark UI revealed about the real bottleneck
I had a Spark job that refused to speed up. The cluster had enough CPU and memory available, yet the runtime barely changed as I increased the compute.
Instead of scaling further, I opened the Spark UI to understand where the time was actually going. That is when I realized the bottleneck was not the amount of compute available, but how Spark was executing the query.
In this article, I’ll walk through the experiments and Spark UI observations that helped uncover the bottleneck, understand Spark’s join strategy, and eventually reduce the shuffle during the join.
If you’ve ever looked at a slow Spark job and wondered whether the problem was the cluster, the code, or something deeper in the execution plan, this kind of hands-on problem solving is exactly what we focus on in TrendyTech’s Ultimate Data Engineering Masters Program. The goal is not just to learn Spark concepts, but to understand how they behave in real-world scenarios and how to approach performance problems like a Data Engineer.
Table of Contents
Problem: More Compute, Similar Runtime
To understand why adding more executors wasn’t improving the Spark job, I simplified the problem to a basic join and observed how Spark executed it.
I joined two tables: fact_orders with 20 million rows and dim_customers with 2 million rows, using customer_id as the join key.
The job ran on a 4-worker Databricks cluster with 16 cores and 64 GB of memory. I configured 128 shuffle partitions and disabled Adaptive Query Execution (AQE) so that I could observe Spark’s baseline execution behavior without runtime optimizations changing the physical plan.
At first glance, everything looked normal.
All 128 tasks completed successfully. There was no data skew, no task failure, and no memory pressure. The work was also evenly distributed across the executors.
But when I looked at the Spark UI metrics, something stood out.
The query took 28 seconds to complete and involved approximately 537 MB of shuffle read across the cluster. But what caught my attention was the median task duration, which was only around 0.2 seconds.
That difference made me look deeper into how Spark was executing the join.
So I opened the Spark SQL physical plan.
What the Spark UI Revealed
When I looked at the Spark SQL physical plan, I could see that Spark had chosen a SortMergeJoin, with an Exchange on both sides of the join:
+- SortMergeJoin
:- Exchange (hashpartitioning(customer_id, 128))
+- Exchange (hashpartitioning(customer_id, 128))
The two Exchange operators were the important part.
Before Spark could perform the join, both datasets had to be shuffled across the cluster based on customer_id. In other words, rows with the same customer_id had to be brought into the same partition so that Spark could join them.
The Spark UI stage metrics reflected the same behavior:
- ~494 MB Shuffle Write
- Shuffle Read: ~537 MB
- Total Runtime: 28 seconds
The physical plan made the issue clear. Before Spark could perform the join, both datasets had to go through an Exchange, resulting in a shuffle across the cluster.
The next question was:
Could I reduce this data movement instead of simply adding more executors?
Adding more executors gave Spark additional CPU, memory, and parallelism, but it did not remove the Exchange operators from the physical plan.
For this join, Spark still had to redistribute the data by customer_id before the SortMergeJoin could happen.
So instead of asking:
“How can I give Spark more compute?”
I started asking a different question:
“Can I avoid shuffling both tables in the first place?”
That led to the next experiment: Broadcast Join.
Solution 1: Reducing Shuffle with a Broadcast Join
In the previous execution plan, Spark used a SortMergeJoin, which required both datasets to be shuffled based on customer_id before the join could happen.
But there was an important difference between the two tables:
- fact_orders: 20 million rows
- dim_customers: 2 million rows
Since dim_customers was significantly smaller than fact_orders, shuffling both tables wasn’t necessarily the best approach.
This is where a Broadcast Join can help.
Instead of redistributing both datasets across the cluster, Spark sends a copy of the smaller table to the executors. Each executor can then join its local partition of the larger fact_orders table with the broadcasted dim_customers data.
To see how this changed the execution plan, I explicitly broadcast the smaller table:
Note: Adaptive Query Execution (AQE) remained disabled for this experiment, just as it was in the baseline run. The broadcast was applied explicitly so that I could clearly observe the change in Spark’s execution plan.
When I opened the Spark SQL physical plan again, the difference was visible.
Spark has replaced the SortMergeJoin with a BroadcastHashJoin.
Spark SQL physical plan showing the 2M-row dim_customers table going through a BroadcastExchange while the 20M-row fact_orders table avoids the shuffle Exchange
More importantly, the Exchange on the 20-million-row fact_orders table was gone. The larger table no longer needed to be repartitioned by customer_id for the join.
The smaller dim_customers table instead went through a BroadcastExchange, allowing Spark to send a copy of it to the executors.
So instead of shuffling both sides of the join, Spark changed the data movement pattern: broadcast the smaller table and avoid shuffling the larger one.
But did removing the large-table shuffle actually make the overall job significantly faster?
The Spark UI showed something interesting.
What Changed After the Broadcast Join?
The physical plan looked much better. The 20-million-row fact_orders table no longer went through an Exchange, which meant Spark had eliminated the shuffle on the larger side of the join.
But when I looked at the overall execution time, the improvement was not as significant as I initially expected.
The reason became clearer when I looked deeper into the Spark UI.
Before Spark could perform the BroadcastHashJoin, it first had to prepare the smaller dim_customers table and distribute it across the cluster through a BroadcastExchange.
That broadcast operation also had a cost.
So while the Broadcast Join removed the expensive shuffle of the 20-million-row fact table, Spark still had to spend time preparing and distributing the 2-million-row dimension table to the executors.
This was an important observation.
A Broadcast Join does not eliminate all data movement. Instead, it changes which data moves. Rather than repartitioning both sides of the join, Spark distributes the smaller table and keeps the larger table from being shuffled.
For an asymmetric join like this, where one table is significantly larger than the other, that can be a much better execution strategy.
But Broadcast Joins also have an important limitation.
The table being broadcast must be small enough to be handled efficiently in memory. As the size of the smaller table increases, broadcasting it becomes more expensive and may no longer be a practical option.
That raised the next question:
What happens when both tables are too large to broadcast?
If neither side can be treated as the smaller lookup table, can we still avoid shuffling both datasets every time they are joined?
That is where bucketing comes in.
Understanding Spark performance becomes much easier when you can connect what you see in the Spark UI with the decisions Spark is making underneath. We go much deeper into this approach in TrendyTech’s Databricks Performance Tuning Program, where join strategies, shuffle behavior, execution plans, and Spark performance bottlenecks are explored through hands-on scenarios.
Solution 2: Using Bucketing for Large-to-Large Joins
Broadcast Join worked because one side of the join was significantly smaller. But what happens when both tables are too large to broadcast?
To test this, I changed the setup.
This time, both datasets had 20 million rows. The orders table was stored as bucketed_orders and the customers table as bucketed_customers.
Both tables were:
- Bucketed on customer_id
- Written using 128 buckets
- Sorted by customer_id
The idea behind bucketing in Apache Spark is to organize the data in advance based on the join key. If two tables are bucketed using the same join key and compatible bucket configuration, Spark can potentially reuse that existing data distribution instead of reshuffling the data every time the tables are joined.
I then ran the same join:
Based on the setup, I expected Spark to reuse the existing bucket layout and avoid the shuffle.
But when I opened the Spark SQL physical plan, I saw something unexpected.
There was still an Exchange.
Spark UI showing approximately 499.4 MiB of shuffle read and shuffle write during the 20M × 20M bucketed join.
Both tables were bucketed on the same join key and used the same number of buckets.
So why was Spark still shuffling the data?
The answer was hidden in one small detail in the physical plan.
Why Was Spark Still Shuffling the Bucketed Tables?
To understand why the Exchange was still present, I looked more closely at the Spark SQL physical plan.
One detail stood out:
hashpartitioning(cast(customer_id as bigint), 128)
Spark was applying a cast() to customer_id before partitioning the data.
That led me back to the schemas of the two tables.
In bucketed_orders, customer_id was stored as an INT.
In bucketed_customers, customer_id was stored as a BIGINT.
Because the data types did not match, Spark had to apply an implicit cast to one side of the join.
Spark SQL physical plan showing an implicit cast on customer_id before the Exchange operation.
This small schema difference was enough to prevent Spark from reusing the existing bucket distribution for the join.
Even though both tables had been bucketed on customer_id using 128 buckets, the join expression now involved cast(customer_id as bigint) on one side. As a result, the existing bucket layout could no longer satisfy the distribution required by the join, and Spark introduced an Exchange.
So the problem wasn’t the number of buckets or the join key itself.
It was the schema mismatch between the join columns.
The next step was straightforward: align the data types and run the same join again.
Note: INT and BIGINT may both store whole numbers, but Spark treats them as different data types. INT is a 32-bit integer type, while BIGINT is a 64-bit integer type. When they are used as join keys, Spark may need to implicitly cast one side to a common type. In this experiment, that cast changed the join expression from customer_id to cast(customer_id as bigint), which prevented Spark from reusing the existing bucket distribution and introduced an Exchange.
Fixing the Schema Mismatch
Once I identified the schema mismatch, the fix was straightforward.
I aligned the data type of customer_id in both tables before writing the bucketed data. In this case, I cast customer_id in bucketed_customers from BIGINT to INT so that it matched bucketed_orders.
I then ran the same join again with the aligned schemas.
This time, the Spark SQL physical plan looked different.
The Exchange that appeared earlier was gone. Both bucketed tables could now feed into the SortMergeJoin without Spark having to repartition the data across the cluster.
The change was also visible in the execution time.
Before fixing the schema mismatch, the bucketed join took approximately 26 seconds. After aligning the customer_id data types, the same join completed in approximately 5 seconds.
Spark UI showing the bucketed join completing in approximately 5 seconds after schema alignment, compared with 26 seconds before the fix.
The cluster had not changed. The datasets were still the same size, and both tables still contained 20 million rows.
The difference was that Spark could now reuse the existing bucket distribution instead of introducing another shuffle.
This experiment highlighted an important detail about Spark bucketing: matching the bucket count and join key is not enough. The data types of the join columns also need to be compatible with the bucket distribution Spark is trying to reuse.
A small schema mismatch such as INT versus BIGINT can be enough to introduce an Exchange and bring the shuffle back.
What This Experiment Taught Me About Spark Bucketing?
At first, both tables appeared to be correctly bucketed. They used the same customer_id column, the same 128 buckets, and the same sorting configuration.
But the Spark UI showed that this alone was not enough.
The INT versus BIGINT mismatch on customer_id caused Spark to apply an implicit cast during the join. Once the join expression no longer matched the existing bucket distribution, Spark introduced an Exchange and the shuffle returned.
After aligning the schemas, the Exchange disappeared and the same bucketed join completed in approximately 5 seconds instead of 26 seconds.
The important lesson here is not simply that bucketing makes joins faster.
Bucketing helps when Spark can actually reuse the way the data has already been distributed. That means details such as the join key, bucket configuration, and schema of the join columns matter.
And this is exactly why checking the Spark SQL physical plan is so important.
Two queries can look almost identical in code but result in very different execution plans.
Performance tuning is rarely about memorizing a list of configurations.
It is about knowing where to look when a Spark job slows down and understanding what the Spark UI and execution plan are telling you. This is the same hands-on approach we follow in the Databricks Performance Tuning Program, where you learn to diagnose and optimize Spark workloads through practical scenarios.
Conclusion: The Real Bottleneck Wasn’t Compute
When this Spark job refused to speed up, adding more compute seemed like the obvious solution.
But the Spark UI told a different story.
The initial SortMergeJoin required both datasets to go through an Exchange, resulting in significant shuffle across the cluster. When one side of the join was smaller, a Broadcast Join changed the execution plan and avoided shuffling the larger fact table.
But when both tables were large, broadcasting was no longer a practical option.
That’s where bucketing became useful.
Even then, the first bucketed join still shuffled approximately 499.4 MiB of data. The reason turned out to be something easy to miss: customer_id was an INT in one table and a BIGINT in the other.
That small schema mismatch introduced an implicit cast, prevented Spark from reusing the existing bucket distribution, and brought the Exchange back.
Once the schemas were aligned, the Exchange disappeared and the same 20M × 20M bucketed join went from approximately 26 seconds to 5 seconds, without changing the cluster size.
The biggest learning from these experiments was simple:
Before adding more executors to a slow Spark job, look at how the data is moving.
Open the Spark UI. Check the physical plan. Look for Exchange operators and understand why Spark is shuffling the data.
Sometimes the problem isn’t that Spark needs more compute.
It’s that the execution plan is making Spark move more data than necessary.
“A slow Spark job does not always need a bigger cluster. Sometimes the real problem is hidden in the execution plan, in an Exchange you didn’t expect, or even in something as small as a schema mismatch.”
Performance tuning is rarely about adding more compute or finding one configuration that makes every Spark job faster. It is about understanding where the time is being spent, how Spark is moving the data, and which optimization actually makes sense for that workload.
If you want to build that understanding through hands-on Spark and Databricks scenarios, explore TrendyTech’s Databricks Performance Tuning Program.
And if you’re looking for a broader learning path covering Data Engineering from fundamentals to advanced technologies, explore the Ultimate Data Engineering Masters Program.
Documentation References
Databricks Optimization and Performance Tuning Advanced Program by Sumit Mittal (TrendyTech)
Frequently asked questions (FAQs)
It comes down to cardinality on both sides, not just row count:
1. One table is small enough to comfortably fit in executor memory (typically tens to low hundreds of MB, adjustable via spark.sql.autoBroadcastJoinThreshold) → broadcast join. It removes the Exchange on the large side entirely.
2. Both tables are large and get joined repeatedly on the same key → bucketing is the better long-term investment, since it avoids repeated shuffles across many queries, not just one.
3. Neither condition holds → you’re likely stuck with a SortMergeJoin and should focus on reducing shuffle volume (column pruning, filtering earlier, partition sizing) rather than eliminating it.
Not usually. Check the Spark UI first: if median task duration is tiny but total shuffle read/write is large, you’re compute-rich but I/O-bound. That’s a signal the physical plan, not the cluster, is the bottleneck. Look at the SQL tab for Exchange operators before reaching for more executors; they tell you Spark is redistributing data across the network, which more nodes won’t fix.
Before digging further, check three things that line up exactly: bucket count, join key, and the data type of the join column on both sides. A mismatch like INT vs BIGINT forces an implicit cast, which changes the join expression Spark actually evaluates. A bucket layout that was valid for customer_id isn’t valid for cast(customer_id as bigint). Confirm this in the physical plan (explain()), not just by inspecting the DDL, since the cast is often invisible until you look at the actual join expression.
No, it changes what moves, not whether anything moves. The broadcast side still gets serialized and pushed to every executor via BroadcastExchange, and that has a real cost, especially at scale or under concurrent broadcasts (it can also cause driver or executor memory pressure or OOM if the table grows unexpectedly).Treat the threshold as a dynamic setting rather than a one-time default, and be sure to monitor broadcast size against production traffic instead of relying on sampled dev data.
A repeatable sequence:
1. Check stage-level shuffle read and write in the Spark UI to confirm shuffle is the cost driver.
2. Pull the physical plan (explain(true)) and look specifically for Exchange nodes and any implicit cast() in the join condition.
3. If a cast is present, trace it back to the schema definitions. This is the most commonly missed culprit.
4. Decide broadcast vs. bucketing vs. accepting the shuffle based on table sizes and query recurrence
5. Re-run and compare shuffle bytes and stage duration before and after. Don’t just guess by looking at total runtime, since other stages can mask the improvement.
How do I decide between a broadcast join and bucketing?
It comes down to cardinality on both sides, not just row count:
1. One table is small enough to comfortably fit in executor memory (typically tens to low hundreds of MB, adjustable via spark.sql.autoBroadcastJoinThreshold) → broadcast join. It removes the Exchange on the large side entirely.
2. Both tables are large and get joined repeatedly on the same key → bucketing is the better long-term investment, since it avoids repeated shuffles across many queries, not just one.
3. Neither condition holds → you’re likely stuck with a SortMergeJoin and should focus on reducing shuffle volume (column pruning, filtering earlier, partition sizing) rather than eliminating it.
My Spark job has spare CPU/memory but the runtime won’t shift . Is scaling out even the right move?
Not usually. Check the Spark UI first: if median task duration is tiny but total shuffle read/write is large, you’re compute-rich but I/O-bound. That’s a signal the physical plan, not the cluster, is the bottleneck. Look at the SQL tab for Exchange operators before reaching for more executors; they tell you Spark is redistributing data across the network, which more nodes won’t fix.
Why would bucketed tables still shuffle on a join? What should I check before assuming bucketing “isn’t working”?
Before digging further, check three things that line up exactly: bucket count, join key, and the data type of the join column on both sides. A mismatch like INT vs BIGINT forces an implicit cast, which changes the join expression Spark actually evaluates. A bucket layout that was valid for customer_id isn’t valid for cast(customer_id as bigint). Confirm this in the physical plan (explain()), not just by inspecting the DDL, since the cast is often invisible until you look at the actual join expression.
Is a broadcast join always “free” once it avoids the big shuffle?
No, it changes what moves, not whether anything moves. The broadcast side still gets serialized and pushed to every executor via BroadcastExchange, and that has a real cost, especially at scale or under concurrent broadcasts (it can also cause driver or executor memory pressure or OOM if the table grows unexpectedly).Treat the threshold as a dynamic setting rather than a one-time default, and be sure to monitor broadcast size against production traffic instead of relying on sampled dev data.
What’s the actual workflow for diagnosing shuffle-heavy jobs, beyond “look at the Spark UI”?
A repeatable sequence:
1. Check stage-level shuffle read and write in the Spark UI to confirm shuffle is the cost driver.
2. Pull the physical plan (explain(true)) and look specifically for Exchange nodes and any implicit cast() in the join condition.
3. If a cast is present, trace it back to the schema definitions. This is the most commonly missed culprit.
4. Decide broadcast vs. bucketing vs. accepting the shuffle based on table sizes and query recurrence
5. Re-run and compare shuffle bytes and stage duration before and after. Don’t just guess by looking at total runtime, since other stages can mask the improvement.
