Why More Executors Stopped Helping: Skew, Memory, Files, and the Cost Curve

A nightly job runs in forty minutes on twenty executors. The team doubles the cluster and it runs in thirty-eight. The bill doubles. Somebody suggests doubling again. The example is invented, and the shape of it is not: past some point, adding machines stops changing the runtime, because the runtime is being set by something that more machines do not affect.

This article is about what actually sets that limit — skew, file layout, memory, serialization, and resource allocation — and about the places where making a job faster and making it cheaper pull in opposite directions. Configuration defaults are from Spark 4.2.0 documentation, checked in September 2026, and they change between versions.

The one sentence that explains most slow jobs

A stage finishes when its last task finishes. Everything else is commentary. Suppose twenty-one tasks, twenty of a minute each and one of twenty minutes: if there are enough slots to start them all together, the stage takes twenty minutes and most of the cluster is idle for nineteen of them — while being billed.

The “start together” part is an assumption worth checking rather than a rule. With twenty slots the long task may be queued behind a short one and begin a minute in, making the stage twenty-one minutes; how many tasks run at once depends on available cores and what each task needs, not on the executor count alone. That is also why the stage is bounded by the last completion time rather than by the longest duration.

This is why the first diagnostic is never “how big is the cluster” but “what is the distribution of task durations within the slow stage.” The Spark UI and the event log give the median and the maximum. A gap of an order of magnitude says one task is doing something different — and the next question is what. Compare the per-task input and shuffle bytes and the record counts: if the slow task processed far more data, uneven distribution is the leading hypothesis. Hardware does not redistribute the data — a bigger machine still gives that partition to one task — though it can reduce what the imbalance costs, since extra memory removes spill and faster storage shortens the read. So “more hardware” is a way to make a skewed job cheaper to tolerate, not a way to remove the skew.

If the volumes match and one task is still slow, look at what else differs: output rows produced, GC time, spill, time spent waiting on shuffle fetches, and the host. Equal input bytes do not guarantee equal work — a user function or an amplifying join can make one partition far more expensive — so rule that out before blaming the machine.

Where the work really is equal and one host is slow, replacing it helps, and speculative execution helps by running a second attempt somewhere else. A host that recurs across slow tasks raises that hypothesis; it does not confirm it, because the data can be what recurs — a co-located cache, a partition read from storage nearest that node, or simply the executor that happens to hold the large partition. Separate the two before acting: check whether the same host is slow on a task with a different partition, and whether the same partition is slow when it runs elsewhere.

Speculative execution gives you that second observation cheaply, and it is an observation rather than a verdict. A second attempt elsewhere that finishes quickly is evidence for the host — or for a transient condition that has since passed. One that is also slow is evidence against the host, and consistent with an expensive partition or with something both attempts share: a slow shuffle source, a throttled object store or external service, a saturated network path, an oversubscribed cluster. Read the attempts side by side — input and output rows, GC, spill, fetch wait, and the host metrics for each — rather than treating the comparison as a switch with two positions.

Skew: when one partition holds the job hostage

Data skew means the rows are not spread evenly across partitions. The usual causes are recognizable: one customer or one merchant accounting for a large share of the rows, a join key that is null for many records so that all nulls hash together, a date partition that includes a backfill, or a hash of a low-cardinality column.

Adaptive execution handles part of this automatically. Its skew-join optimization is enabled by default and splits a partition that is both more than a factor (default 5.0) larger than the median and above a size threshold (default 256 MB). What it does not cover is worth knowing: skew in an aggregation rather than a join, skew that arrives before the shuffle, and skew whose cause is a single enormous row group in the source files.

The manual techniques are ordinary and not in the documentation: filter or handle null keys explicitly rather than letting them collide; split a hot key into several by adding a random suffix and aggregating twice, a technique usually called salting; or broadcast the small side so the skewed side never shuffles at all. See partitioning, shuffles, and data skew for worked versions.

Partitions and files: two numbers that decide task count

