Stop Blaming Executor Memory: The Real Reasons Your Spark Jobs Are Slow

After a decade of building and debugging large-scale data pipelines across financial services, payments processing, and analytics platforms, I can tell you that almost every slow Spark job I've investigated had the same root cause — and it wasn't the one the team thought it was. 

The default response when a Spark job is slow is to add more executor memory, increase the number of executors, or bump spark.sql.shuffle.partitions. Sometimes that helps. Usually it doesn't. What I've found, consistently, is that the real problems are structural — a join strategy mismatch that silently multiplies your intermediate dataset by ten times, a single slow task on a degraded node that holds an entire stage hostage, or a decrypt chain that re-reads source data six times when it only needed to read it once.

This article is organized around five patterns I keep seeing across teams. Each one looks different on the surface but traces back to a misunderstanding of how Spark actually executes your code. For each pattern, I'll describe what it looks like, when it bites you, the failure mode, and how to fix it. 

Pattern 1: The OR Join That Quietly Multiplies Your Data

What It Looks Like

A join condition with an OR clause. Usually introduced when a business requirement adds a secondary matching rule — match on primary card number, or if the transaction is a virtual card transaction, match on the underlying physical PAN. The SQL looks reasonable. The engineer tests it on a sample, and it returns the right rows.

When It Bites You

At scale. With 100 million transaction rows and 50 million account rows, this query starts running for hours. The output size is also wrong — much larger than expected before DISTINCT trims it down.

The Failure Mode

Spark cannot use a hash join or sort-merge join when the join condition contains OR. It falls back to BroadcastNestedLoopJoin — for every row in the left table, scan every row in the right table. That's O(n x m). On real datasets, this produces an intermediate result in the hundreds of GB before any downstream filter runs. I've watched a pipeline that should produce 8 GB of output generate 400 GB of intermediate data because of exactly this pattern, taking a 20-minute job to 4 hours. 

You can verify this in 30 seconds: run df.explain(formatted) and look for BroadcastNestedLoopJoin in the physical plan. If you see it on a join involving any table over a few million rows, it's almost certainly unintentional.

The Fix

Split the join into two equi-join legs and UNION ALL the results:

-- Leg 1: primary match (equi-join — uses SortMergeJoin or BroadcastHashJoin)
SELECT txn.*, acct.*
FROM transactions txn
JOIN accounts acct ON txn.card_number = acct.card_number

UNION ALL

-- Leg 2: fallback match, filtered scope only
SELECT txn.*, acct.*
FROM transactions txn
JOIN accounts acct ON txn.fpan = acct.physical_pan
WHERE txn.transaction_type = 'VIRTUAL'

Each leg is a proper equi-join. Apply DISTINCT at the end to deduplicate rows that matched both. The performance difference is routinely an order of magnitude. 

Pattern 2: The Straggler Task That Nobody Notices Until It's Too Late

What It Looks Like

A stage that should take 10 minutes takes 3 hours. The Spark UI shows nearly all tasks completed quickly. One or two tasks are still running with a disproportionately long duration.

When It Bites You

Jobs running on shared YARN or cloud infrastructure where any node can have a bad disk, a noisy neighbor, or degraded network throughput. Also common in stages that call external services per partition — one slow API response can cause a single partition's tasks to take 100x longer than the others.

The Failure Mode

A stage doesn't complete until the last task completes. Not the median. Not p95. The absolute last one. If 2,200 tasks finish in under 2 minutes and one takes 3 hours and 7 minutes, the stage takes 3 hours and 7 minutes. The other 2,199 executors sit idle. This is the straggler problem, and it's distinct from data skew.

The diagnostic: in the Stage detail view, check the task duration distribution. If MAX is dramatically higher than p99, that's a straggler (hardware or external service issue). If p75 is already much higher than p50, that's skew (data distribution issue). They require different fixes, and many teams treat them identically. 

The Fix

For stragglers caused by degraded infrastructure, enable Spark speculation:

spark.speculation=true
spark.speculation.multiplier=3    # task must be 3x slower than median
spark.speculation.quantile=0.9    # wait for 90% completion before speculating

Speculation re-launches slow tasks on a different executor and uses whichever copy finishes first. The caveat: don't use this on stages that write to non-idempotent sinks. For read-heavy or compute-heavy stages — including external decryption calls — it's often the single most impactful config change you can make.

Pattern 3: The df.rdd Decrypt Chain That Recomputes Everything Six Times

What It Looks Like

A pipeline that calls an external encryption or decryption service per record, implemented as a series of df.rdd.mapPartitions() calls, one per column that needs to be processed.

When It Bites You

When you have multiple columns to decrypt. Each .rdd call creates a new computation starting from the original DataFrame — Spark re-reads from source, re-executes all upstream joins and filters, and then runs the decryption for that column. With six columns to decrypt, you're doing that six times.

The Failure Mode

