Apache Spark Performance Tuning

The reasoning

The sort-merge version hash-partitions both sides on the join key and sorts each partition before merging: two Exchanges and two Sorts. The broadcast version ships the small side to every executor and joins in place, leaving only the aggregation's Exchange. Both return identical rows, which the harness asserts, so this is a pure cost reduction.

The responsibility is that a hint overrides the optimizer's size check permanently. The dimension is small today; the hint will still be there when it is 800 MB, collecting all of it to the driver and shipping a copy to every executor. The failure arrives as a broadcastTimeout or a driver OOM long after anyone remembers writing the line.

So the better default is usually to let AQE make the same decision from the actual shuffled size at runtime, which gets the win without the hard-coded assumption. Reach for the explicit hint when the estimate is genuinely wrong — after a chain of filters where Catalyst's idea of the size has drifted — and leave a comment saying what the size was when you wrote it.

The reasoning

The join-first version shuffles every order row, widened with customer columns, and then aggregates. The aggregate-first version reduces orders to one row per customer before anything is joined, so the join and everything after it operate on a much smaller set. On real data that ratio is where the saving is.

Catalyst will not make this rewrite for you. It pushes filters and prunes columns freely, and it does not move an aggregation through a join in the general case, because that transformation is only valid under conditions it usually cannot prove. This is one of the optimisations that is genuinely yours to write, which is why it is worth having as a reflex.

The validity condition is worth being able to state, because an interviewer will push on it: the pre-aggregation has to be at a grain the join preserves. Summing per customer and then joining a customer dimension is safe because the join is many-to-one on that key. If the join could fan out — a dimension with duplicate keys, or a join on a coarser key — pre-aggregating changes the answer, and the fact that the harness asserts identical rows here is exactly the check that would catch it.

The reasoning

The plan shows the effect in the scan's ReadSchema: only the projected columns are read, which on a wide table is a large saving in a columnar format because the other columns are physically elsewhere and never fetched. Narrower rows then move through any shuffle downstream.

Catalyst's column pruning already does this when it can see the whole query — it walks back from the final projection and reads only what is referenced. So for a straightforward pipeline the explicit select changes nothing, and writing it is documentation rather than optimisation.

Where it matters is where the optimizer's view is broken. A cache boundary fixes the schema at the point you cached, so caching a wide DataFrame and then selecting three columns caches all of them. A UDF taking a struct or the whole row forces those columns to be materialised. And writing an intermediate to storage ends the optimisation window entirely. In all three the explicit projection is the only thing that narrows the read — which is why 'select early' survives as advice even though the optimizer usually beats you to it.

The reasoning

The setting applies to every shuffle in the application, so it is tuned for the largest stage and wasteful everywhere else. A selective filter followed by an aggregation produces a handful of rows spread across 200 partitions — 200 tasks scheduled, 200 fetches issued, and almost no work done. Task overhead is milliseconds each and it adds up against a stage that should have taken one.

The historical fix was to set it per job, which meant either a compromise number or changing the config between stages. AQE replaces that: it looks at the actual bytes written by the shuffle and coalesces adjacent partitions into targets of a reasonable size, so a small result gets a small number of tasks without anyone choosing a number.

So the current advice is to leave it high enough for the largest stage and let AQE coalesce downward, rather than to tune it. It still matters in one direction: AQE coalesces but does not split, so a value far too low leaves partitions too large and there is nothing to correct it. Erring high with AQE on is the shape to recommend.

The reasoning

The condition is reuse across *actions*, not across lines of code. A DataFrame referenced once produces one job, and caching adds a write to storage memory that nothing will read. Worse, cached blocks occupy the same unified region execution needs, so an unnecessary cache can push a later stage into spilling.

It flips when two branches each trigger an action over shared, expensive work — and both halves matter. Shared but cheap is not worth it: re-reading a small Parquet file twice costs less than holding it in memory. Expensive but used once has nothing to amortise.

When it does apply, the mechanics matter as much as the decision. cache() is lazy, so follow it with count() to materialise every partition rather than letting the first branch populate it partially. Prefer MEMORY_AND_DISK, because MEMORY_ONLY silently drops partitions that do not fit and recomputes them while still reporting as cached. And unpersist when the last reader is done, because memory held after that is directly causing spill somewhere else.

