A Single Reducer Owns 55% of a 2 TB Join
An e-commerce batch platform runs a nightly MapReduce join on a 200-node YARN cluster (20 vcores, 128 GB RAM per node). Queue etl uses the Capacity Scheduler with 30% capacity and 50% maximum; the other two queues are about 40% idle at night. The job reads a 2 TB fact table (customer_id, order_id, amount) and joins a 40 MB customer dimension, then aggregates by customer. It runs 3,000 map tasks and 400 reduce tasks. One reducer processes 55% of all records and takes 5 of the job's 6 hours; about 30% of fact rows have customer_id = NULL. Containers are 12 GB / 2 vcores with mapreduce.map.java.opts=-Xmx8192m; many tasks die with 'Container killed by YARN for exceeding memory limits. 12.5 GB of 12 GB physical memory used'. ApplicationMaster containers are 1.5 GB. The team's first reaction is to raise the reducer count to 2,000 and double container memory.
1. Explain why neither proposed change fixes the straggler, and interpret the 12.5 GB versus 12 GB memory kill.
Read the worked answer
The straggler is a partitioning problem, not a parallelism problem. The default hash partitioner sends every row with the same key to the same reduce task. With 30% of 2 TB of rows carrying customer_id = NULL, roughly 600 GB of records hash to one partition, so one reducer receives 55% of the data no matter how many reducers exist. Going from 400 to 2,000 reducers leaves that single NULL partition untouched while adding 1,600 container startups, more shuffle connections and 2,000 small output files. The job still takes about 5 hours.
The memory kill is a container-sizing problem. mapreduce.map.java.opts=-Xmx8192m sets the Java heap, but YARN accounts for the whole process RSS: metaspace, code cache, thread stacks, direct byte buffers used by shuffle and compression codecs, and native libraries sit outside the heap. 8 GB heap plus ~4 GB of non-heap is entirely consistent with 12.5 GB RSS against a 12 GB physical limit, so YARN kills the container. The standard guidance is Xmx ~ 0.75-0.8 x mapreduce.map.memory.mb. Doubling the container to 24 GB without touching Xmx does not help; you would set Xmx to about 18 GB and keep the ratio. Also check yarn.nodemanager.pmem-check-enabled (this is the physical check that fired) and yarn.nodemanager.vmem-check-enabled, which on JDK 8 often produces spurious virtual-memory kills and is commonly disabled.
So the correct sequence is: fix the key distribution, then size containers by RSS budget, then re-measure.
2. Design the join and aggregation fix, including how to handle NULL keys and how to size reducers afterwards.
Read the worked answer
Two changes, applied together:
Eliminate the reduce-side join for the dimension. The customer dimension is 40 MB on disk and far smaller after decompression into a hash table. Use a map-side join: Hive MapJoin with
hive.auto.convert.join.noconditionaltask.sizeraised to 64 MB, or DistributedCache in raw MapReduce. All 3,000 mappers build the hash table and no fact rows are shuffled for the join at all. Broadcast the dimension only if executor/container headroom allows; 40 MB per task is acceptable, but do not broadcast the 400 MB dimension the same way.Handle the NULL key explicitly. Three workable patterns: filter NULL keys into a separate small aggregation that computes the single 'unknown customer' group and union it back; or replace NULL with a sentinel and salt it with a random 0-9 suffix, aggregate partially by (sentinel, salt) in stage one and combine the ten partial results in stage two; or use
hive.groupby.skewindata=true, which adds a redistribution stage at the cost of an extra MR job. The separate-aggregation pattern is cheapest here because the NULL group collapses to one row.
Sizing after the fix: the reducers now only aggregate, so shuffle volume is the distinct customer count, not 2 TB. With 8 GB heap and 12 GB containers, a reducer sorting 2-3 GB of shuffled data is safe; 400-800 reducers gives 2.5-5 GB each if the full 2 TB had to be shuffled, so 400 is already conservative once the join is removed. Keep mapreduce.reduce.shuffle.input.buffer.percent at 0.7 of heap, and set mapreduce.reduce.memory.mb to 1.25-1.33 x Xmx. Validate with counters: reduce-shuffle-bytes should drop by an order of magnitude.
3. The job only gets about 1,200 vcores while two other queues sit idle. Explain the scheduler arithmetic and list the realistic ways to give this job more of the idle cluster.
Read the worked answer
The cluster has 200 nodes x 20 vcores = 4,000 vcores. Queue etl capacity is 30% = 1,200 vcores and maximum is 50% = 2,000 vcores. The job asks for 3,000 mappers x 2 vcores = 6,000 vcores, so it needs at least three waves even at the maximum. It is currently pinned near its 1,200-vcore capacity guarantee while the other queues hold their 40% share.
Idle capacity in other queues is not automatically available. The Capacity Scheduler guarantees each queue its percentage; another queue's unused capacity can only be borrowed if etl's maximum capacity permits it, and even then the borrowable amount is limited by what the scheduler can assign without violating guarantees. Raising maximum-capacity to 70% would let etl grow into idle capacity, but the idle queues must give it up. If they will not release it, preemption is the mechanism: enable the scheduler monitor (yarn.resourcemanager.scheduler.monitor.enable=true) and the proportional/intra-queue preemption policy so containers in over-allocated queues are reclaimed for etl.
A second, frequently missed cap is user-limit-factor (default 1). Even when the queue has headroom, a single submitting user is limited to 1x the queue capacity. If one service account submits all nightly jobs, raise the factor or split submissions across accounts.
Practical options, in order: raise etl maximum capacity and enable intra-queue preemption for the night window; negotiate a shared queue for this job; or use node labels to give the ETL workload a dedicated partition. Note that ApplicationMaster memory (1.5 GB) is not the bottleneck and changing it will not add vcores.