F
FNU Aditya Ranjan
Guest
Why adding executors rarely fixes the most expensive Spark problems
A slow Spark job is often treated as a sizing problem: add executors, increase memory, raise shuffle partitions, and try again. That can hide the real issue. The expensive failures I see in Spark architectures usually come from unnecessary shuffles, skewed joins, poor file layout, stale or missing statistics, oversized scans, and pipelines designed without regard for the physical execution plan. Spark 4.0.4 can mitigate several of these problems through Adaptive Query Execution (AQE), but AQE is not a substitute for sound data layout and query design. This article presents an architecture-first way to diagnose Spark cost before reaching for more compute.
When a Spark workload misses its SLA, the first conversation often starts with infrastructure. How many executors did we have? How much memory? Should we increase executor cores? Would a larger cluster finish faster?
Those are reasonable questions, but they are often asked too early. Spark performance is the result of a chain of architectural decisions: how data is stored, how it is partitioned, how much of it is scanned, where data moves across the network, how joins are planned, whether a few partitions dominate the stage, and only then how much compute is available to execute the plan.
That distinction matters in cloud environments because an inefficient plan can still finish successfully. The bill simply absorbs the inefficiency. A job that completes after adding more compute can look “fixed” even though the architecture is still forcing Spark to read, shuffle, sort, spill, or recompute far more data than necessary.
I prefer to treat Spark performance as an execution-plan problem first and a cluster-sizing problem second. Before increasing capacity, I want to know what work Spark is actually being asked to perform.
A useful way to think about Spark cost is to follow the movement of data rather than the number of machines.
At a high level, a Spark SQL or DataFrame workload can become expensive in five places: input scanning, repartitioning and shuffle, joins and aggregations, memory pressure/spill, and output layout. These costs compound. A poor file layout can increase the scan. A wide transformation can then shuffle that larger dataset. A skewed join can make one or two tasks run much longer than the rest. The output can finally create another small-file problem for the next pipeline.
Apache Spark's own performance-tuning documentation reflects this architecture. It separately addresses file partitioning, statistics, join strategy, caching, shuffle partitioning, Adaptive Query Execution, skew handling, and storage-partition joins. That is a useful reminder: there is no single “Spark tuning” knob.
Before looking at executor memory, I look at how much data the job reads and how that data is physically organized.
For file-based sources, Spark uses settings such as `spark.sql.files.maxPartitionBytes` when packing files into input partitions. In Spark 4.0.4, the documented default is 128 MB. Spark also accounts for the estimated cost of opening files through `spark.sql.files.openCostInBytes`. Those settings matter, but they cannot repair a dataset that is badly organized for its access patterns.
If a query needs one day of data but the storage layout forces it to inspect months of files, the first optimization target is not executor memory. It is the amount of work entering the plan.
The same principle applies to columns. Columnar formats such as Parquet allow Spark to avoid reading unnecessary columns when the query and source support column pruning. Selecting every column “just in case” increases I/O, serialization, memory pressure, and often the amount of data that later participates in a shuffle.
My first three questions are therefore simple: What fraction of the dataset is being scanned? Are filters being pushed as early as possible? Are we reading columns that never contribute to the result?
Shuffles are not inherently bad. Distributed joins and aggregations frequently require data to move between executors. The mistake is treating that movement as free.
Spark's `spark.sql.shuffle.partitions` controls the number of partitions used for shuffles in joins and aggregations; Spark 4.0.4 documents a default of 200. A fixed number is obviously not ideal for every workload. Two hundred partitions may be excessive for a small intermediate result and far too few for a very large one.
AQE helps here. Spark can coalesce post-shuffle partitions using runtime map-output statistics, reducing the number of tiny downstream tasks. Current Spark documentation notes that AQE is enabled by default and can adapt the physical plan while a query is executing.
But AQE does not mean engineers can stop thinking about shuffle. If the logical design repeatedly repartitions the same large dataset, performs avoidable global sorts, or joins at a grain much finer than the business requirement, adaptive execution is optimizing an unnecessarily expensive plan.
A practical review question is: for every `Exchange` in the physical plan, can I explain why the data must move? If the answer is no, I have found something worth investigating.
One of the easiest Spark problems to misdiagnose is skew.
A stage may contain hundreds of tasks, most of which finish quickly, while a handful run for much longer. Adding executors may do very little because the bottleneck is not a shortage of parallel task slots. Too much data has landed in a small number of partitions.
Spark AQE includes skew-join optimization for sort-merge joins. In Spark 4.0.4, `spark.sql.adaptive.skewJoin.enabled` is documented as enabled by default. Spark can identify skewed shuffle partitions and split them into smaller pieces, replicating the other join side when needed.
That is valuable, but I still want to understand why the skew exists. Common causes include hot business keys, null or default values concentrated under one key, and joining at an inappropriate grain. Sometimes the correct solution is architectural: filter earlier, separate exceptional keys, pre-aggregate, or redesign the join. Runtime mitigation and data-model correction solve different parts of the problem.
Join performance is another place where architecture and metadata meet.
Spark can automatically broadcast a sufficiently small relation, avoiding a shuffle of the larger side. Spark 4.0.4 documents `spark.sql.autoBroadcastJoinThreshold` with a 10 MB default. Developers can also provide join hints such as `BROADCAST`, `MERGE`, `SHUFFLE_HASH`, and `SHUFFLE_REPLICATE_NL`, although Spark does not guarantee that every requested strategy is applicable to every join.
The more important point is that planning depends on statistics. Spark's documentation explicitly warns that missing or inaccurate statistics can hinder its ability to choose an optimal plan. Catalog statistics can be updated with `ANALYZE TABLE`, while `EXPLAIN COST` or `DataFrame.explain(mode="cost")` can expose estimates used by the optimizer.
This is why I do not treat metadata maintenance as housekeeping. Statistics are part of query performance. A platform that carefully tunes compute but allows optimizer statistics to become unreliable is making execution decisions with incomplete information.
Memory tuning has a place, especially when workloads genuinely require large in-memory working sets. But an out-of-memory error or heavy spill is not automatically proof that the executor is too small.
Spark's tuning guide emphasizes the interaction among memory consumption, serialization, garbage collection, and the cost of Java objects. From an architecture perspective, I first ask why the working set is so large. Are we caching data that is used once? Did a join multiply rows unexpectedly? Are we aggregating before filtering? Are partitions too large because the upstream partitioning strategy is poor?
Increasing memory without answering those questions can convert a deterministic design problem into a more expensive design problem that happens less often.
Consider a hypothetical daily transformation. The numbers below are illustrative; they are not benchmark results from a production system.
A pipeline reads a large fact dataset, joins two dimensions, aggregates by business key, and writes a curated table. The first instinct is to double compute because several tasks spill and the job misses its runtime target.
An architecture review discovers four issues: the scan includes data outside the required date range; the job carries unused columns into the shuffle; one dimension is small enough to be a broadcast candidate; and a default business key creates a heavily skewed partition.
The important outcome is not a fabricated percentage improvement. The point is the order of operations. If those four issues are corrected first, the workload presented to the cluster is fundamentally different. Only after measuring that new plan does cluster sizing become meaningful.
Table 1. Illustrative diagnostic sequence; not measured production results.
When a Spark job becomes expensive, I use a repeatable review rather than randomly changing configuration values.
First, I inspect the physical plan. I look for large scans, repeated exchanges, sort-merge joins, global sorts, and places where the estimated row counts appear inconsistent with reality.
Second, I inspect stage behavior in the Spark UI: task-duration spread, input size, shuffle read/write, spill, and whether a small number of tasks dominate the stage.
Third, I connect those observations back to the data architecture. Is the source layout compatible with the filters? Is the join grain correct? Are statistics current? Is skew inherent in the business data? Are we producing a file layout that will make tomorrow's job worse?
Only then do I tune resources. Dynamic resource allocation can help a Spark application request resources when demand rises and return them when they are no longer needed. Spark 4.0.4 supports dynamic allocation across its coarse-grained cluster managers, but the feature has setup requirements and is not a replacement for reducing unnecessary work.
The sequence matters: measure the plan, remove avoidable work, then size compute for the work that remains.
I think of Spark optimization as four layers.
Figure 1. Architecture-first Spark cost model (conceptual; no benchmark data).
Layer 1 — Data layout: file size, partitioning, clustering, column selection, and whether the storage organization matches common access patterns.
Layer 2 — Logical work: filters, joins, aggregations, deduplication, sorting, and the grain at which the pipeline operates.
Layer 3 — Physical execution: scan partitions, shuffle partitions, join strategy, AQE, skew handling, caching, spill, and task distribution.
Layer 4 — Infrastructure: executor count, cores, memory, autoscaling, instance shape, and cluster policy.
Teams often start at Layer 4 because it is the easiest layer to change. The largest opportunities, however, frequently sit above it. If Layer 1 or Layer 2 creates unnecessary work, Layer 4 merely pays to execute it faster.
For shared Spark platforms, performance should not depend entirely on individual engineers remembering every tuning technique.
I would standardize a few things: capture physical plans for important workloads; monitor shuffle volume and spill alongside runtime; flag extreme task-duration skew; maintain table statistics where the optimizer relies on them; define file-size and partitioning conventions for major data products; and review cost together with SLA compliance.
I would also avoid organization-wide “magic settings.” A shuffle partition count, broadcast threshold, executor size, or caching policy that helps one workload can hurt another. Platform defaults should provide safe starting points, while workload-level decisions should be justified by measurements.
That approach changes performance engineering from emergency tuning into architecture governance.
The most expensive Spark job is not necessarily the one with the largest cluster. It is the one where nobody can explain why the cluster is doing so much work.
Modern Spark gives engineers increasingly capable runtime optimization. AQE can coalesce shuffle partitions, adapt join strategies, and mitigate skew. Storage-partition joins can eliminate some shuffles when compatible data sources expose useful partitioning. Dynamic allocation can adjust resource usage as demand changes.
Those capabilities are valuable, but they work best when the underlying architecture gives Spark a reasonable problem to solve.
When a job slows down, I would resist the urge to begin with “How much more compute do we need?” Start with a different question: What work are we making Spark do, and which part of that work should not exist?
That question tends to produce cheaper systems, more predictable SLAs, and fixes that survive the next increase in data volume.
1. Apache Spark 4.0.4 — Performance Tuning. Used to verify file partitioning, shuffle partitions, statistics, join strategies, Adaptive Query Execution, skew handling, and Storage Partition Join behavior. https://spark.apache.org/docs/4.0.4/sql-performance-tuning.html
2. Apache Spark 4.0.4 — Tuning Guide. Used as foundational guidance for memory, serialization, garbage collection, and general Spark performance analysis. https://spark.apache.org/docs/4.0.4/tuning.html
3. Apache Spark 4.0.4 — Job Scheduling. Used to verify dynamic resource allocation behavior and its setup requirements. https://spark.apache.org/docs/4.0.4/job-scheduling.html
4. Apache Spark 4.0.4 — Configuration. Used to cross-check Spark configuration names and semantics. https://spark.apache.org/docs/4.0.4/configuration.html
A slow Spark job is often treated as a sizing problem: add executors, increase memory, raise shuffle partitions, and try again. That can hide the real issue. The expensive failures I see in Spark architectures usually come from unnecessary shuffles, skewed joins, poor file layout, stale or missing statistics, oversized scans, and pipelines designed without regard for the physical execution plan. Spark 4.0.4 can mitigate several of these problems through Adaptive Query Execution (AQE), but AQE is not a substitute for sound data layout and query design. This article presents an architecture-first way to diagnose Spark cost before reaching for more compute.
The Cluster Is Usually the First Suspect
When a Spark workload misses its SLA, the first conversation often starts with infrastructure. How many executors did we have? How much memory? Should we increase executor cores? Would a larger cluster finish faster?
Those are reasonable questions, but they are often asked too early. Spark performance is the result of a chain of architectural decisions: how data is stored, how it is partitioned, how much of it is scanned, where data moves across the network, how joins are planned, whether a few partitions dominate the stage, and only then how much compute is available to execute the plan.
That distinction matters in cloud environments because an inefficient plan can still finish successfully. The bill simply absorbs the inefficiency. A job that completes after adding more compute can look “fixed” even though the architecture is still forcing Spark to read, shuffle, sort, spill, or recompute far more data than necessary.
I prefer to treat Spark performance as an execution-plan problem first and a cluster-sizing problem second. Before increasing capacity, I want to know what work Spark is actually being asked to perform.
Where the Cost Actually Appears
A useful way to think about Spark cost is to follow the movement of data rather than the number of machines.
At a high level, a Spark SQL or DataFrame workload can become expensive in five places: input scanning, repartitioning and shuffle, joins and aggregations, memory pressure/spill, and output layout. These costs compound. A poor file layout can increase the scan. A wide transformation can then shuffle that larger dataset. A skewed join can make one or two tasks run much longer than the rest. The output can finally create another small-file problem for the next pipeline.
Apache Spark's own performance-tuning documentation reflects this architecture. It separately addresses file partitioning, statistics, join strategy, caching, shuffle partitioning, Adaptive Query Execution, skew handling, and storage-partition joins. That is a useful reminder: there is no single “Spark tuning” knob.
1. Start With the Scan, Not the Executor
Before looking at executor memory, I look at how much data the job reads and how that data is physically organized.
For file-based sources, Spark uses settings such as `spark.sql.files.maxPartitionBytes` when packing files into input partitions. In Spark 4.0.4, the documented default is 128 MB. Spark also accounts for the estimated cost of opening files through `spark.sql.files.openCostInBytes`. Those settings matter, but they cannot repair a dataset that is badly organized for its access patterns.
If a query needs one day of data but the storage layout forces it to inspect months of files, the first optimization target is not executor memory. It is the amount of work entering the plan.
The same principle applies to columns. Columnar formats such as Parquet allow Spark to avoid reading unnecessary columns when the query and source support column pruning. Selecting every column “just in case” increases I/O, serialization, memory pressure, and often the amount of data that later participates in a shuffle.
My first three questions are therefore simple: What fraction of the dataset is being scanned? Are filters being pushed as early as possible? Are we reading columns that never contribute to the result?
2. A Shuffle Is a Design Decision With a Price Tag
Shuffles are not inherently bad. Distributed joins and aggregations frequently require data to move between executors. The mistake is treating that movement as free.
Spark's `spark.sql.shuffle.partitions` controls the number of partitions used for shuffles in joins and aggregations; Spark 4.0.4 documents a default of 200. A fixed number is obviously not ideal for every workload. Two hundred partitions may be excessive for a small intermediate result and far too few for a very large one.
AQE helps here. Spark can coalesce post-shuffle partitions using runtime map-output statistics, reducing the number of tiny downstream tasks. Current Spark documentation notes that AQE is enabled by default and can adapt the physical plan while a query is executing.
But AQE does not mean engineers can stop thinking about shuffle. If the logical design repeatedly repartitions the same large dataset, performs avoidable global sorts, or joins at a grain much finer than the business requirement, adaptive execution is optimizing an unnecessarily expensive plan.
A practical review question is: for every `Exchange` in the physical plan, can I explain why the data must move? If the answer is no, I have found something worth investigating.
3. Data Skew Turns Parallelism Into a Long Tail
One of the easiest Spark problems to misdiagnose is skew.
A stage may contain hundreds of tasks, most of which finish quickly, while a handful run for much longer. Adding executors may do very little because the bottleneck is not a shortage of parallel task slots. Too much data has landed in a small number of partitions.
Spark AQE includes skew-join optimization for sort-merge joins. In Spark 4.0.4, `spark.sql.adaptive.skewJoin.enabled` is documented as enabled by default. Spark can identify skewed shuffle partitions and split them into smaller pieces, replicating the other join side when needed.
That is valuable, but I still want to understand why the skew exists. Common causes include hot business keys, null or default values concentrated under one key, and joining at an inappropriate grain. Sometimes the correct solution is architectural: filter earlier, separate exceptional keys, pre-aggregate, or redesign the join. Runtime mitigation and data-model correction solve different parts of the problem.
4. Join Strategy Depends on Information Spark Can Trust
Join performance is another place where architecture and metadata meet.
Spark can automatically broadcast a sufficiently small relation, avoiding a shuffle of the larger side. Spark 4.0.4 documents `spark.sql.autoBroadcastJoinThreshold` with a 10 MB default. Developers can also provide join hints such as `BROADCAST`, `MERGE`, `SHUFFLE_HASH`, and `SHUFFLE_REPLICATE_NL`, although Spark does not guarantee that every requested strategy is applicable to every join.
The more important point is that planning depends on statistics. Spark's documentation explicitly warns that missing or inaccurate statistics can hinder its ability to choose an optimal plan. Catalog statistics can be updated with `ANALYZE TABLE`, while `EXPLAIN COST` or `DataFrame.explain(mode="cost")` can expose estimates used by the optimizer.
This is why I do not treat metadata maintenance as housekeeping. Statistics are part of query performance. A platform that carefully tunes compute but allows optimizer statistics to become unreliable is making execution decisions with incomplete information.
5. More Memory Can Hide the Wrong Problem
Memory tuning has a place, especially when workloads genuinely require large in-memory working sets. But an out-of-memory error or heavy spill is not automatically proof that the executor is too small.
Spark's tuning guide emphasizes the interaction among memory consumption, serialization, garbage collection, and the cost of Java objects. From an architecture perspective, I first ask why the working set is so large. Are we caching data that is used once? Did a join multiply rows unexpectedly? Are we aggregating before filtering? Are partitions too large because the upstream partitioning strategy is poor?
Increasing memory without answering those questions can convert a deterministic design problem into a more expensive design problem that happens less often.
An Illustrative Example: The Job That 'Needed' a Bigger Cluster
Consider a hypothetical daily transformation. The numbers below are illustrative; they are not benchmark results from a production system.
A pipeline reads a large fact dataset, joins two dimensions, aggregates by business key, and writes a curated table. The first instinct is to double compute because several tasks spill and the job misses its runtime target.
An architecture review discovers four issues: the scan includes data outside the required date range; the job carries unused columns into the shuffle; one dimension is small enough to be a broadcast candidate; and a default business key creates a heavily skewed partition.
The important outcome is not a fabricated percentage improvement. The point is the order of operations. If those four issues are corrected first, the workload presented to the cluster is fundamentally different. Only after measuring that new plan does cluster sizing become meaningful.
| Observation | Likely architectural question | What to test before scaling |
|---|---|---|
| Large scan | Is pruning effective? | Narrow date/partition filters and selected columns |
| Heavy shuffle | Is repartitioning necessary? | Inspect Exchange nodes and join/aggregation grain |
| Long-tail tasks | Is the data skewed? | Compare partition/task sizes and hot keys |
| Spill | Why is the working set large? | Reduce input/intermediate data before adding memory |
Table 1. Illustrative diagnostic sequence; not measured production results.
Architecture Review: What I Check Before Scaling Up
When a Spark job becomes expensive, I use a repeatable review rather than randomly changing configuration values.
First, I inspect the physical plan. I look for large scans, repeated exchanges, sort-merge joins, global sorts, and places where the estimated row counts appear inconsistent with reality.
Second, I inspect stage behavior in the Spark UI: task-duration spread, input size, shuffle read/write, spill, and whether a small number of tasks dominate the stage.
Third, I connect those observations back to the data architecture. Is the source layout compatible with the filters? Is the join grain correct? Are statistics current? Is skew inherent in the business data? Are we producing a file layout that will make tomorrow's job worse?
Only then do I tune resources. Dynamic resource allocation can help a Spark application request resources when demand rises and return them when they are no longer needed. Spark 4.0.4 supports dynamic allocation across its coarse-grained cluster managers, but the feature has setup requirements and is not a replacement for reducing unnecessary work.
The sequence matters: measure the plan, remove avoidable work, then size compute for the work that remains.
A Better Mental Model for Spark Cost
I think of Spark optimization as four layers.
| Layer 1 | Data layout | Reduce unnecessary input |
|---|---|---|
| Layer 2 | Logical work | Reduce unnecessary transformations |
| Layer 3 | Physical execution | Execute required work efficiently |
| Layer 4 | Infrastructure | Size resources for remaining work |
Figure 1. Architecture-first Spark cost model (conceptual; no benchmark data).
Layer 1 — Data layout: file size, partitioning, clustering, column selection, and whether the storage organization matches common access patterns.
Layer 2 — Logical work: filters, joins, aggregations, deduplication, sorting, and the grain at which the pipeline operates.
Layer 3 — Physical execution: scan partitions, shuffle partitions, join strategy, AQE, skew handling, caching, spill, and task distribution.
Layer 4 — Infrastructure: executor count, cores, memory, autoscaling, instance shape, and cluster policy.
Teams often start at Layer 4 because it is the easiest layer to change. The largest opportunities, however, frequently sit above it. If Layer 1 or Layer 2 creates unnecessary work, Layer 4 merely pays to execute it faster.
What I Would Put Into a Production Standard
For shared Spark platforms, performance should not depend entirely on individual engineers remembering every tuning technique.
I would standardize a few things: capture physical plans for important workloads; monitor shuffle volume and spill alongside runtime; flag extreme task-duration skew; maintain table statistics where the optimizer relies on them; define file-size and partitioning conventions for major data products; and review cost together with SLA compliance.
I would also avoid organization-wide “magic settings.” A shuffle partition count, broadcast threshold, executor size, or caching policy that helps one workload can hurt another. Platform defaults should provide safe starting points, while workload-level decisions should be justified by measurements.
That approach changes performance engineering from emergency tuning into architecture governance.
The Hidden Cost Is Architectural Debt
The most expensive Spark job is not necessarily the one with the largest cluster. It is the one where nobody can explain why the cluster is doing so much work.
Modern Spark gives engineers increasingly capable runtime optimization. AQE can coalesce shuffle partitions, adapt join strategies, and mitigate skew. Storage-partition joins can eliminate some shuffles when compatible data sources expose useful partitioning. Dynamic allocation can adjust resource usage as demand changes.
Those capabilities are valuable, but they work best when the underlying architecture gives Spark a reasonable problem to solve.
When a job slows down, I would resist the urge to begin with “How much more compute do we need?” Start with a different question: What work are we making Spark do, and which part of that work should not exist?
That question tends to produce cheaper systems, more predictable SLAs, and fixes that survive the next increase in data volume.
Further Reading and Technical References
1. Apache Spark 4.0.4 — Performance Tuning. Used to verify file partitioning, shuffle partitions, statistics, join strategies, Adaptive Query Execution, skew handling, and Storage Partition Join behavior. https://spark.apache.org/docs/4.0.4/sql-performance-tuning.html
2. Apache Spark 4.0.4 — Tuning Guide. Used as foundational guidance for memory, serialization, garbage collection, and general Spark performance analysis. https://spark.apache.org/docs/4.0.4/tuning.html
3. Apache Spark 4.0.4 — Job Scheduling. Used to verify dynamic resource allocation behavior and its setup requirements. https://spark.apache.org/docs/4.0.4/job-scheduling.html
4. Apache Spark 4.0.4 — Configuration. Used to cross-check Spark configuration names and semantics. https://spark.apache.org/docs/4.0.4/configuration.html