The reasoning

Read parallelism comes from the source layout. Spark packs files into splits bounded by spark.sql.files.maxPartitionBytes, with openCostInBytes accounting for the per-file cost, so two big files give two tasks and forty thousand small ones give far too many. No configuration of executors changes it, which is why the fix for an under-parallel scan is always file layout or the split size.

Shuffle parallelism is the one people know: spark.sql.shuffle.partitions applies to every Exchange, and AQE then coalesces downward based on the bytes actually written. That is why the same job can show 200 partitions in the plan and eight tasks in the UI — the plan is provisional and the runtime number is the real one.

Write parallelism is whatever the final stage happened to have, which is the part that surprises people. If the last operation was a shuffle you get that count; if it was a narrow chain from the scan you get the scan's. That is how a job ends up writing 200 files for a tiny result, and why an explicit repartition or coalesce before the write is the only way to control file count deliberately.

The reasoning

Bucketing pre-partitions the data on disk by hashing the key into a fixed number of files. Two tables bucketed the same way on the same key have matching keys in corresponding buckets, so a sort-merge join between them needs no Exchange on either side. You pay the shuffle once at write time and every subsequent join is free of it.

That is a large win for a table joined the same way many times a day, and it comes with real constraints. The bucket count is fixed at write time and both sides must match — a mismatch means Spark shuffles anyway and you have paid the write cost for nothing. It requires a Spark-managed table via saveAsTable rather than a plain path write, which not every platform wants. And every writer to that table has to bucket, or the layout is broken for everyone.

So the decision is about stability and repetition. A dimension joined on the same surrogate key by twenty models is a strong candidate; a table whose access pattern is still changing is not, because rebucketing means rewriting it. It is also worth comparing against the alternatives first — if one side is broadcastable the shuffle is already gone, and if AQE's coalescing has made the shuffle cheap enough, the constraint may not be worth taking on.

The reasoning

Coalescing merges small post-shuffle partitions into targets of a reasonable size, which is what makes a high shuffle-partitions setting safe. It will not fire when there is no shuffle, and it only merges — it never splits, so a value set far too low stays too low.

Join conversion re-checks the actual shuffled size of each side and switches a sort-merge join to a broadcast when one fits. It cannot fire before the shuffle has run, so the first plan you print always shows the sort-merge; and it will not convert a join type broadcast cannot serve, such as a full outer join.

Skew join splitting divides oversized partitions across tasks and replicates the matching side. It applies to joins only — a skewed groupBy or window partition is untouched — and requires the partition to exceed both an absolute size threshold and a multiple of the median, so a stage where every partition is large triggers nothing. All three share one precondition worth stating: AQE re-plans at stage boundaries, so a job with no shuffle gets none of it.

The reasoning

MEMORY_AND_DISK keeps what fits in memory and spills the rest to local disk, so a cache larger than available storage memory still helps. That predictability is why it is the default for DataFrames and the right choice almost always.

MEMORY_ONLY drops partitions that do not fit and recomputes them on next access. Nothing warns you: the Storage tab shows the DataFrame cached at some percentage, and the missing partitions are quietly recomputed from lineage every time they are read. A cache that is 60% resident over an expensive lineage can be worse than no cache, because you pay the memory *and* the recomputation.

The serialised variants store each partition as one compact byte array rather than many live objects. That is several times smaller and dramatically reduces garbage collection pressure, at the cost of deserialising on every read — worth it for large caches or when GC time is already high. The _2 variants replicate each partition to a second executor, which is for expensive-to-recompute data on unreliable capacity and rarely worth double the memory otherwise.

The reasoning

cache is a performance shortcut that does not change the plan: the lineage behind the cached DataFrame is still there, so the logical plan keeps accumulating, Catalyst keeps re-analysing a larger tree, and if a cached block is lost it is recomputed through the whole chain.

