Data Skew, Stragglers, and Broadcast vs Shuffle Joins

Data skew and the straggler

Parallelism only helps if the work is spread evenly. Data skew is when one key has vastly more rows than the others: a NULL that swallows a third of the rows, one mega-customer, one hot product. In a group-by or join, all the rows for a key go to one task, so the task holding the giant key runs far longer than the rest. The whole stage cannot finish until that one straggler finishes, so a job that should take a minute takes twenty, and that one executor may run out of memory. You already surfaced a hot key like this in Level 5's join fan-out lesson, where a GROUP BY key COUNT(*) exposed the one customer_id with far more rows than the rest; that same count is the first skew diagnostic here.

You spot skew by comparing the slowest task to a typical task in its stage. A task running many times a typical task's duration is the tell. (Spark's Adaptive Query Execution, on by default since Spark 3.2, even detects and splits skewed partitions automatically, when a partition is both more than five times the median partition size and larger than 256 MB.) The beginner fixes are to salt the hot key (append a random suffix so it spreads across tasks), filter out the junk key (a NULL you do not need), or let AQE handle it.

One honest caveat about the exercise below: it divides each task's duration by the stage average, which is a one-line proxy you can write in SQL, but the straggler itself pulls that average up, so the ratio understates the skew. The slow task here runs about 2.6x the average, yet against a typical (non-straggler) task it is closer to 6x, which is what would trip AQE's five-times-the-median rule. When you can, compare to the median or a typical task, not the mean.

Broadcast join vs shuffle join

The most common join question a junior gets is "you are joining a huge fact table to a tiny lookup table, how do you avoid a shuffle." The answer is a broadcast join.

Table

If one side is small enough to broadcast, you avoid shuffling the huge table entirely. Spark auto-broadcasts under 10 MB by default.

join kind	when	data movement
broadcast join	one side is small (under ~10 MB)	copy the small table to every executor; the big table never moves
shuffle (sort-merge) join	both sides are large	shuffle both sides so same-key rows meet

• If one side is small, the engine copies that whole table to every executor and each executor joins its slice of the big table locally. The huge table never moves. No shuffle.
• If both sides are large, you cannot broadcast, so the engine shuffles both sides so matching keys meet. This is a sort-merge join, and it is the expensive default for big-to-big joins.

Spark decides automatically: any table under spark.sql.autoBroadcastJoinThreshold (default 10 MB, or 10485760 bytes) is broadcast. That default is really 10 MiB (10,485,760 bytes), so the byte comparison, not a rounded "10 MB", is what actually decides. Setting the threshold to -1 disables it. Knowing that number, and that it is the small side that gets broadcast, is a clean junior answer.

Reading join and skew data with SQL

The exercises compute both: a per-task straggler ratio (each task's duration over its stage average) that surfaces the skewed task, and, for a set of join inputs, which dimension tables are small enough to broadcast under the 10 MB threshold.

Common mistake: trying to broadcast a table that is not actually small. Only the small side of a join broadcasts, and broadcasting a large table copies it to every executor and runs them out of memory, so check the size against the threshold first.

Interview nuance: "huge fact table joined to a tiny dimension, avoid the shuffle" wants "broadcast the small table so the big one never moves." "One task is 10x slower than the rest" wants "data skew: a hot key overloaded one task; salt it, filter the null, or let AQE split it."

On a real platform this differs. Redshift co-locates join keys with a distribution key so matching rows already share a node; Snowflake runs elastic virtual warehouses over shared micro-partitions. The broadcast-vs-shuffle and skew reasoning is the same idea those systems express differently. The task_metrics and join_inputs tables here let you compute the decision the optimizer makes.