Two distinct sub-problems compound each other. First, going to RDD bypasses Catalyst entirely — no predicate pushdown, no column pruning, no Tungsten execution. Second, without a persist checkpoint before the chain, every decrypt call lineages all the way back to the source. I've seen this double the runtime of a job compared to the same pipeline with a single persist() before the decrypt chain.

On top of that, the external call latency per partition is dominated by the number of HTTP round trips, not the payload size. Cutting your batch size in half doubles your request count and roughly doubles your wall-clock time for that stage. Most teams set an initial batch size and never revisit it.

The Fix

Two changes, applied together:

• Persist the input DataFrame before starting the decrypt chain. This means the join and filter logic runs once, and each decrypt call reads from the cached result.
• Increase the batch size for external calls. Test at several sizes — going from 20,000 to 40,000 records per batch often cuts stage time by 30-50% with no change to correctness.

val base = rawDf.filter(...).join(key1, ...).persist(StorageLevel.MEMORY_AND_DISK)

val step1 = decryptColumn(base, secret1)   // reads from cache
val step2 = decryptColumn(step1, secret2)  // reads from cache  
val step3 = decryptColumn(step2, secret3)  // reads from cache

Without persist, step2 re-executes everything step1 did from source. With persist, each step reads from the in-memory result of the previous. 

Pattern 4: The shuffle.partitions Setting That Nobody Updates

What It Looks Like

A job that works fine in staging — where data volumes are 10% of production — but runs slowly, spills to disk, or produces thousands of tiny output files in production. 

When It Bites You

When the default spark.sql.shuffle.partitions=200 is left unchanged. 200 partitions made sense as a default for medium datasets but is almost always wrong at production scale — either too few (huge partitions, memory pressure) or too many (tiny partitions, scheduling overhead, small files problem).

The Failure Mode

Too few partitions means each executor handles a disproportionately large chunk of data. With 200 partitions on a 1 TB shuffle, each partition is 5 GB. That will spill to disk. Too many partitions means thousands of 1 MB tasks — the scheduling overhead becomes significant, and your output has thousands of tiny files that hurt downstream readers.

With Adaptive Query Execution (AQE) enabled in Spark 3.2+, this problem largely manages itself. AQE merges small post-shuffle partitions automatically and can handle modest skew. But AQE can't help if it's disabled, and it can't fix the upstream causes of extreme skew.

The Fix

Enable AQE if you're on Spark 3.2+:

spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.adaptive.skewJoin.enabled=true

If you need to set shuffle.partitions manually, target roughly 128-256 MB per partition post-shuffle. For a 500 GB shuffle, that means 2,000-4,000 partitions. Set it high and let AQE coalesce down — that's cheaper than setting it low and getting OOM errors. 

Pattern 5: The Incremental Job That Degrades Silently Over Time

What It Looks Like

A job that runs in 15 minutes when first deployed and runs in 4 hours six months later. No code changes. No obvious data quality issues. The team attributes it to data growth.

When It Bites You

When the job fails a few times in a row, and the recovery accumulates multiple windows' worth of data. Or when the watermark logic was designed for small windows but nobody anticipated that the underlying join tables would grow significantly. 

The Failure Mode

Two separate causes, often confused. First, if the watermark is a single timestamp and the job has been failing, recovery runs can accumulate large backlogs. A job that normally processes 2 hours of data may need to process 48 hours on first successful recovery, with no change to the resource configuration. Second, growth in reference data (like an accounts table or lookup table used in a join) increases the size of every run regardless of whether the incremental input grew. I've seen a 30-minute job become a 3-hour job purely because the accounts table grew from 10 million rows to 80 million rows over 18 months, while the OR join condition (see Pattern 1) meant that growth was amplified into the intermediate result.

The Fix

Two design principles that pay off over the lifetime of the pipeline:

• Track processed partitions explicitly rather than using a single timestamp watermark. This makes recovery granular — you can replay specific missing partitions without re-processing everything after them.
• Add a fast-path no-op check before initializing the full Spark session. Check whether any new partitions exist first. A 5-second check that exits early is much better than a 2-minute executor startup that discovers there's nothing to process.

For the reference table growth problem: if your lookup table grows significantly, revisit whether it can be broadcast (small enough to fit in executor memory) or whether the join itself needs to be redesigned.

Quick Diagnostic Reference

Use this table to map what you observe in the Spark UI to the likely pattern and first action to take:

The Common Thread

Every pattern here traces back to the same underlying issue: Spark is executing something different from what the engineer intended. The OR join was intended as a flexible matching rule; Spark turned it into a nested loop. The decrypt chain was intended as six independent transformations; Spark turned it into six full re-reads of source data. The incremental job was intended to process one window of data; without proper watermark design, it occasionally processes twelve.

The Spark UI has everything you need to see this — task distribution, input and output sizes, physical plans, spill metrics. Most teams open it when something breaks and close it once they find the obvious error. Opening it proactively, forming a hypothesis, and then confirming or refuting it in the metrics is the practice that separates engineers who consistently improve pipeline performance from those who add executor memory and hope for the best.

The mistake isn't choosing the wrong config. It's not understanding what Spark is actually doing with your code.