[scalding] Spill failed in Map cleanup phase
235 views
Skip to first unread message
Hagai Attias
Apr 16, 2015, 3:19:05 AM4/16/15
to cascadi...@googlegroups.com
Hi,
I'm running a scalding job over a very large dataset.
In it's last phase, I get the following exception in the Map Cleanup phase:
Note that running the same job on a smaller dataset does not cause this. Any idea?
| ERROR | cascading.tuple.hadoop.TupleSerialization$SerializationElementWriter | failed serializing token: null with classname: com.akamai.csi.jobs.needles.scalding.fpneedle.model.AggregationObject caught Throwable, no trap available, rethrowing |
Ken Krugler
Apr 16, 2015, 8:52:33 AM4/16/15
to cascadi...@googlegroups.com
Is com.akamai.csi.jobs.needles.scalding.fpneedle.model.AggregationObject a custom class you're using for data in a Tuple?
And if so, does it implement Hadoop's Writable interface?
If so, that could give you a serialization issue when the map output doesn't fit into memory, and thus the Tuple has to be serialized to disk.
-- Ken
From: Hagai Attias
Sent: April 16, 2015 12:19:05am PDT
To: cascadi...@googlegroups.com
Subject: [scalding] Spill failed in Map cleanup phase
--
You received this message because you are subscribed to the Google Groups "cascading-user" group.
To unsubscribe from this group and stop receiving emails from it, send an email to cascading-use...@googlegroups.com.
To post to this group, send email to cascadi...@googlegroups.com.
Visit this group at http://groups.google.com/group/cascading-user.
To view this discussion on the web visit https://groups.google.com/d/msgid/cascading-user/7cc48b71-3c07-420e-9d4c-69696916a3dd%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.
--------------------------
Ken Krugler
custom big data solutions & training
Hadoop, Cascading, Cassandra & Solr
--------------------------
Ken Krugler
custom big data solutions & training
Hadoop, Cascading, Cassandra & Solr
Hagai Attias
Apr 16, 2015, 9:24:37 AM4/16/15
to cascadi...@googlegroups.com
AggregationObject is a custom case class which extends scala.Serializable. It doesn't implement Hadoop's Writable interface explicitly.
It truly seems like a serialization issue when records are sorted and spill, but I'm not sure how to solve.
Ken Krugler
Apr 16, 2015, 1:13:12 PM4/16/15
to cascadi...@googlegroups.com
From: Hagai Attias
Sent: April 16, 2015 6:24:37am PDT
To: cascadi...@googlegroups.com
Subject: Re: [scalding] Spill failed in Map cleanup phase
AggregationObject is a custom case class which extends scala.Serializable. It doesn't implement Hadoop's Writable interface explicitly.It truly seems like a serialization issue when records are sorted and spill, but I'm not sure how to solve.
1. AggregationObject could also extend Hadoop's Writeable
2. Hadoop has Java serialization support, via the JavaSerialization package. You could install it via:
if (!serializations.contains(JavaSerialization.class.getName())) {
serializations.add(JavaSerialization.class.getName());
job.setStrings("io.serializations", serializations.toArray(new String[0]));
}
3. For Scalding I think the standard approach is to define a case class - I believe that gets handled automatically.
-- Ken
To view this discussion on the web visit https://groups.google.com/d/msgid/cascading-user/a58c6769-f3be-4a2c-bbe2-d734fcc6c58d%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.
Oscar Boykin
Apr 16, 2015, 2:02:37 PM4/16/15
to cascadi...@googlegroups.com
Kryo is set up by default in scalding and it is being used here (not Kryo in the call stack just a few lines from the top).
But the exception is thrown at:
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer$Buffer.write(MapTask.java:1089)and by that time, the data is already serialized. I think something else is going on here to trigger the IOException.
So, the root cause is:
Caused by: java.io.EOFException
What version of Hadoop is this (and scalding)?
Are you using an inner case class by chance (you cannot use inner case classes, which we say on the wiki, but people sometimes miss that). So, your case classes that you use as records need to be top-level classes due to the way scalac compiles them.
To view this discussion on the web visit https://groups.google.com/d/msgid/cascading-user/F8AFF803-C2F2-4362-9FEF-57FC95E66433%40transpac.com.
Hagai Attias
Apr 16, 2015, 2:37:59 PM4/16/15
to cascadi...@googlegroups.com
Yeah I know the inner case classes limitation, all my case classes are top level.
scalding version is 0.10.
Message has been deleted
Hagai Attias
Apr 19, 2015, 3:35:43 AM4/19/15
to cascadi...@googlegroups.com
A quick update,
I reduced the code in order to identify the problematic step. As suspected, it seems the error is coming from a leftJoin operation between two Grouped pipes.
Now the stack trace is very similar but the root cause is "value too long to fit in integer", again from the Map cleanup phase. So it still seems like a failure in the sort and spill phase after map logic is done. Here's the updated trace:
failed serializing token: null with classname: com.akamai.csi.jobs.needles.scalding.fpneedle.model.keys.PolicySelectorKey
java.io.IOException: Spill failed
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer$Buffer.write(MapTask.java:1089)
at java.io.DataOutputStream.write(DataOutputStream.java:90)
at java.io.DataOutputStream.write(DataOutputStream.java:90)
at java.io.ByteArrayOutputStream.writeTo(ByteArrayOutputStream.java:109)
at com.twitter.chill.KryoPool$2$1.writeOutputTo(KryoPool.java:79)
at com.twitter.chill.hadoop.KryoSerializer.serialize(KryoSerializer.java:51)
at cascading.tuple.hadoop.TupleSerialization$SerializationElementWriter.write(TupleSerialization.java:750)
at cascading.tuple.io.TupleOutputStream.writeElement(TupleOutputStream.java:114)
at cascading.tuple.io.TupleOutputStream.write(TupleOutputStream.java:89)
at cascading.tuple.io.TupleOutputStream.writeTuple(TupleOutputStream.java:64)
at cascading.tuple.hadoop.io.HadoopTupleOutputStream.writeIndexTuple(HadoopTupleOutputStream.java:161)
at cascading.tuple.hadoop.io.IndexTupleSerializer.serialize(IndexTupleSerializer.java:37)
at cascading.tuple.hadoop.io.IndexTupleSerializer.serialize(IndexTupleSerializer.java:28)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:937)
at org.apache.hadoop.mapred.MapTask$OldOutputCollector.collect(MapTask.java:526)
at cascading.tap.hadoop.util.MeasuredOutputCollector.collect(MeasuredOutputCollector.java:69)
at cascading.flow.hadoop.stream.HadoopCoGroupGate.receive(HadoopCoGroupGate.java:88)
at cascading.flow.hadoop.stream.HadoopCoGroupGate.receive(HadoopCoGroupGate.java:43)
at cascading.flow.stream.FunctionEachStage$1.collect(FunctionEachStage.java:80)
at cascading.tuple.TupleEntryCollector.safeCollect(TupleEntryCollector.java:145)
at cascading.tuple.TupleEntryCollector.add(TupleEntryCollector.java:133)
at com.twitter.scalding.FlatMapFunction$$anonfun$operate$2.apply(Operations.scala:48)
at com.twitter.scalding.FlatMapFunction$$anonfun$operate$2.apply(Operations.scala:46)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at com.twitter.scalding.FlatMapFunction.operate(Operations.scala:46)
at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:99)
at cascading.flow.stream.FunctionEachStage.receive(FunctionEachStage.java:39)
at cascading.flow.stream.FunctionEachStage$1.collect(FunctionEachStage.java:80)
at cascading.tuple.TupleEntryCollector.safeCollect(TupleEntryCollector.java:145)
at cascading.tuple.TupleEntryCollector.add(TupleEntryCollector.java:133)
at com.twitter.scalding.MapsideReduce.add(Operations.scala:143)
at com.twitter.scalding.MapsideReduce.flush(Operations.scala:162)
at cascading.flow.stream.OperatorStage.complete(OperatorStage.java:292)
at cascading.flow.stream.Duct.complete(Duct.java:81)
at cascading.flow.stream.OperatorStage.complete(OperatorStage.java:296)
at cascading.flow.stream.Fork.complete(Fork.java:60)
at cascading.flow.stream.SourceStage.map(SourceStage.java:105)
at cascading.flow.stream.SourceStage.run(SourceStage.java:58)
at cascading.flow.hadoop.FlowMapper.run(FlowMapper.java:127)
at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:417)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:332)
at org.apache.hadoop.mapred.Child$4.run(Child.java:268)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:396)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1408)
at org.apache.hadoop.mapred.Child.main(Child.java:262)
Caused by: cascading.CascadingException: java.io.IOException: value too long to fit in integer
at cascading.tuple.hadoop.util.IndexTupleCoGroupingComparator.compare(IndexTupleCoGroupingComparator.java:50)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.compare(MapTask.java:987)
at org.apache.hadoop.util.QuickSort.sortInternal(QuickSort.java:100)
at org.apache.hadoop.util.QuickSort.sort(QuickSort.java:64)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.sortAndSpill(MapTask.java:1277)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.access$1900(MapTask.java:724)
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer$SpillThread.run(MapTask.java:1222)
Caused by: java.io.IOException: value too long to fit in integer
at org.apache.hadoop.io.WritableUtils.readVInt(WritableUtils.java:331)
at cascading.tuple.hadoop.io.HadoopTupleInputStream.readVInt(HadoopTupleInputStream.java:70)
at cascading.tuple.hadoop.io.HadoopTupleInputStream.getNumElements(HadoopTupleInputStream.java:42)
at cascading.tuple.hadoop.util.DeserializerComparator.compareTuples(DeserializerComparator.java:147)
at cascading.tuple.hadoop.util.IndexTupleCoGroupingComparator.compare(IndexTupleCoGroupingComparator.java:41)
... 6 more
On Thursday, April 16, 2015 at 10:19:05 AM UTC+3, Hagai Attias wrote:
Hagai Attias
Apr 19, 2015, 7:24:41 AM4/19/15
to cascadi...@googlegroups.com
Changing to skewJoin (.sketched.leftJoin) triggered back the EOFException.
Message has been deleted
Hagai Attias
Apr 20, 2015, 7:50:09 AM4/20/15
to cascadi...@googlegroups.com
One more update, reducing the code further revealed that the error comes from a GroupBy operation.
I have a pipe which is Grouped, and then I go like
pipe
.ToTypedPipe
.groupBy { case (k, v) => PolicyHostSelectorKey(k.appId, k.host, k.selector) }
Here's PolicyHostSelectorKey:
case class PolicyHostSelectorKey(appId: String, host: String, selector: String) extends Serializable {
}
object PolicyHostSelectorKey {
implicit val ord: Ordering[PolicyHostSelectorKey] = Ordering.by(PolicyHostSelectorKey.unapply)
}And now the error pops when keys are being ordered. Is something wrong with my ordering?
Here's the updated stacktrace:
Caused by: cascading.CascadingException: unable to compare Tuples, likely a CoGroup is being attempted on fields of different types or custom comparators are incorrectly set on Fields, lhs: '7.175E-43' rhs: 'PolicyHostSelectorKey(klrs_16124,somedomain.com,ARGS_NAMES:{"cid":"1090cbb008","options":{"data":{"styles":[{"name":"Image With Caption","props":{"color":{"value":""},"font-family":{"value":""},"font-weight":{"value":""},"text-align":{"value":""},"font-size":{"value":""},"font-style":{"value":""},"line-height":{"value":""}},"selector":".mcnTextContent"}],"numberOfCaptions":1,"captionPosition":"right","captionWidth":"half","captions":[{"image":{"alt":"","width":1024,"src":"https:\/\/domain.com\/98943935dd3a74794c6fe19c8\/images\/77d330cf-ed79-42e9-8918-06950bd9228c.png","height":683},"text":"Your text caption goes here"}],"align":"center","selectedCaption":0},"socket_id":"44455.18049364","html":"<table border)'
at cascading.tuple.hadoop.util.TupleElementComparator.compare(TupleElementComparator.java:91)
at cascading.tuple.hadoop.util.TupleElementComparator.compare(TupleElementComparator.java:33)
at cascading.tuple.hadoop.util.DeserializerComparator.compareTuples(DeserializerComparator.java:160)
... 7 more
Caused by: java.lang.ClassCastException: java.lang.Float cannot be cast to com.akamai.csi.jobs.needles.scalding.fpneedle.model.keys.PolicyHostSelectorKey
at com.akamai.csi.jobs.needles.scalding.fpneedle.model.keys.PolicyHostSelectorKey$$anonfun$1.apply(PolicyHostSelectorKey.scala:19)
at scala.math.Ordering$$anonfun$by$1.apply(Ordering.scala:219)
at scala.math.Ordering$$anonfun$by$1.apply(Ordering.scala:219)
at scala.math.Ordering$$anon$9.compare(Ordering.scala:200)
at cascading.tuple.hadoop.util.TupleElementComparator.compare(TupleElementComparator.java:87)
... 9 more Oscar Boykin
Apr 20, 2015, 10:20:37 PM4/20/15
to cascadi...@googlegroups.com
I don't see any problem (and you're not doing anything unusual), but if you're getting EOF exceptions and failures to decode integers, my guess is that there is some kind of corruption going on when the data is being written or read.
I've seen similar issues in the past with compression and some kind of thread bugs with Hadoop reusing compressors that were in bad states.
I don't know what to tell you. I really doubt you are doing anything wrong at the scalding level, I think there is some kind of misconfiguration at the cluster level.
That said, can you try using a more recent version of scalding (or even just trying a more recent cascading by including the latest in the 2.x series).
--
You received this message because you are subscribed to the Google Groups "cascading-user" group.
To unsubscribe from this group and stop receiving emails from it, send an email to cascading-use...@googlegroups.com.
To post to this group, send email to cascadi...@googlegroups.com.
Visit this group at http://groups.google.com/group/cascading-user.
To view this discussion on the web visit https://groups.google.com/d/msgid/cascading-user/798bd745-95a9-4195-a823-4858afe21ce2%40googlegroups.com.
Hagai Attias
Apr 21, 2015, 4:19:19 AM4/21/15
to cascadi...@googlegroups.com
Ok. How 'bout upgrading chill? I'm using 0.3.6.
Oscar Boykin
Apr 21, 2015, 1:42:41 PM4/21/15
to cascadi...@googlegroups.com
0.3.6 is likely binary incompatible with 0.4.x, which is the next latest versions, I don't think you can just upgrade that, you'd need to use a later version of scalding.
Is there any reason you're using such an old version of scalding?
To view this discussion on the web visit https://groups.google.com/d/msgid/cascading-user/19f44c59-c090-4e72-ae18-07393b5f7e6a%40googlegroups.com.
Hagai Attias
Apr 21, 2015, 1:45:56 PM4/21/15
to cascadi...@googlegroups.com
Actually no reason. Just upgraded to 0.13.1, will run again and update.
Hagai Attias
Apr 26, 2015, 1:24:47 PM4/26/15
to cascadi...@googlegroups.com
Well, updating to 0.13.1 seems to resolve the previous issue and introduced a new one.
Now in the last phase, 2 out of 100 reducers fails due to OOM. Looking at the stacktrace I thought that maybe it comes from one of mkString on the sets I have, but reducing / removing those sets did not change anything. The only mkString is now inside a groupBy operation.
I'm running with
| mapred.reduce.child.java.opts | -Xmx2576980378 |
| mapred.child.java.opts | -Xmx200m |
Maybe the stacktrace can be helpful this time:
2015-04-26 15:41:09,749 INFO org.apache.hadoop.mapred.TaskLogsTruncater: Initializing logs' truncater with mapRetainSize=-1 and reduceRetainSize=-1 2015-04-26 15:41:09,750 FATAL org.apache.hadoop.mapred.Child: 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 com.esotericsoftware.kryo.io.Input.readString(Input.java:448) at com.esotericsoftware.kryo.serializers.DefaultSerializers$StringSerializer.read(DefaultSerializers.java:157) at com.esotericsoftware.kryo.serializers.DefaultSerializers$StringSerializer.read(DefaultSerializers.java:146) at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:732) at com.twitter.chill.TraversableSerializer.read(Traversable.scala:43) at com.twitter.chill.TraversableSerializer.read(Traversable.scala:21) at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:651) at com.esotericsoftware.kryo.serializers.FieldSerializer$ObjectField.read(FieldSerializer.java:605) at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:221) at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:732) at com.twitter.chill.TraversableSerializer.read(Traversable.scala:43) at com.twitter.chill.TraversableSerializer.read(Traversable.scala:21) at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:629) at com.twitter.chill.SerDeState.readObject(SerDeState.java:58) at com.twitter.chill.KryoPool.fromBytes(KryoPool.java:105) at com.twitter.chill.hadoop.KryoDeserializer.deserialize(KryoDeserializer.java:51) at cascading.tuple.hadoop.TupleSerialization$SerializationElementReader.read(TupleSerialization.java:628) at cascading.tuple.hadoop.io.HadoopTupleInputStream.readType(HadoopTupleInputStream.java:105) at cascading.tuple.hadoop.io.HadoopTupleInputStream.getNextElement(HadoopTupleInputStream.java:52) at cascading.tuple.io.TupleInputStream.readTuple(TupleInputStream.java:78) at cascading.tuple.hadoop.io.TupleDeserializer.deserialize(TupleDeserializer.java:40) at cascading.tuple.hadoop.io.TupleDeserializer.deserialize(TupleDeserializer.java:28) at org.apache.hadoop.mapred.Task$ValuesIterator.readNextValue(Task.java:1261) at org.apache.hadoop.mapred.Task$ValuesIterator.next(Task.java:1199) at org.apache.hadoop.mapred.ReduceTask$ReduceValuesIterator.moveToNext(ReduceTask.java:255) at org.apache.hadoop.mapred.ReduceTask$ReduceValuesIterator.next(ReduceTask.java:251) at cascading.flow.hadoop.util.TimedIterator.next(TimedIterator.java:74) at cascading.flow.hadoop.HadoopGroupByClosure$1.next(HadoopGroupByClosure.java:113) at cascading.flow.hadoop.HadoopGroupByClosure$1.next(HadoopGroupByClosure.java:71) at cascading.pipe.joiner.InnerJoin$JoinIterator.next(InnerJoin.java:190)
Hagai Attias
Apr 27, 2015, 4:56:05 PM4/27/15
to cascadi...@googlegroups.com
Well it seems that the "spill failed" issue occurs when tweaking io.sort.spill.percent. Here's another exmaple.
Running with the default of 0.8, although increases spilled records, seems to overcome this. Still, not sure why.
Now left to deal with the OOM and GC Overhead issues.
Oscar Boykin
Apr 27, 2015, 5:11:28 PM4/27/15
to cascadi...@googlegroups.com
TraversableSerializer suggests you are sending lists, sets or maps.
If you are aggregating those things, you can create very large records that will OOM.
I don't see the code, so I don't know what is going on.
--
You received this message because you are subscribed to the Google Groups "cascading-user" group.
To unsubscribe from this group and stop receiving emails from it, send an email to cascading-use...@googlegroups.com.
To post to this group, send email to cascadi...@googlegroups.com.
Visit this group at http://groups.google.com/group/cascading-user.
To view this discussion on the web visit https://groups.google.com/d/msgid/cascading-user/3259988c-f83f-450e-9fc5-63eca8a9aa25%40googlegroups.com.
Reply all
Reply to author
Forward
0 new messages