Diagnosing Reduce-Side Skew in a Daily Order-Customer Join
A 40-node Hadoop cluster runs a MapReduce job that joins 800 million order records (about 6 TB) with a 2 million row customer dimension table (about 400 MB). The job uses 200 reducers. Counters show that 35% of order records carry the default customer key 'UNKNOWN'. Three reducers finish in about 3.5 hours while the other 197 finish in 15 minutes. The team already tried increasing reducers from 100 to 200 and saw no improvement; it plans to raise them to 400. The job also writes 120,000 small output files because every reducer writes one file per partition. The cluster has 40 nodes, each with 12 disks and 128 GB RAM. The deadline for the daily load is 06:00, and the job currently ends at 07:10. The operations lead asks for a fix that does not require buying hardware.
1. Diagnose the long-tail reducers. Explain why adding reducers will not solve the problem.
Read the worked answer
The symptom is classic reduce-side skew. The join keys are partitioned by hash, and 35% of 800 million records - about 280 million records - share the single key 'UNKNOWN'. All records with the same key must go to the same reducer, because the reducer needs to group that key. One reducer therefore receives hundreds of millions of records while most reducers receive a normal share.
Adding reducers changes the number of hash buckets but does not split a single key. Whether there are 200 or 400 reducers, every 'UNKNOWN' record still maps to exactly one bucket. More reducers can even make the tail relatively worse by increasing the number of small reducers and output files. The real fixes are to remove the hot key from the join path, avoid the reduce-side join entirely, or split the hot key across multiple reducers with a deliberate two-stage pattern. Simply tuning reducer count treats a data-distribution problem as a parallelism problem.
2. Propose two redesign options and compare their trade-offs.
Read the worked answer
Option A - map-side join with DistributedCache. The customer dimension is about 400 MB, which fits comfortably in the memory available on each node. Broadcast the dimension table to mappers and perform the join in the map phase. There is no reduce-side shuffle of 6 TB, so the hot key cannot concentrate on one reducer. Trade-off: the dimension must fit in memory and be refreshed for each run; a slowly changing dimension needs a versioned snapshot.
Option B - filter and salt the hot key. Write 'UNKNOWN' orders to a side output for separate handling, then salt the remaining hot key with a random suffix so records distribute across reducers. A second job aggregates the salted partial results. Trade-off: two jobs and more code complexity, but it works when the dimension is too large to broadcast.
A third, simpler improvement is to treat 'UNKNOWN' as invalid data and exclude it from the join, reporting it through counters. Compare options by operational cost: Option A is fastest for this data size, while Option B is more general. Either is better than raising reducers to 400, which leaves the skew untouched.
3. Explain how the 120,000 small output files affect HDFS and propose a fix.
Read the worked answer
Every file, directory, and block is an object in the NameNode namespace and consumes roughly 150 bytes of NameNode heap. 120,000 files is not catastrophic by itself, but it is per run; if the job runs daily and outputs are retained, the namespace and heap pressure accumulate. Small files also hurt downstream Hive or MapReduce reads because each file becomes a separate task or block split, increasing scheduling overhead and reducing throughput.
Fix at the output stage rather than only cleaning up later. First, reduce the number of output partitions where semantically valid; 200 reducers each writing one file per partition is the direct cause. Second, run a compaction step that merges small files into target-size files (for example 128–256 MB) before publishing the table. Third, enable output compression so merged files are smaller on disk. Fourth, use a CombineFileInputFormat-style reader for downstream jobs so existing small files are grouped into fewer splits. Finally, set a retention policy so intermediate output is deleted after the reporting table is built. These steps address both NameNode heap and read performance.