Jonathan, I am running it on Prod cluster in MR mode, not locally. I started to see the issue when input size grew. A few days ago I found a workaround of putting this property: mapred.child.java.opts=-Xmx1024m But I think this is a temporary solution and the job will fail when the input size will grow again.
Dmitriy, Thanks a lot for the investigation. I'll try it. -----Original Message----- From: Dmitriy Ryaboy [mailto:[email protected]] Sent: 22 ноября 2011 г. 2:21 To: [email protected] Subject: Re: java.lang.OutOfMemoryError when using TOP udf Ok so this: thirdLevelsTopVisitorsWithBots = FOREACH thirdLevelsByCategory { count = COUNT(thirdLevelsSummed); result = TOP( (int)(count * (double) ($THIRD_LEVELS_PERCENTAGE + $BOTS_PERCENTAGE) ), 3, thirdLevelsSummed); GENERATE FLATTEN(result); } requires "count" to be calculated before TOP can be applied. Since count can't be calculated until the reduce side, naturally, TOP can't start working on the map side (as it doesn't know its arguments yet). Try generating the counts * ($TLP + $BP) separately, joining them in (I am guessing you have no more than a few K categories -- in that case, you can do a replicated join), and then do group and TOP on. On Mon, Nov 21, 2011 at 1:53 PM, Jonathan Coveney <[email protected]> wrote: > You're right pablomar...hmm > > Ruslan: are you running this in mr mode on a cluster, or locally? > > I'm noticing this: > [2011-11-16 12:34:55] INFO (SpillableMemoryManager.java:154) - first > memory handler call- Usage threshold init = 175308800(171200K) used = > 373454552(364701K) committed = 524288000(512000K) max = > 524288000(512000K) > > It looks like your max memory is 512MB. I've had issues with bag > spilling with less than 1GB allocated (-Xmx1024mb). > > 2011/11/21 pablomar <[email protected]> > >> i might be wrong, but it seems the error comes from >> while(itr.hasNext()) >> not from the add to the queue >> so i don't think it is related to the number of elements in the queue >> ... maybe the field lenght? >> >> On 11/21/11, Jonathan Coveney <[email protected]> wrote: >> > Internally, TOP is using a priority queue. It tries to be smart >> > about pulling off excess elements, but if you ask it for enough >> > elements, it >> can >> > blow up, because the priority queue is going to have n elements, >> > where n >> is >> > the ranking you want. This is consistent with the stack trace, >> > which died on updateTop which is when elements are added to the priority >> > queue. >> > >> > Ruslan, how large are the limits you're setting? ie (int)(count * >> (double) >> > ($THIRD_LEVELS_PERCENTAGE + $BOTS_PERCENTAGE) ) >> > >> > As far as TOP's implementation, I imagine you could get around the >> > issue >> by >> > using a sorted data bag, but that might be much slower. hmm. >> > >> > 2011/11/21 Ruslan Al-fakikh <[email protected]> >> > >> >> Ok. Here it is: >> >> https://gist.github.com/1383266 >> >> >> >> -----Original Message----- >> >> From: Dmitriy Ryaboy [mailto:[email protected]] >> >> Sent: 21 ноября 2011 г. 20:32 >> >> To: [email protected] >> >> Subject: Re: java.lang.OutOfMemoryError when using TOP udf >> >> >> >> Ruslan, I think the mailing list is set to reject attachments -- >> >> can you post it as a github gist or something similar, and send a link? >> >> >> >> D >> >> >> >> On Mon, Nov 21, 2011 at 6:11 AM, Ruslan Al-Fakikh >> >> <[email protected]> wrote: >> >> > Hey Dmitriy, >> >> > >> >> > I attached the script. It is not a plain-pig script, because I >> >> > make some preprocessing before submitting it to cluster, but the >> >> > general idea of what I submit is clear. >> >> > >> >> > Thanks in advance! >> >> > >> >> > On Fri, Nov 18, 2011 at 12:07 AM, Dmitriy Ryaboy >> >> > <[email protected]> >> >> wrote: >> >> >> Ok, so it's something in the rest of the script that's causing >> >> >> this to happen. Ruslan, if you send your script, I can probably >> >> >> figure out why (usually, it's using another, non-agebraic udf >> >> >> in your foreach, or for pig 0.8, generating a constant in the foreach). >> >> >> >> >> >> D >> >> >> >> >> >> On Thu, Nov 17, 2011 at 9:59 AM, pablomar >> >> >> <[email protected]> wrote: >> >> >>> according to the stack trace, the algebraic is not being used >> >> >>> it says >> >> >>> updateTop(Top.java:139) >> >> >>> exec(Top.java:116) >> >> >>> >> >> >>> On 11/17/11, Dmitriy Ryaboy <[email protected]> wrote: >> >> >>>> The top udf does not try to process all data in memory if the >> >> >>>> algebraic optimization can be applied. It does need to keep >> >> >>>> the topn numbers in memory of course. Can you confirm >> >> >>>> algebraic mode is >> >> used? >> >> >>>> >> >> >>>> On Nov 17, 2011, at 6:13 AM, "Ruslan Al-fakikh" >> >> >>>> <[email protected]> >> >> >>>> wrote: >> >> >>>> >> >> >>>>> Hey guys, >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> I encounter java.lang.OutOfMemoryError when using TOP udf. >> >> >>>>> It seems that the udf tries to process all data in memory. >> >> >>>>> >> >> >>>>> Is there a workaround for TOP? Or maybe there is some other >> >> >>>>> way of getting top results? I cannot use LIMIT since I need >> >> >>>>> to 5% of data, not a constant number of rows. >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> I am using: >> >> >>>>> >> >> >>>>> Apache Pig version 0.8.1-cdh3u2 (rexported) >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> The stack trace is: >> >> >>>>> >> >> >>>>> [2011-11-16 12:34:55] INFO (CodecPool.java:128) - Got >> >> >>>>> brand-new decompressor >> >> >>>>> >> >> >>>>> [2011-11-16 12:34:55] INFO (Merger.java:473) - Down to the >> >> >>>>> last merge-pass, with 21 segments left of total size: >> >> >>>>> 2057257173 bytes >> >> >>>>> >> >> >>>>> [2011-11-16 12:34:55] INFO >> >> >>>>> (SpillableMemoryManager.java:154) - first memory handler >> >> >>>>> call- Usage threshold init = >> >> >>>>> 175308800(171200K) used = >> >> >>>>> 373454552(364701K) committed = 524288000(512000K) max = >> >> >>>>> 524288000(512000K) >> >> >>>>> >> >> >>>>> [2011-11-16 12:36:22] INFO >> >> >>>>> (SpillableMemoryManager.java:167) - first memory handler >> >> >>>>> call - Collection threshold init = >> >> >>>>> 175308800(171200K) used = >> >> >>>>> 496500704(484863K) committed = 524288000(512000K) max = >> >> >>>>> 524288000(512000K) >> >> >>>>> >> >> >>>>> [2011-11-16 12:37:28] INFO (TaskLogsTruncater.java:69) - >> >> >>>>> Initializing logs' >> >> >>>>> truncater with mapRetainSize=-1 and reduceRetainSize=-1 >> >> >>>>> >> >> >>>>> [2011-11-16 12:37:28] FATAL (Child.java:318) - Error running >> child : >> >> >>>>> java.lang.OutOfMemoryError: Java heap space >> >> >>>>> >> >> >>>>> at >> >> >>>>> java.util.Arrays.copyOfRange(Arrays.java:3209) >> >> >>>>> >> >> >>>>> at java.lang.String.<init>(String.java:215) >> >> >>>>> >> >> >>>>> at >> >> >>>>> java.io.DataInputStream.readUTF(DataInputStream.java:644) >> >> >>>>> >> >> >>>>> at >> >> >>>>> java.io.DataInputStream.readUTF(DataInputStream.java:547) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.data.BinInterSedes.readCharArray(BinInterSede >> >> >>>>> s.java >> >> >>>>> :210) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.data.BinInterSedes.readDatum(BinInterSedes.ja >> >> >>>>> va:333 >> >> >>>>> ) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.data.BinInterSedes.readDatum(BinInterSedes.ja >> >> >>>>> va:251 >> >> >>>>> ) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.data.BinInterSedes.addColsToTuple(BinInterSed >> >> >>>>> es.jav >> >> >>>>> a:555) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.data.BinSedesTuple.readFields(BinSedesTuple.j >> >> >>>>> ava:64 >> >> >>>>> ) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.data.InternalCachedBag$CachedBagIterator.hasN >> >> >>>>> ext(In >> >> >>>>> ternalCach >> >> >>>>> edBag.java:237) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.builtin.TOP.updateTop(TOP.java:139) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.builtin.TOP.exec(TOP.java:116) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.builtin.TOP.exec(TOP.java:65) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.backend.hadoop.executionengine.physicalLayer. >> >> >>>>> expres >> >> >>>>> sionOperat >> >> >>>>> ors.POUserFunc.getNext(POUserFunc.java:245) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.backend.hadoop.executionengine.physicalLayer. >> >> >>>>> expres >> >> >>>>> sionOperat >> >> >>>>> ors.POUserFunc.getNext(POUserFunc.java:287) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.backend.hadoop.executionengine.physicalLayer. >> >> >>>>> relati >> >> >>>>> onalOperat >> >> >>>>> ors.POForEach.processPlan(POForEach.java:338) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.backend.hadoop.executionengine.physicalLayer. >> >> >>>>> relati >> >> >>>>> onalOperat >> >> >>>>> ors.POForEach.getNext(POForEach.java:290) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.backend.hadoop.executionengine.physicalLayer. >> >> >>>>> Physic >> >> >>>>> alOperator >> >> >>>>> .processInput(PhysicalOperator.java:276) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.backend.hadoop.executionengine.physicalLayer. >> >> >>>>> relati >> >> >>>>> onalOperat >> >> >>>>> ors.POForEach.getNext(POForEach.java:240) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer >> >> >>>>> .PigMa >> >> >>>>> pReduce$Re >> >> >>>>> duce.runPipeline(PigMapReduce.java:434) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer >> >> >>>>> .PigMa >> >> >>>>> pReduce$Re >> >> >>>>> duce.processOnePackageOutput(PigMapReduce.java:402) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer >> >> >>>>> .PigMa >> >> >>>>> pReduce$Re >> >> >>>>> duce.reduce(PigMapReduce.java:382) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.pig.backend.hadoop.executionengine.mapReduceLayer >> >> >>>>> .PigMa >> >> >>>>> pReduce$Re >> >> >>>>> duce.reduce(PigMapReduce.java:251) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:176) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java: >> >> >>>>> 572) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:414) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.hadoop.mapred.Child$4.run(Child.java:270) >> >> >>>>> >> >> >>>>> at >> >> >>>>> java.security.AccessController.doPrivileged(Native >> >> >>>>> Method) >> >> >>>>> >> >> >>>>> at >> >> >>>>> javax.security.auth.Subject.doAs(Subject.java:396) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.hadoop.security.UserGroupInformation.doAs(UserGro >> >> >>>>> upInfo >> >> >>>>> rmation.ja >> >> >>>>> va:1127) >> >> >>>>> >> >> >>>>> at >> >> >>>>> org.apache.hadoop.mapred.Child.main(Child.java:264) >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> stderr logs >> >> >>>>> >> >> >>>>> Exception in thread "Low Memory Detector" >> >> >>>>> java.lang.OutOfMemoryError: Java heap space >> >> >>>>> >> >> >>>>> at >> >> >>>>> sun.management.MemoryNotifInfoCompositeData.getCompositeData >> >> >>>>> (Memor >> >> >>>>> yNotifInfo >> >> >>>>> CompositeData.java:42) >> >> >>>>> >> >> >>>>> at >> >> >>>>> sun.management.MemoryNotifInfoCompositeData.toCompositeData( >> >> >>>>> Memory >> >> >>>>> NotifInfoC >> >> >>>>> ompositeData.java:36) >> >> >>>>> >> >> >>>>> at >> >> >>>>> sun.management.MemoryImpl.createNotification(MemoryImpl.java >> >> >>>>> :168) >> >> >>>>> >> >> >>>>> at >> >> >>>>> >> >> >> >> >> sun.management.MemoryPoolImpl$CollectionSensor.triggerAction(MemoryPoolImpl. >> >> >>>>> java:300) >> >> >>>>> >> >> >>>>> at >> >> >>>>> sun.management.Sensor.trigger(Sensor.java:120) >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> >> >> >>>>> Thanks in advance! >> >> >>>>> >> >> >>>> >> >> >>> >> >> >> >> >> > >> >> > >> >> > >> >> > -- >> >> > Best Regards, >> >> > Ruslan Al-Fakikh >> >> > >> >> >> >> >> > >> >
