Spark Performance and Cost Factors

In an illustrative scenario, a nightly job runs in forty minutes on twenty executors. The team doubles the cluster and it runs in thirty-eight. On executor-minute pricing the bill nearly doubles — forty executors for thirty-eight minutes against twenty for forty — and on a fixed cluster it may not change at all. Somebody suggests doubling again. Past some point, adding machines stops changing the runtime, because the runtime is being set by something that more machines do not affect.

Skew, file layout, memory, serialization, and allocation can limit runtime in different ways. The configuration defaults here apply to Spark 4.2.0.

The last task sets a stage’s completion time

A stage finishes when its last task finishes. Most of the rest follows from that. 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, once the slow stage has been located within the job’s total time — queueing, planning, and external I/O can dominate too — is not “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, at the price of a duplicate attempt, 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.

Manual techniques change how the data is distributed, and each has a condition that keeps the result correct. Handle null keys explicitly rather than letting them collide — but drop them only where the query’s meaning allows it, as in an inner equality join where they could match nothing; in an outer join, a null-safe join, or a grouped aggregate they are part of the answer and have to be set aside and preserved. Split a hot key into several by adding a random suffix and aggregating twice, a technique usually called salting — valid only for aggregates whose partial states merge: an average is carried as a sum and a count and combined at the end rather than averaged twice, and a distinct count needs a mergeable representation. Or broadcast the small side so the skewed side is not shuffled for that join — which removes that join’s shuffle, not the shuffles of the operations before it. See partitioning, shuffles, and data skew for worked versions.

How file layout and partition settings affect task count

Setting	Default	What it decides
spark.sql.files.maxPartitionBytes	128 MB	The byte size the file-source planner targets per scan partition — a planning estimate, not the decoded size in memory, and a file that cannot be split is not divided by it
spark.sql.files.openCostInBytes	4 MB	The planning estimate for opening a file — a 10 KB file is costed as if it were 4 MB when files are packed into scan partitions; it is not an I/O charge, and several small files can share one task
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; each costs an open, its metadata, and scheduling attention that a larger file amortizes, whether or not it gets a task of its own. 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 one combination of data volume, available parallelism, and operator cost, and wrong for the others: 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 and the operator can spill: 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; state that cannot be spilled still fails. A stage whose tasks report large spill volumes is telling you to give it more memory, less data per task, or a different plan — which of them depends on whether every task spills or one does.

Two more memory facts save time when a container is killed rather than slow. Off-heap overhead is real and provisioned separately: when spark.executor.memoryOverhead is not set, YARN and Kubernetes derive it as the larger of 10% of executor memory (40% for non-JVM Kubernetes jobs) and a 384 MiB minimum — so a 1 GiB heap gets 384 MiB of overhead, not 102 MiB; the container’s total is heap plus overhead plus off-heap, and PySpark memory is a separate addition only when spark.executor.pyspark.memory is set, since otherwise Python processes share the overhead budget. 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 this will be 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

Often the cheapest performance work, weighed against the cost of rewriting the data, 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, decommissioning, or a shuffle plugin backed by reliable storage — 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	When it helps	What it costs, and when it fails
Adding executors	While the stage still has parallelizable tasks waiting for slots	Past that point, cost rises with resources while time barely moves
Caching a reused dataset	When the recomputation it saves costs more than materializing and re-reading it — not automatically on the second read	Memory held for its lifetime, and eviction or a disk-backed level when it does not fit
Dynamic allocation on a bursty workload	When work arrives in bursts and there is warm capacity to reclaim between them	Scaling up from little adds latency while executors are requested; 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	When the work tolerates losing an executor — a task rerun, or a job restart where checkpoints are absent	Often substantially cheaper; an interruption costs whatever must be redone, from one task to the whole job
Writing many small files to parallelize the write	When the writer is the bottleneck and the file count stays reasonable	Every downstream reader pays the open cost, and too many tiny tasks and files slow the writer too

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 — on executor-minute pricing — about 1.2 times 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

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