Setting	Default	What it decides
spark.sql.files.maxPartitionBytes	128 MB	How much file data one reading task gets
spark.sql.files.openCostInBytes	4 MB	The assumed cost of opening a file — a 10 KB file is costed as if it were 4 MB
spark.sql.shuffle.partitions	200	How many partitions a shuffle produces, regardless of the data’s size
spark.sql.adaptive.advisoryPartitionSizeInBytes	64 MB	The target size when adaptive execution merges small post-shuffle partitions — but only when coalescePartitions.parallelismFirst is set to false. Left at its default of true, Spark “ignores the target size … and only respect the minimum partition size” of 1 MB to maximize parallelism, which the documentation calls “the default case”

The open-cost default explains the small-file problem in one line: thousands of tiny files are not cheap because they are small; the engine treats each as a fixed minimum of work, and the scheduler pays per task regardless. File compaction is usually the highest-value change available in a pipeline that writes many small outputs.

The fixed default of 200 shuffle partitions is the other classic. It is right for some volume of data and wrong for every other: too few for a large join, absurdly many for a small aggregation. Adaptive execution’s coalescing softens the second case by merging small partitions after the shuffle, which is one of the strongest arguments for leaving it enabled.

Memory: what executors do when they run out

Spark divides executor heap deliberately. spark.memory.fraction defaults to 0.6 — the fraction of heap minus 300 MB used for execution and storage — and the documentation is unusually direct that “leaving this at the default value is recommended,” noting that lowering it makes spills and cache eviction more frequent. Within that region, spark.memory.storageFraction (default 0.5) is the part immune to eviction; raising it leaves less working memory so that “tasks may spill to disk more often.”

A spill is what happens when an operation’s working set does not fit: the data is written to local disk and read back. It is a safety mechanism, not an error, and it turns a memory-speed operation into a disk-speed one. A stage whose tasks report large spill volumes is telling you to give it more memory, less data per task, or a different plan — in that order of ease, not of preference.

Two more memory facts save time when a container is killed rather than slow. Off-heap overhead is real and provisioned separately: spark.executor.memoryOverhead defaults to 10% of executor memory (40% for non-JVM Kubernetes jobs), and the container’s total is the sum of heap, overhead, off-heap, and PySpark memory. And caching is not free storage: persisting a DataFrame that is used once costs memory and buys nothing.

Serialization and the JVM’s own overheads

Anything that crosses a machine boundary — every shuffle, every broadcast, every cached block stored in serialized form — is serialized first. Spark’s tuning guide calls this “often… the first thing you should tune,” and the default is deliberately conservative: spark.serializer is Java serialization, which “works with any Serializable Java object but is quite slow,” with Kryo recommended when speed matters and described as “often as much as 10x” faster and more compact. Note what this setting governs: the serialization of JVM objects. DataFrame and SQL operations work on Spark’s own internal representation through generated encoders, so a query written entirely in SQL or the DataFrame API is largely unaffected by spark.serializer. The setting earns its keep on the RDD path: shuffling and caching RDD records, and cached blocks held in serialized form. Two nearby things it does not govern, because they are easy to lump in. Closures are serialized by a separate Java serializer regardless of this setting, so a closure capturing a large object is fixed by capturing less or broadcasting it rather than by changing the serializer. And a Dataset of a custom type goes through an encoder, so the lever there is which encoder you supply — a different decision from the global setting.

There is a trap in adopting Kryo halfway. Registration is not required by default, and without it Kryo writes the full class name with every object, which the documentation notes “can cause significant performance overhead.” Switching the serializer without registering the classes buys a fraction of the improvement people expect.

Related, and often larger: garbage collection. The tuning guide notes that GC cost is proportional to the number of Java objects, so representations with fewer, larger objects cost less. This is part of why built-in operations on structured data outperform equivalent logic in user code that materializes many small objects per row.

Layout beats tuning