checkpoint writes the DataFrame to reliable storage and replaces its lineage with a read of that location. The plan is cut, so planning stops getting slower and a lost partition is re-read rather than recomputed. The cost is a real write, and unlike cache it is not free to discard.

The symptom described — each action slower than the last with no more data — is usually plan growth rather than execution cost, and it shows up as driver time before any task is scheduled. Iterative algorithms and long chains of withColumn are the classic producers. localCheckpoint is the cheaper cousin that truncates lineage without the reliable write, which is fine when losing an executor is acceptable and not when it is not.

The reasoning

Spark's unified memory model gives execution and storage one region with a soft boundary. Execution can evict storage when it needs room, but only down to a protected floor — so cached blocks below that floor cannot be evicted and are held regardless of what execution needs.

Four caches held for the whole job means that floor is occupied for the whole job. Later stages get less execution memory than they otherwise would, exceed it, and spill. The spill shows up in a stage that has nothing to do with the caching, which is why the connection is rarely made — the symptom and the cause are far apart in both the code and the UI.

So unpersist when the last reader has run, and prefer to cache the narrowest thing that is actually reused rather than a wide intermediate. The Storage tab is the diagnostic: if it shows several large cached DataFrames while a late stage reports heavy spill, that is the pattern. And the first question is still whether each cache was earning its place, because a cache with one reader is pure cost.

The reasoning

Start from the partition target rather than the cluster. Aiming for partitions of roughly 128–256 MB, 500 GB gives on the order of 2,500 tasks. Then decide how many waves you are willing to run: with 5 cores per executor, 20 executors gives 100 concurrent tasks and about 25 waves, which is a reasonable shape for a batch job with a generous window.

Memory follows from what a task holds, not from total data. Each task needs room for its partition plus whatever the aggregation accumulates, and 5 tasks share an executor's pool — so a few gigabytes per core is a starting point, adjusted upward if the aggregation's state is large or the data expands a lot on decode.

The number to flag as uncertain is that expansion. 500 GB of Parquet is compressed and columnar; in memory it can be several times larger, and the ratio depends entirely on the data — high-cardinality strings expand far more than dictionary-encoded low-cardinality columns. Everything else in the estimate is arithmetic; that one is a measurement, and the honest answer is to size from a sample run rather than from a rule.

The reasoning

Dynamic allocation adds executors when tasks are queued and removes them when they have been idle. The saving is real for jobs with uneven shapes — a wide scan followed by a narrow aggregation, or a pipeline with long serial sections — because you stop paying for capacity nothing is using.

The prerequisite is the external shuffle service. Shuffle files live on the executor's local disk, so removing an executor that has written shuffle output makes that output unreachable and triggers FetchFailedException and stage recomputation. With the shuffle service those files are served by a separate process on the node and survive the executor's removal. Enabling dynamic allocation without it is a known way to make a job less reliable while trying to make it cheaper.

Where it does not help: a job that is busy end to end has nothing to release, so the overhead is pure. And acquisition is not instant — on Kubernetes or YARN under contention there is a real delay before new executors are running, so a spiky job can spend its time waiting for capacity. Setting a sensible minimum avoids scaling from zero on every stage boundary.

The reasoning

Bytes read is usually the largest term and the easiest to move. Check whether partition pruning is actually happening, whether the projection is narrow, and whether the source is thousands of small files. A job scanning a whole table because a predicate wrapped the partition column in a function is a one-line fix with an enormous return.

Shuffle volume is next, and the levers are the ones from the join and aggregation questions: broadcast where the side fits, aggregate before widening, project before shuffling. Each reduces bytes moved rather than time spent, which is what the bill responds to on most platforms.

Then wasted capacity: stages running with far fewer tasks than cores, executors idle through a long serial section, a cluster sized for the largest stage and paid for throughout. Dynamic allocation and AQE address much of this. Cluster size comes last because shrinking a cluster that is doing unnecessary work just makes the job take longer for the same total — the reductions above are what make a smaller cluster viable.

The reasoning

A job meeting its SLA with five hours of headroom is not a problem, and every change to it carries a risk of altering the answer. The tuning questions in this bank all have harness assertions that the rewrite returns identical rows precisely because that risk is real, and in production nobody is checking.

