Failure Modes and Production Debugging

Broadcast Join Memory Exhaustion: The most common production failure with broadcast joins is memory exhaustion from stale statistics. The optimizer decides to broadcast a table it believes is 2 gigabytes, but the actual size is 15 gigabytes due to recent data growth or incorrect statistics. Each executor tries to build a 15 gigabyte in memory hash table, exceeding available memory. You see out of memory errors, long garbage collection pauses (5 to 30 seconds), and executor crashes. This creates cascading failures. When one executor crashes, its tasks are rescheduled on other executors, increasing load there. If those executors are also near memory limits, they crash too. A single bad broadcast can bring down an entire job or even destabilize a shared cluster affecting other users. Recovery requires aborting the query, updating statistics, and rerunning, which costs tens of minutes to hours of delay. Shuffle Hash Join Skew Stragglers: Data skew creates extreme tail latency. Imagine a join on user_id where one user (perhaps a test account or bot) has 10 million events while typical users have 50 to 100 events. Hash partitioning sends all 10 million rows to a single partition. That partition might be 800 gigabytes while others are 2 gigabytes. Building an 800 gigabyte hash table fails with out of memory, or if it spills to disk, takes 2 to 3 hours while other tasks finish in 5 minutes. The symptom is that a stage shows 199 out of 200 tasks completed in minutes, but one task runs for hours. The entire job waits on that single straggler. You cannot simply increase parallelism because the problem is one logical partition, not lack of parallelism. Solutions include salting the join key (adding a random suffix to split hot keys), using sort merge instead, or filtering out anomalous keys if they are not needed for the business logic. Sort Merge Join Disk Spill Amplification: Sort merge join is robust to memory pressure but vulnerable to disk I/O bottlenecks. When a partition exceeds memory during sort, the engine spills sorted runs to disk. If local disk is slow (spinning disks or network attached storage with high latency), or if disk is shared with other workloads causing contention, spill operations can extend task time by 5 to 10 times. For example, sorting a 12 gigabyte partition in memory might take 90 seconds. If that same partition needs to spill because only 4 gigabytes of memory is available, the engine sorts in 4 gigabyte chunks, spills three runs to disk, and then merges them. On a congested disk with 100 millisecond per operation latency, this can take 10 to 15 minutes. If multiple concurrent tasks are spilling, disk becomes the bottleneck and total job time balloons. Cartesian Join Explosion: A missing or incorrect join predicate can turn into an accidental Cartesian join. If you join two tables with 10 million rows each without a proper join key, the engine attempts to produce 100 trillion output rows. In practice, the job writes terabytes of shuffle data, saturates network and disk, and runs for hours or days before failing or being killed. The symptom is explosive shuffle write metrics. A normal join might shuffle 2 terabytes. A Cartesian join will shuffle 100 to 1000 times more. Engines try to detect this by checking if a join stage produces far more output rows than input rows, but by the time you notice, significant resources are wasted. Prevention is better: always validate join predicates and look for suspicious query plans. Debugging Approach: When a distributed join fails or runs slowly, start with metrics. Check shuffle read and write bytes to identify if data movement is the bottleneck. Look at task duration histograms: if p50 is 2 minutes but p99 is 60 minutes, you have skew. Check memory metrics for garbage collection time and heap usage: if garbage collection time is over 10 percent of task time, memory pressure is likely. For broadcast failures, confirm table sizes match statistics. For skew, inspect key distribution with a GROUP BY query on the join key. For sort merge slowness, check disk I/O wait time in task metrics. The right fix depends on which resource is the bottleneck: add memory, increase shuffle partitions to reduce per partition size, use broadcast hints, or rewrite the query to filter data earlier.