Unresponsive reduce tasks

[Kevin Weil <ke...@gmail.com>]

I have a script that I can run across a varying number of days' worth of
data.  Over 1 day, the script works perfectly, and runs in about 5 minutes
on my cluster.  Running it over 7 days, it generally doesn't finish one of
the reduce tasks -- 3 of the four finish properly, and the last one times
out after being unresponsive for 10 minutes, or even 30 minutes if I change
the hadoop mapred.task.timeout setting.  In each reduce task, the copy phase
takes about 20 minutes, the sort phase takes 10 seconds, and it is in the
final 1/3 where it eventually goes unresponsive.  When I log in to the
server it's running on during this final 1/3, I see one of the eight CPUs
pegged at 100% (the others at 1-2%) and hardly any disk or network IO.
Sending a kill -3 to the java process to make it dump its thread state, I
see the following curious thing:

"main" prio=10 tid=0x00000000450b2400 nid=0x26db runnable
[0x0000000040b02000..0x0000000040b02ed0]
   java.lang.Thread.State: RUNNABLE
    at java.util.ArrayList.size(ArrayList.java:177)
    at org.apache.pig.impl.plan.OperatorPlan.getRoots(OperatorPlan.java:75)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan.attachInput(PhysicalPlan.java:60)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach.attachInputToPlans(POForEach.java:358)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach.getNext(POForEach.java:164)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator.processInput(PhysicalOperator.java:227)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PODistinct.getNext(PODistinct.java:86)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator.getNext(PhysicalOperator.java:277)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POProject.processInputBag(POProject.java:364)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POProject.getNext(POProject.java:171)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc.processInput(POUserFunc.java:123)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc.getNext(POUserFunc.java:175)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc.getNext(POUserFunc.java:241)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach.processPlan(POForEach.java:237)
    at
org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach.getNext(POForEach.java:166)
    at
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.runPipeline(PigMapReduce.java:238)
    at
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.reduce(PigMapReduce.java:208)
    at
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce$Reduce.reduce(PigMapReduce.java:121)
    at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:318)
    at
org.apache.hadoop.mapred.TaskTracker$Child.main(TaskTracker.java:2207)

No matter how many times I send kill -3 to this, I see the same stack.  How
in the world are we possibly sitting in, or spending all of our time in,
java.util.ArrayList.size in OperatorPlan::getRoots?  When this reduce task
gets killed by hadoop after being unresponsive for 10 or 30 mins, it starts
up on another server, and running kill -3 shows the exact same stack.

I'm happy to provide more context if it will help.

Thanks,
Kevin