So the first question is which constraint is being violated. If it is cost, measure it — a job's share of the bill is often much smaller than its runtime suggests, and a 40-minute job on a small cluster can be cheaper than a 4-minute one on a large one. If it is a dependency waiting on it, the fix might be scheduling rather than speed. If it is 'it feels slow', that is not a constraint.

Where the effort genuinely belongs: jobs that fail or retry, jobs close enough to their window that a bad day breaches it, and jobs whose cost is a visible line item. Saying that plainly — with the numbers — is a better answer than producing a 20% improvement nobody needed, and it is the kind of judgement the question is actually testing for.

The reasoning

Format first. A columnar format lets a query read only the columns it needs and skip row groups using statistics, so a report touching three of forty columns reads a fraction of the bytes. That is a larger effect than any codec choice, and it is why converting text sources to Parquet at landing is close to always right.

The codec is where the parallelism decision hides. gzip compresses well and is not splittable, so a single 20 GB .csv.gz is exactly one task — no configuration, no cluster size and no repartition changes that, because the file cannot be divided. Snappy and zstd are splittable in the container formats that matter and are the reason Parquet defaults to snappy. zstd gives noticeably better compression at similar speed and is the sensible modern choice where it is supported.

Then file sizing, which is the third and most-neglected term: aim for files in the hundreds of megabytes. Too small and you pay listing and per-file overhead; too large and you lose the ability to parallelise and to skip. Getting format, codec and file size right at the write is a one-time decision that every subsequent read benefits from, which is a much better return than tuning the readers afterwards.

The reasoning

AQE's skew join handling costs a setting. It splits oversized partitions across tasks and replicates the matching side, which handles the common case with no code change and nothing to maintain. Confirming whether it is enabled and whether your partition actually clears its thresholds takes minutes.

Broadcasting is second because it removes the problem rather than managing it: with no shuffle there is no partitioning by key and therefore no hot partition. It is only available when one side genuinely fits, which is why it is second rather than first — but when it is available it is strictly better than anything downstream of it.

Salting is third because it works everywhere and costs the most: an N-fold copy of the small side, an unreadable synthetic join key, a salt factor to maintain, and a re-aggregation to strip it. Repartitioning is not on the list at all — every row with the hot key hashes to one partition however many partitions exist, so more partitions makes every other partition smaller and leaves the problem exactly where it was. Suggesting it is the clearest signal that someone has not understood why skew happens.

The reasoning

The most common is applying a function to the partition column: casting it, formatting it, or calling to_date on it means the engine can no longer match the predicate against the partition values, so PartitionFilters comes back empty and every directory is listed. Comparing the column to a literal on the *other* side of the function is the fix — transform the constant, never the column.

A UDF in the predicate stops pushdown entirely, because the optimizer cannot reason about it. So does filtering after something that ends the optimisation window: a cache fixes the plan at the point you cached, and an intermediate written to storage and read back starts a new query with no knowledge of the filter you are about to apply.

The way to catch it is to read the scan node rather than to reason about the query. PartitionFilters, PushedFilters and ReadSchema are three lines that say exactly what the reader will do, and comparing them against what you expected takes seconds. Making that part of review — or asserting on it in a test for a critical job — is how a regression like this gets found before the bill does.

The reasoning

Physical planning is where estimates matter. A side estimated at a thousand rows looks broadcastable; at forty million it is not, and the plan committed to the broadcast before anything ran — which is how a broadcastTimeout or a driver OOM happens. Join *order* in a multi-table query is chosen the same way, so a bad estimate can put the largest intermediate in the worst place.

The estimates come from statistics, and without ANALYZE TABLE those are file sizes. Compressed Parquet under-reports in-memory size systematically, and after a chain of filters and joins the propagated estimate can be off by orders of magnitude — selectivity of a predicate on a column with no histogram is essentially a guess.

Two fixes and they compose. Collect statistics on the tables and columns that matter, so the initial plan is made from measurements rather than file sizes. And enable AQE, which re-plans at each stage boundary using the bytes actually written — which is why it fixes this class of problem structurally rather than one query at a time. Comparing the provisional plan against the AQE final plan is the habit that shows you which estimates were wrong.

