The Spark Deep Dive: Intermediate

The driver is the JVM that runs your main. It holds the SparkSession, turns your transformations into a logical plan, optimizes that plan, cuts it into stages at every shuffle, and hands individual tasks to executors. It also keeps the bookkeeping: which task ran where, which shuffle block lives on which executor, which partitions are cached. Executors are JVMs the cluster manager launches on worker nodes. Each has a fixed core count and a fixed heap, runs 1 task per core at a time, caches partitions in its memory, and serves the shuffle files it wrote to any executor that asks. In most deployments the cluster manager is YARN or Kubernetes, and its only job is to allocate those containers and tear them down; it never sees your data or your plan. A 2 TB job and a 2 MB job look identical to it.

Nothing runs until the write on the last line, because every transformation is lazy: read only extends the plan, and so do filter and groupBy. The write is an action, and 1 action produces 1 job. The driver then walks the plan backwards looking for shuffles. filter is narrow: each output partition depends on exactly 1 input partition, so it stays in the same stage as the read. groupBy is wide: the rows for a given title_id are spread across every input partition and must be moved to one place, so the driver cuts a stage boundary there. The job has 2 stages. Stage 1 reads and filters, with 1 task per input split; at the default spark.sql.files.maxPartitionBytes of 128 MB, 1 TB of Parquet becomes about 8,192 tasks. Stage 2 sums per title, with 1 task per shuffle partition; at the default spark.sql.shuffle.partitions of 200 it runs 200 tasks however much data arrived.

At the stage boundary every stage 1 task hashes each row's title_id into 1 of 200 buckets, sorts its output by bucket, and writes 1 file with an index marking where each bucket begins to its executor's local disk. Every stage 2 task then opens a connection to each executor that wrote a segment for its bucket and pulls those bytes across the network. With 8,192 upstream tasks and 200 downstream tasks that is up to 1,638,400 fetches, and each one costs a disk read and a network transfer before the bytes are deserialized. The filter before the shuffle is the reason the job is affordable: if premium viewers produce 10% of the rows, the shuffle moves 100 GB instead of 1 TB. Each question later in this lesson comes down to how many bytes cross that boundary and how evenly they spread over the 200 tasks.

Check 2 more rows of that table on this job. Stage 1 wrote 210 GB of shuffle from a 2 TB scan, about 10%, which says the premium filter ran before the shuffle and column pruning worked; had it read 1.9 TB, the SQL tab's plan would show the filter above the join rather than inside the scan, and moving it cuts the shuffle 10 times. And the Jobs list showed 1 job, so nothing was recomputed; a script that calls count() for a log line and then write() runs the whole lineage twice and appears as 2 jobs with identical stage 1 rows.

The join question hands you 2 table sizes and asks which strategy you want and why. 'Orders is 2 TB, products is 40 MB, the join on product_id takes 50 minutes, what would you do?' The answer the interviewer wants names the strategy the planner picked by default, the strategy you would force, and the bytes that each one moves across the network. Spark's 3 join strategies are 3 answers to the same problem: rows with equal keys live on different executors and must meet somewhere.

Sort-merge join shuffles both sides by the join key into the same number of partitions. It then sorts each partition and walks the 2 sorted lists together. Every byte of both tables crosses the network once, and the sort needs memory or it spills. The planner picks it whenever both sides exceed the broadcast threshold. Broadcast hash join leaves the large side where it is. The driver collects the small side, ships a copy to every executor, and each task builds a hash table from that copy and probes it with its own partition of the large side. The large side moves zero bytes; the small side moves once per executor. Shuffle hash join is the middle path: shuffle both sides, then build a hash table from the smaller partition instead of sorting. It saves the sort but needs the smaller side's partition to fit in memory, and the planner prefers sort-merge over it unless spark.sql.join.preferSortMergeJoin is turned off.

Orders to customers flips the arithmetic. Broadcasting 30 GB means the driver materializes 30 GB in its heap, then 48 executors each receive and hold 30 GB, with 1.4 TB crossing the network and every executor spending 30 GB of memory on a hash table before a task runs. Sort-merge moves 2 TB plus 30 GB once and holds 1 partition at a time. Sort-merge wins, and the detail worth adding is why the broadcast fails in practice: the driver's collect of the small side is bounded by spark.driver.maxResultSize, and Spark refuses any broadcast over 8 GB. The join cost here is the shuffle of orders, and if orders joins customers every night, the way to remove that shuffle is bucketing.

Write both tables bucketed by customer_id into the same number of buckets, say 1,024, and sorted within each bucket. Bucket 17 of orders and bucket 17 of customers then hold the same set of customer_id values, because both hash the key the same way into 1,024 buckets, so the planner reads matching buckets into the same task and runs the merge with no exchange in the plan. The shuffle still happens, once, when each table is written; every nightly join afterwards reads 2 TB and shuffles none of it. The conditions are strict: same key and same bucket count on both sides, recorded in the catalog so the planner can see them. A bucketed table rewritten by a job that forgot the clause loses the property silently, and the next night's plan shows the exchange again.