Before any configuration change, the cheapest performance work is arranging so that less data is read at all: columnar storage so that unused columns are not read, partitioning on the column that queries filter by so that partition pruning can skip whole directories, and file statistics that let predicate pushdown skip row groups. Work that is never read is never shuffled, never serialized, and never spilled.

Resources: how many executors, and of what size

Given a fixed budget of cores and memory, the same resources can be packaged as many small executors or fewer large ones, and neither is universally right. Many small executors give finer scheduling granularity and smaller failure units, but multiply the per-executor overhead and increase the number of shuffle connections. Fewer large executors amortize overhead and keep more data local, but lose a larger unit of work when one dies, and can suffer longer GC pauses. Spark’s tuning guide gives the one durable anchor: aim for “2-3 tasks per CPU core in your cluster.”

Dynamic resource allocation lets an application return executors it is not using and ask for them again later. It is disabled by default, and enabling it requires one of several shuffle arrangements — an external shuffle service, shuffle tracking, or decommissioning — because, as the documentation explains, if an executor is removed before its shuffle output has been consumed, that output “must be recomputed unnecessarily.” One more documented caveat matters in practice: executors holding cached data are not removed by default, so a long-running application with cached DataFrames may release far less than expected.

Where speed and cost disagree

Choice	Faster	Cheaper
Adding executors past the point of diminishing returns	Slightly	No — cost rises roughly with resources while time barely moves
Caching a reused dataset	Yes, for the second and later reads	Only if the saved recomputation exceeds the memory held
Dynamic allocation on a bursty workload	No — waiting for executors adds latency	Capacity is returned, which is not the same as a smaller bill: on a fixed or reserved cluster it frees contention for other work, and the charge only falls if the underlying nodes scale down and you are billed for what runs
Interruptible (spot) capacity	No — an interruption costs a rerun	Often substantially, for work that tolerates restarts
Writing many small files to parallelize the write	Yes, for the writer	No — every downstream reader pays the open cost

The spot row deserves its own sentence, because it is where the two goals diverge most sharply. AWS describes Spot Instances as spare capacity at a steep discount in exchange for returning it when EC2 needs it back, and states plainly that “it is always possible that your Spot Instance might be interrupted.” For a fault-tolerant batch job with checkpointing, that is an excellent trade. For a job with a hard deadline and expensive restart, the discount is a bet you may lose exactly when it matters.

The useful discipline is to state which one you are optimizing before changing anything, because the same measurement supports opposite conclusions. A job that finishes in twelve minutes instead of twenty, on a cluster twice the size, got faster and more expensive; whether that was a good trade depends on a deadline someone should be able to name.

A diagnosis order

1. Find the slow stage, then compare median and maximum task duration inside it.
2. If they diverge, gather evidence before concluding. Bytes and records per task, output rows, GC time, spill, shuffle fetch wait, and which host ran it. Uneven input points at skew; equal input with one slow host points at the machine. Neither is proof on its own — equal bytes can still mean unequal work if a UDF or a join amplifies some rows, and a skewed task can still be helped by more memory if it was spilling.
3. If they agree, look at input and shuffle volume per task, plus spill. Too much data per task is a partitioning question; spill is a memory question.
4. Read the plan for avoidable work — unpushed filters, an unexpected shuffle, a join strategy that does not match the data.
5. Only then change resources, and measure both the runtime and the cost, not just the runtime.
6. Say which of the two you were optimizing, and record it with the change. Meeting a deadline and lowering a bill pull in opposite directions — more executors buy time and cost money — so a change made for one reason gets judged later by the other unless the intent is written down.

For the operational side of this loop — what to record, and how to keep a tuned job tuned — see optimizing Spark jobs in production.

References

Spark 4.2.0 documentation and AWS documentation were checked in September 2026. Defaults change between versions; confirm them for the version you run.

• Apache Spark 4.2.0 documentation: Tuning Spark, Configuration, Performance Tuning, and Job Scheduling
• Amazon Web Services, Spot Instance interruptions (Amazon EC2 User Guide)