The reasoning

dropDuplicates keeps an arbitrary row per key. Which one survives depends on partitioning and task completion order, so changing shuffle partitions, adding a repartition, or enabling AQE coalescing can all change which duplicate wins. The job was always non-deterministic; the tuning change just moved the dice.

first() and last() in an aggregation without an ordering behave the same way, and monotonically_increasing_id encodes the partition id directly — so any change to partitioning changes the ids it generates. Using it as a stable key is the specific mistake, and it survives testing because a small dataset has stable partitioning.

The third category is genuine precision: changing a decimal to a double, or altering the order of a floating-point aggregation, produces slightly different sums because floating-point addition is not associative. A different partition count changes the order partial sums are combined in, so the total's last digits move. That is usually acceptable and it is worth knowing before someone reports it as a bug — and it is why the harness assertions in this bank compare exact rows.

The reasoning

The calculation: 1 TB is roughly 1,000,000 MB. Divide by the target size per partition. 1,000,000 / 200 ≈ 5,000. 1,000,000 / 128 ≈ 8,000. Both are defensible; a bare "5,000" without the target is not, because the number is meaningless without it.

Why that band. Too large and a single task has to hold its partition through a sort or an aggregation, which is where executor OOM and heavy spill come from. Too small and you pay fixed per-task overhead — scheduling, launch, and a small write — on work that takes milliseconds, and the driver becomes the bottleneck long before the executors do.

The second check is the cluster. Task count should be a comfortable multiple of total cores so every core stays fed and stragglers get absorbed — two to three times is the usual rule. 5,000 tasks on 126 cores is about 40 waves, which is fine. 5,000 tasks on 8 cores is not a partitioning decision, it is a cluster that is too small for the data.

And the honest modern caveat: with AQE on, this number is a **starting point rather than the final one**. `spark.sql.adaptive.coalescePartitions.enabled` lets Spark merge small post-shuffle partitions at runtime toward `spark.sql.adaptive.advisoryPartitionSizeInBytes`, so over-provisioning here is much cheaper than it used to be. Setting it too low is still a real mistake, because AQE can coalesce partitions but it cannot split one that is too big.

The reasoning

**Cores.** Reserve about 6 cores per node for the OS and the node manager, leaving 42. Executor cores go in the 4–6 band — below that you lose the benefit of shared broadcast data and JVM warmup, above that HDFS and cloud-storage throughput per executor stops scaling and GC pauses get long. Take 6: 42 / 6 = 7 executors per node, across 3 workers, so **21 executors and 126 cores**.

**Memory.** Reserve about 15% of the 384 GiB for the OS and daemons, leaving ~326 GiB, divided by 7 executors ≈ 46.6 GiB each. Split that into heap and off-heap overhead rather than handing it all to the heap: `spark.executor.memory=40G` and `spark.executor.memoryOverhead=6G` sums to 46 G, just inside the budget. Overhead is not optional — it is where shuffle buffers, the Python worker on PySpark, and native memory live, and omitting it is the direct cause of "Container killed by YARN for exceeding memory limits".

**Parallelism.** `spark.default.parallelism` at 2–3× total cores ≈ 250–380; 320 is a reasonable pick. `spark.sql.shuffle.partitions` comes from the data, not the cores: 1 TB at ~200 MB per partition ≈ 5,000. Turn AQE on so Spark can coalesce that down at runtime when a stage turns out smaller than the estimate.

The part worth saying out loud at the end is that this is a **starting configuration, not an answer**. You run it once, look at the Spark UI for spill, GC time and straggler tasks, and adjust. An interviewer who wanted a memorised table would not have given you the instance type.

The reasoning

An executor heap divides into reserved memory, user memory for your own objects, and the Spark region — `spark.memory.fraction`, 0.6 by default. Inside that region, `spark.memory.storageFraction` (0.5) marks how much storage is *guaranteed*, not how much it gets.

Execution memory is transient: it holds hash tables for aggregations and joins, sort buffers, and shuffle write buffers, and it is released when the task ends. Storage memory is what `cache()` and `persist()` fill, and it stays until eviction.