Set aside 1 core and 1 GB per node for the operating system and the cluster manager's agent. That leaves 15 cores and 63 GB. The long-standing guidance caps an executor at 5 cores, because file system client throughput per JVM flattens past around 5 concurrent readers and because a heap large enough to feed 15 tasks pauses for garbage collection long enough to show up as task time. Dividing 15 cores by 5 gives 3 executors per node; 63 GB divided by 3 gives 21 GB per executor container. Spark reserves spark.executor.memoryOverhead outside the heap, 10% of executor memory or 384 MB, whichever is larger, for off-heap buffers and the Python worker if you use one, so request a 19 GB heap and the container lands at 21 GB. Across 16 nodes that is 48 executors and 240 task slots.

A 19 GB heap gives a task far less than 19 GB. Spark holds back 300 MB, then applies spark.memory.fraction, 0.6 by default, to the rest: (19 GB - 300 MB) x 0.6 = 11.2 GB of unified memory. Execution and storage share that pool. The executor's 5 concurrent tasks share it too, so each one can count on about 2.2 GB for sorting and hashing before it spills, and less if cached partitions occupy the storage half. That's the number the slow-job section was missing: a shuffle partition near 1.2 GB, once deserialized into objects with a sort buffer on top, spills against a 2.2 GB ceiling, and a 128 MB to 256 MB partition does not.

Both rounding decisions in that table have a reason. The shuffle partition count is rounded to a multiple of the slot count, because 500 partitions on 240 slots run 2 full waves and then 20 tasks on a cluster that is 92% idle; 480 fills every slot in both waves. And stage 1 is quoted as 35 waves rather than 34.1 because the fractional wave runs at the tail at the full cost of a wave. At 20 seconds a task the stage takes under 12 minutes, which fits the 30 minute window with room for the join; if the measured task time comes back at 60 seconds, the same arithmetic says 35 minutes and the answer changes to more nodes or a smaller input. Set the count with spark.sql.shuffle.partitions, or set it high and let adaptive execution coalesce small partitions toward its 64 MB advisory size, which absorbs a filter rate that changes from day to day.

Every shuffle in this lesson hashed a key into 200 partitions and assumed the keys spread out. Skew is what happens when they do not: 1 title_id for a live event carries 40% of a night's playback rows, so the partition that key hashes into holds 40% of the bytes, and the task that owns it runs 40% of the join alone on 1 core while 199 cores finish and wait. The interviewer's version is usually 'the stage has sat at 199 of 200 tasks for 5 hours, what is going on', and the answer needs 3 parts: how you confirm it, which of 3 fixes you reach for, and what each costs.

The task summary from the slow-job section already said it: max duration 5 hours 33 minutes against a median of 25 seconds, and 1 task reading 96 GB of shuffle against a median of 0.6 GB. The second number matters more than the first, because a slow task with a normal read size points at a bad node or garbage collection rather than skew. Then find the key, because the fix depends on how many hot keys there are and whether one of them is NULL:

Adaptive query execution is on by default since Spark 3.2, and its skew join handling has 2 conditions, both default: a shuffle partition counts as skewed when it is larger than 5 times the median partition and larger than 256 MB. Spark then splits the 96 GB partition into pieces near its advisory size and pairs each piece with a copy of the matching partition from the other side, so the join runs as many tasks instead of 1. It costs nothing to enable and works for sort-merge joins, which makes it the first answer. It also did not fire last night, and the reason is the shape of stage 3: the aggregate on title_id reuses the join's partitioning, so splitting the hot partition would force a second shuffle, and Spark skips the split rather than add one. Setting spark.sql.adaptive.forceOptimizeSkewedJoin accepts the extra shuffle. An aggregation on its own is safe: a sum or count runs as a partial aggregate before the exchange and a final one after it, so a hot key's rows collapse to 1 row per map task before they move. Salting covers the rest: joins adaptive execution will not split, aggregations whose partials do not shrink, and Spark versions before 3.0.

The large side gets a random integer from 0 to 63 appended to its key, so the hot key hashes into 64 partitions of 1.5 GB instead of 1 of 96 GB, and 1.5 GB sits under the 2.2 GB task budget from the sizing section. The small side is exploded 64 times so that every salted key still finds its match, and that is the price: a 30 GB catalog becomes 1.9 TB of shuffle, which is why the catalog gets projected to the 3 columns the join needs first, at which point it weighs 2 GB and the explosion costs 128 GB. Pick the salt count from the UI: the hot partition's bytes divided by the per-task budget, rounded up to a power of 2. A plain sum or count on the hot key needs no salt. A distinct count over a huge group does, because its partials do not shrink: salt by a hash of the counted column, count distinct per (title_id, salt), then sum the 64 partials.

With 1 or 2 hot keys known in advance, split the join. Rows whose title_id is in the hot set join the 2 matching catalog rows through a broadcast, so the large side never shuffles for them; every other row takes the sort-merge path, now with no partition above 2% of the total; union the 2 results. It costs 2 scans of the large side instead of 1, which pays off when the hot keys stay hot from day to day and loses when they change nightly, so it sits behind the adaptive fix. The NULL key is its own case. In an inner join it matches nothing, and Spark infers IsNotNull on the join key and drops those 10% before the shuffle, which is why no NULL partition shows in the task summary. In a left or outer join the rows must survive and all hash to 1 partition, so filter them out before the join and union them back with empty catalog columns.