The important part is that **unified memory management makes the boundary soft**. If nothing is cached, execution can use the whole region. If execution needs memory and storage is over its guaranteed share, storage blocks are evicted to make room. But it does not go the other way: **execution memory cannot be evicted**, so storage cannot force execution to give anything back beyond the guaranteed line. That asymmetry exists because evicting a cached block costs a recomputation, while evicting a half-built hash table would mean failing the task.

What that means in practice: a cached DataFrame vanishing under a heavy shuffle is not a bug, it is the design. And on a 20 G executor the arithmetic is 20 × 0.6 = 12 G of Spark memory, of which 6 G is storage's guaranteed floor and the rest is contested.

The reasoning

**Listing.** Before anything runs, the driver enumerates the input. On object storage that is API calls, and a million files is a million of them — a serial-ish cost paid before the first row is read. On top of that the driver holds the resulting file metadata in memory, which is the first place a large listing shows up as driver pressure.

**Task creation.** A file smaller than the split size generally becomes its own task. A million small files is therefore roughly a million tasks, where 5,000 files of 200 MB would have been about 5,000.

**Scheduling.** The driver schedules every task individually and tracks its state. At a million tasks, the driver is the bottleneck and the executors sit waiting to be given work.

**Startup overhead.** Each task pays a fixed cost — launch, deserialise the closure, open the file, close it, report back. That cost is the same whether the file holds one kilobyte or two hundred megabytes. When the file is 1 MB, essentially all of the time is overhead.

Add executor utilisation to the list: tasks that finish in milliseconds leave cores idle between assignments, so a cluster that looks busy in the scheduler is doing almost no useful work. The good shape for this data is roughly 5,000 files of 200 MB — the same total bytes, two orders of magnitude less overhead.

The reasoning

**At write time**, the file count is the partition count — one file per partition per output directory. `coalesce(n)` reduces partitions without a shuffle and is the cheap fix when the data is already roughly balanced; `repartition(n)` costs a shuffle but distributes evenly, which matters when a `coalesce` would leave one enormous partition. If you are also partitioning by a column, `repartition(col)` before the write puts each value's rows on one task, which is what stops each partition directory getting one small file per input task.

**Cap the file size** rather than guessing at partition counts: `df.write.option("maxRecordsPerFile", 1000000)` splits any partition that would produce more, which keeps output files bounded even when the data volume moves between runs.

**For files that already exist**, compaction is a separate job: read a directory, repartition to a sensible count, write it back atomically. 1,000 files of 5 MB become 20 of 250 MB. On Delta or Iceberg this is a built-in operation (`OPTIMIZE`, `rewrite_data_files`) rather than something you hand-roll, and it handles the atomic swap for you.

The trap to avoid is `coalesce(1)`. It does produce one file, and it does it by collapsing the entire final stage to a single task — so the whole write becomes serial, and if that partition does not fit in one executor it fails outright. When someone asks for one file, they usually want *few* files.

The reasoning

Two requirements have to hold at once. The column must appear in the **WHERE clause of the queries that matter**, or pruning never triggers and you have paid for the layout without getting anything. And its **cardinality has to be low enough** that a partition is worth reading as a unit — the working target is partitions in the hundreds of megabytes.

`event_date` satisfies both for most analytical tables: everything filters on a time range, and a day of data is a real chunk. `country` can work as a second level when queries filter on it, though it skews badly if one country dominates. `user_id` satisfies the first and fails the second catastrophically: millions of directories, each with a handful of rows, which reproduces the small-file problem *and* makes listing the table slow enough that planning takes longer than the query.

The other half of the answer is that partitioning is not the only tool, and reaching for it reflexively is the mistake. **Partitioning is coarse physical separation; clustering, sorting or Z-ordering give you skipping within a partition without creating directories.** For a high-cardinality column you want to filter on, clustering is the right instrument — partition by date, cluster by user_id, and you get both.

Worth saying out loud: this decision is hard to reverse. Changing a partition scheme means rewriting the table, so a bad choice is lived with far longer than a bad query.