You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@spark.apache.org by Soren Macbeth <so...@yieldbot.com> on 2014/05/04 19:35:42 UTC
bug using kryo as closure serializer
apologies for the cross-list posts, but I've gotten zero response in the
user list and I guess this list is probably more appropriate.
According to the documentation, using the KryoSerializer for closures is
supported. However, when I try to set `spark.closure.serializer` to
`org.apache.spark.serializer.KryoSerializer` thing fail pretty miserably.
The first thing that happens it that is throws exceptions over and over
that it cannot locate my registrator class, which is located in my assembly
jar like so:
14/05/04 12:03:20 ERROR serializer.KryoSerializer: Failed to run
spark.kryo.registrator
java.lang.ClassNotFoundException: pickles.kryo.PicklesRegistrator
at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:270)
at
org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:63)
at
org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:61)
at scala.Option.foreach(Option.scala:236)
at
org.apache.spark.serializer.KryoSerializer.newKryo(KryoSerializer.scala:61)
at
org.apache.spark.serializer.KryoSerializerInstance.<init>(KryoSerializer.scala:116)
at
org.apache.spark.serializer.KryoSerializer.newInstance(KryoSerializer.scala:79)
at
org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:180)
at
org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
at
org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:415)
at
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
at
org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:724)
Now, I would expect it not to be able to find this class since it hasn't
yet fetched my assembly jar to the executors. Once it does fetch my jar,
those expections stop. Next, all the executor task die with the following
exception:
java.nio.ReadOnlyBufferException
at java.nio.ByteBuffer.array(ByteBuffer.java:961)
at
org.apache.spark.serializer.KryoSerializerInstance.deserialize(KryoSerializer.scala:136)
at
org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:193)
at
org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
at
org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:415)
at
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
at
org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:724)
AFAIK, I'm not doing anything out of the ordinary, just turning on kryo and
using the registrator mechanism to register a couple custom serializers.
The reason I tried turning on kryo for closure in the first place is
because of a different bug that I was hitting during fetching and
deserializing of tasks from my executors, which I detailed here:
http://apache-spark-user-list.1001560.n3.nabble.com/Crazy-Kryo-Exception-td5257.html
Here's hoping some on this list can help me track down what's happening as
I didn't get a single reply on the user list.
Re: bug using kryo as closure serializer
Posted by Reynold Xin <rx...@databricks.com>.
Kryo does generate code for serialization, so the CPU overhead is quite
lower than Java (which I think just uses reflection). As I understand, they
also have a new implementation that uses unsafe intrinsics, which should
lead to even higher performance.
The generated byte[] size was a lot smaller in Kryo, especially for arrays
of objects of the same type. It doesn't need to write the class name for
every object, and it also reduces the size of ints and such using zig zag
encoding.
I don't have numbers around anymore, but when I was benchmarking them, kryo
was substantially better than java. The only reason it is not on by default
is because it doesn't always work.
On Sun, May 4, 2014 at 1:39 PM, Mridul Muralidharan <mr...@gmail.com>wrote:
> On a slightly related note (apologies Soren for hijacking the thread),
> Reynold how much better is kryo from spark's usage point of view
> compared to the default java serialization (in general, not for
> closures) ?
> The numbers on kyro site are interesting, but since you have played
> the most with kryo in context of spark (i think) - how do you rate it
> along lines of :
>
> 1) computational overhead compared to java serialization.
> 2) memory overhead.
> 3) generated byte[] size.
>
>
> Particularly given the bugs Patrick and I had looked into in past
> along flush, etc I was always skeptical about using kyro.
> But given the pretty nasty issues with OOM's via java serialization we
> are seeing, wanted to know your thoughts on use of kyro with spark.
> (Will be slightly involved to ensure everything gets registered, but I
> want to go down the path assuming I hear good things in context of
> spark)
>
> Thanks,
> Mridul
>
>
> On Mon, May 5, 2014 at 1:20 AM, Reynold Xin <rx...@databricks.com> wrote:
> > I added the config option to use the non-default serializer. However, at
> > the time, Kryo fails serializing pretty much any closures so that option
> > was never really used / recommended.
> >
> > Since then the Scala ecosystem has developed, and some other projects are
> > starting to use Kryo to serialize more Scala data structures, so I
> wouldn't
> > be surprised if there is a way to work around this now. However, I don't
> > have enough time to look into it at this point. If you do, please do post
> > your findings. Thanks.
> >
> >
> >
> > On Sun, May 4, 2014 at 10:35 AM, Soren Macbeth <so...@yieldbot.com>
> wrote:
> >
> >> apologies for the cross-list posts, but I've gotten zero response in the
> >> user list and I guess this list is probably more appropriate.
> >>
> >> According to the documentation, using the KryoSerializer for closures is
> >> supported. However, when I try to set `spark.closure.serializer` to
> >> `org.apache.spark.serializer.KryoSerializer` thing fail pretty
> miserably.
> >>
> >> The first thing that happens it that is throws exceptions over and over
> >> that it cannot locate my registrator class, which is located in my
> assembly
> >> jar like so:
> >>
> >> 14/05/04 12:03:20 ERROR serializer.KryoSerializer: Failed to run
> >> spark.kryo.registrator
> >> java.lang.ClassNotFoundException: pickles.kryo.PicklesRegistrator
> >> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> >> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> >> at java.security.AccessController.doPrivileged(Native Method)
> >> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> >> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> >> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> >> at java.lang.Class.forName0(Native Method)
> >> at java.lang.Class.forName(Class.java:270)
> >> at
> >>
> >>
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:63)
> >> at
> >>
> >>
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:61)
> >> at scala.Option.foreach(Option.scala:236)
> >> at
> >>
> org.apache.spark.serializer.KryoSerializer.newKryo(KryoSerializer.scala:61)
> >> at
> >>
> >>
> org.apache.spark.serializer.KryoSerializerInstance.<init>(KryoSerializer.scala:116)
> >> at
> >>
> >>
> org.apache.spark.serializer.KryoSerializer.newInstance(KryoSerializer.scala:79)
> >> at
> >>
> >>
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:180)
> >> at
> >>
> >>
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> >> at
> >>
> >>
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> >> at java.security.AccessController.doPrivileged(Native Method)
> >> at javax.security.auth.Subject.doAs(Subject.java:415)
> >> at
> >>
> >>
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> >> at
> >>
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> >> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> >> at
> >>
> >>
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> >> at
> >>
> >>
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> >> at java.lang.Thread.run(Thread.java:724)
> >>
> >> Now, I would expect it not to be able to find this class since it hasn't
> >> yet fetched my assembly jar to the executors. Once it does fetch my jar,
> >> those expections stop. Next, all the executor task die with the
> following
> >> exception:
> >>
> >> java.nio.ReadOnlyBufferException
> >> at java.nio.ByteBuffer.array(ByteBuffer.java:961)
> >> at
> >>
> >>
> org.apache.spark.serializer.KryoSerializerInstance.deserialize(KryoSerializer.scala:136)
> >> at
> >>
> >>
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:193)
> >> at
> >>
> >>
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> >> at
> >>
> >>
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> >> at java.security.AccessController.doPrivileged(Native Method)
> >> at javax.security.auth.Subject.doAs(Subject.java:415)
> >> at
> >>
> >>
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> >> at
> >>
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> >> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> >> at
> >>
> >>
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> >> at
> >>
> >>
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> >> at java.lang.Thread.run(Thread.java:724)
> >>
> >> AFAIK, I'm not doing anything out of the ordinary, just turning on kryo
> and
> >> using the registrator mechanism to register a couple custom serializers.
> >>
> >> The reason I tried turning on kryo for closure in the first place is
> >> because of a different bug that I was hitting during fetching and
> >> deserializing of tasks from my executors, which I detailed here:
> >>
> >>
> >>
> http://apache-spark-user-list.1001560.n3.nabble.com/Crazy-Kryo-Exception-td5257.html
> >>
> >> Here's hoping some on this list can help me track down what's happening
> as
> >> I didn't get a single reply on the user list.
> >>
>
Re: bug using kryo as closure serializer
Posted by Mridul Muralidharan <mr...@gmail.com>.
On a slightly related note (apologies Soren for hijacking the thread),
Reynold how much better is kryo from spark's usage point of view
compared to the default java serialization (in general, not for
closures) ?
The numbers on kyro site are interesting, but since you have played
the most with kryo in context of spark (i think) - how do you rate it
along lines of :
1) computational overhead compared to java serialization.
2) memory overhead.
3) generated byte[] size.
Particularly given the bugs Patrick and I had looked into in past
along flush, etc I was always skeptical about using kyro.
But given the pretty nasty issues with OOM's via java serialization we
are seeing, wanted to know your thoughts on use of kyro with spark.
(Will be slightly involved to ensure everything gets registered, but I
want to go down the path assuming I hear good things in context of
spark)
Thanks,
Mridul
On Mon, May 5, 2014 at 1:20 AM, Reynold Xin <rx...@databricks.com> wrote:
> I added the config option to use the non-default serializer. However, at
> the time, Kryo fails serializing pretty much any closures so that option
> was never really used / recommended.
>
> Since then the Scala ecosystem has developed, and some other projects are
> starting to use Kryo to serialize more Scala data structures, so I wouldn't
> be surprised if there is a way to work around this now. However, I don't
> have enough time to look into it at this point. If you do, please do post
> your findings. Thanks.
>
>
>
> On Sun, May 4, 2014 at 10:35 AM, Soren Macbeth <so...@yieldbot.com> wrote:
>
>> apologies for the cross-list posts, but I've gotten zero response in the
>> user list and I guess this list is probably more appropriate.
>>
>> According to the documentation, using the KryoSerializer for closures is
>> supported. However, when I try to set `spark.closure.serializer` to
>> `org.apache.spark.serializer.KryoSerializer` thing fail pretty miserably.
>>
>> The first thing that happens it that is throws exceptions over and over
>> that it cannot locate my registrator class, which is located in my assembly
>> jar like so:
>>
>> 14/05/04 12:03:20 ERROR serializer.KryoSerializer: Failed to run
>> spark.kryo.registrator
>> java.lang.ClassNotFoundException: pickles.kryo.PicklesRegistrator
>> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
>> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
>> at java.security.AccessController.doPrivileged(Native Method)
>> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
>> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
>> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
>> at java.lang.Class.forName0(Native Method)
>> at java.lang.Class.forName(Class.java:270)
>> at
>>
>> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:63)
>> at
>>
>> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:61)
>> at scala.Option.foreach(Option.scala:236)
>> at
>> org.apache.spark.serializer.KryoSerializer.newKryo(KryoSerializer.scala:61)
>> at
>>
>> org.apache.spark.serializer.KryoSerializerInstance.<init>(KryoSerializer.scala:116)
>> at
>>
>> org.apache.spark.serializer.KryoSerializer.newInstance(KryoSerializer.scala:79)
>> at
>>
>> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:180)
>> at
>>
>> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
>> at
>>
>> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
>> at java.security.AccessController.doPrivileged(Native Method)
>> at javax.security.auth.Subject.doAs(Subject.java:415)
>> at
>>
>> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
>> at
>> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
>> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
>> at
>>
>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
>> at
>>
>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
>> at java.lang.Thread.run(Thread.java:724)
>>
>> Now, I would expect it not to be able to find this class since it hasn't
>> yet fetched my assembly jar to the executors. Once it does fetch my jar,
>> those expections stop. Next, all the executor task die with the following
>> exception:
>>
>> java.nio.ReadOnlyBufferException
>> at java.nio.ByteBuffer.array(ByteBuffer.java:961)
>> at
>>
>> org.apache.spark.serializer.KryoSerializerInstance.deserialize(KryoSerializer.scala:136)
>> at
>>
>> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:193)
>> at
>>
>> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
>> at
>>
>> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
>> at java.security.AccessController.doPrivileged(Native Method)
>> at javax.security.auth.Subject.doAs(Subject.java:415)
>> at
>>
>> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
>> at
>> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
>> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
>> at
>>
>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
>> at
>>
>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
>> at java.lang.Thread.run(Thread.java:724)
>>
>> AFAIK, I'm not doing anything out of the ordinary, just turning on kryo and
>> using the registrator mechanism to register a couple custom serializers.
>>
>> The reason I tried turning on kryo for closure in the first place is
>> because of a different bug that I was hitting during fetching and
>> deserializing of tasks from my executors, which I detailed here:
>>
>>
>> http://apache-spark-user-list.1001560.n3.nabble.com/Crazy-Kryo-Exception-td5257.html
>>
>> Here's hoping some on this list can help me track down what's happening as
>> I didn't get a single reply on the user list.
>>
Re: bug using kryo as closure serializer
Posted by Soren Macbeth <so...@yieldbot.com>.
I just took a peek at KryoSerializer and it looks like you're already using
all the scala stuff from chill in there, so I would imagine that scala
things should serialize pretty well.
Seems like the readonly bytebuffer thing might be some other sort of
downstream bug.
On Sun, May 4, 2014 at 10:24 PM, Reynold Xin <rx...@databricks.com> wrote:
> Technically you only need to change the build file, and change part of a
> line in SparkEnv so you don't have to break your oath :)
>
>
>
> On Sun, May 4, 2014 at 10:22 PM, Soren Macbeth <so...@yieldbot.com> wrote:
>
> > that would violate my personal oath of never writing a single line of
> > scala, but I might be able to do that if I can get past the issue this
> > issue I'm struggling with in this thread.
> >
> > On Sunday, May 4, 2014, Reynold Xin <rx...@databricks.com> wrote:
> >
> > > Thanks. Do you mind playing with chill-scala a little bit and see if it
> > > actually works well for closures? One way to try is to hard code the
> > > serializer to use Kryo with chill-scala, and then run through all the
> > unit
> > > tests.
> > >
> > > If it works well, we can incorporate that in the next release (probably
> > not
> > > 1.0, but after that).
> > >
> > >
> > > On Sun, May 4, 2014 at 9:08 PM, Soren Macbeth <so...@yieldbot.com>
> > wrote:
> > >
> > > > fwiw, it seems like it wouldn't be very difficult to integrate
> > > chill-scala,
> > > > since you're already chill-java and probably get kryo serialization
> of
> > > > closures and all sorts of other scala stuff for free. All that would
> be
> > > > needed would be to include the dependency and then update
> > KryoSerializer
> > > to
> > > > register the stuff in chill-scala.
> > > >
> > > > In that case, you could probably safely make kryo the default
> > serializer,
> > > > which I think would be desirable in general.
> > > >
> > > >
> > > > On Sun, May 4, 2014 at 8:48 PM, Reynold Xin <rx...@databricks.com>
> > wrote:
> > > >
> > > > > Good idea. I submitted a pull request for the doc update here:
> > > > > https://github.com/apache/spark/pull/642
> > > > >
> > > > >
> > > > > On Sun, May 4, 2014 at 3:54 PM, Soren Macbeth <so...@yieldbot.com>
> > > > wrote:
> > > > >
> > > > > > Thanks for the reply!
> > > > > >
> > > > > > Ok, if that's the case, I'd recommend a note to that affect in
> the
> > > docs
> > > > > at
> > > > > > least.
> > > > > >
> > > > > > Just to give some more context here, I'm working on a Clojure DSL
> > for
> > > > > Spark
> > > > > > called Flambo, which I plan to open source shortly. If I could
> I'd
> > > like
> > > > > to
> > > > > > focus on the initial bug that I hit.
> > > > > >
> > > > > > Exception in thread "main" org.apache.spark.SparkException: Job
> > > > aborted:
> > > > > > Exception while deserializing and fetching task:
> > > > > > com.esotericsoftware.kryo.KryoException:
> > > > > > java.lang.IllegalArgumentException: Can not set final
> > > > > > scala.collection.convert.Wrappers field
> > > > > > scala.collection.convert.Wrappers$SeqWrapper.$outer to
> > > > > > clojure.lang.PersistentVector
> > > > > > Serialization trace:
> > > > > > $outer (scala.collection.convert.Wrappers$SeqWrapper)
> > > > > > at
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1020)
> > > > > > at
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1018)
> > > > > > at
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> > > > > > at
> > > > > >
> scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> > > > > > at org.apache.spark.scheduler.DAGScheduler.org
> > > > > >
> > > >
> > $apache$spark$scheduler$DAGScheduler$$abortStage(DAGScheduler.scala:1018)
> > > > > > at
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > > > > > at
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > > > > > at scala.Option.foreach(Option.scala:236)
> > > > > > at
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler.processEvent(DAGScheduler.scala:604)
> > > > > > at
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$start$1$$anon$2$$anonfun$receive$1.applyOrElse(DAGScheduler.scala:190)
> > > > > > at
> akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
> > > > > > at akka.actor.ActorCell.invoke(ActorCell.scala:456)
> > > > > >
> >
>
Re: bug using kryo as closure serializer
Posted by Reynold Xin <rx...@databricks.com>.
Technically you only need to change the build file, and change part of a
line in SparkEnv so you don't have to break your oath :)
On Sun, May 4, 2014 at 10:22 PM, Soren Macbeth <so...@yieldbot.com> wrote:
> that would violate my personal oath of never writing a single line of
> scala, but I might be able to do that if I can get past the issue this
> issue I'm struggling with in this thread.
>
> On Sunday, May 4, 2014, Reynold Xin <rx...@databricks.com> wrote:
>
> > Thanks. Do you mind playing with chill-scala a little bit and see if it
> > actually works well for closures? One way to try is to hard code the
> > serializer to use Kryo with chill-scala, and then run through all the
> unit
> > tests.
> >
> > If it works well, we can incorporate that in the next release (probably
> not
> > 1.0, but after that).
> >
> >
> > On Sun, May 4, 2014 at 9:08 PM, Soren Macbeth <so...@yieldbot.com>
> wrote:
> >
> > > fwiw, it seems like it wouldn't be very difficult to integrate
> > chill-scala,
> > > since you're already chill-java and probably get kryo serialization of
> > > closures and all sorts of other scala stuff for free. All that would be
> > > needed would be to include the dependency and then update
> KryoSerializer
> > to
> > > register the stuff in chill-scala.
> > >
> > > In that case, you could probably safely make kryo the default
> serializer,
> > > which I think would be desirable in general.
> > >
> > >
> > > On Sun, May 4, 2014 at 8:48 PM, Reynold Xin <rx...@databricks.com>
> wrote:
> > >
> > > > Good idea. I submitted a pull request for the doc update here:
> > > > https://github.com/apache/spark/pull/642
> > > >
> > > >
> > > > On Sun, May 4, 2014 at 3:54 PM, Soren Macbeth <so...@yieldbot.com>
> > > wrote:
> > > >
> > > > > Thanks for the reply!
> > > > >
> > > > > Ok, if that's the case, I'd recommend a note to that affect in the
> > docs
> > > > at
> > > > > least.
> > > > >
> > > > > Just to give some more context here, I'm working on a Clojure DSL
> for
> > > > Spark
> > > > > called Flambo, which I plan to open source shortly. If I could I'd
> > like
> > > > to
> > > > > focus on the initial bug that I hit.
> > > > >
> > > > > Exception in thread "main" org.apache.spark.SparkException: Job
> > > aborted:
> > > > > Exception while deserializing and fetching task:
> > > > > com.esotericsoftware.kryo.KryoException:
> > > > > java.lang.IllegalArgumentException: Can not set final
> > > > > scala.collection.convert.Wrappers field
> > > > > scala.collection.convert.Wrappers$SeqWrapper.$outer to
> > > > > clojure.lang.PersistentVector
> > > > > Serialization trace:
> > > > > $outer (scala.collection.convert.Wrappers$SeqWrapper)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1020)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1018)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> > > > > at
> > > > > scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> > > > > at org.apache.spark.scheduler.DAGScheduler.org
> > > > >
> > >
> $apache$spark$scheduler$DAGScheduler$$abortStage(DAGScheduler.scala:1018)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > > > > at scala.Option.foreach(Option.scala:236)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler.processEvent(DAGScheduler.scala:604)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$start$1$$anon$2$$anonfun$receive$1.applyOrElse(DAGScheduler.scala:190)
> > > > > at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
> > > > > at akka.actor.ActorCell.invoke(ActorCell.scala:456)
> > > > >
>
Re: bug using kryo as closure serializer
Posted by Soren Macbeth <so...@yieldbot.com>.
that would violate my personal oath of never writing a single line of
scala, but I might be able to do that if I can get past the issue this
issue I'm struggling with in this thread.
On Sunday, May 4, 2014, Reynold Xin <rx...@databricks.com> wrote:
> Thanks. Do you mind playing with chill-scala a little bit and see if it
> actually works well for closures? One way to try is to hard code the
> serializer to use Kryo with chill-scala, and then run through all the unit
> tests.
>
> If it works well, we can incorporate that in the next release (probably not
> 1.0, but after that).
>
>
> On Sun, May 4, 2014 at 9:08 PM, Soren Macbeth <so...@yieldbot.com> wrote:
>
> > fwiw, it seems like it wouldn't be very difficult to integrate
> chill-scala,
> > since you're already chill-java and probably get kryo serialization of
> > closures and all sorts of other scala stuff for free. All that would be
> > needed would be to include the dependency and then update KryoSerializer
> to
> > register the stuff in chill-scala.
> >
> > In that case, you could probably safely make kryo the default serializer,
> > which I think would be desirable in general.
> >
> >
> > On Sun, May 4, 2014 at 8:48 PM, Reynold Xin <rx...@databricks.com> wrote:
> >
> > > Good idea. I submitted a pull request for the doc update here:
> > > https://github.com/apache/spark/pull/642
> > >
> > >
> > > On Sun, May 4, 2014 at 3:54 PM, Soren Macbeth <so...@yieldbot.com>
> > wrote:
> > >
> > > > Thanks for the reply!
> > > >
> > > > Ok, if that's the case, I'd recommend a note to that affect in the
> docs
> > > at
> > > > least.
> > > >
> > > > Just to give some more context here, I'm working on a Clojure DSL for
> > > Spark
> > > > called Flambo, which I plan to open source shortly. If I could I'd
> like
> > > to
> > > > focus on the initial bug that I hit.
> > > >
> > > > Exception in thread "main" org.apache.spark.SparkException: Job
> > aborted:
> > > > Exception while deserializing and fetching task:
> > > > com.esotericsoftware.kryo.KryoException:
> > > > java.lang.IllegalArgumentException: Can not set final
> > > > scala.collection.convert.Wrappers field
> > > > scala.collection.convert.Wrappers$SeqWrapper.$outer to
> > > > clojure.lang.PersistentVector
> > > > Serialization trace:
> > > > $outer (scala.collection.convert.Wrappers$SeqWrapper)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1020)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1018)
> > > > at
> > > >
> > > >
> > >
> >
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> > > > at
> > > > scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> > > > at org.apache.spark.scheduler.DAGScheduler.org
> > > >
> > $apache$spark$scheduler$DAGScheduler$$abortStage(DAGScheduler.scala:1018)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > > > at scala.Option.foreach(Option.scala:236)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler.processEvent(DAGScheduler.scala:604)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$start$1$$anon$2$$anonfun$receive$1.applyOrElse(DAGScheduler.scala:190)
> > > > at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
> > > > at akka.actor.ActorCell.invoke(ActorCell.scala:456)
> > > >
Re: bug using kryo as closure serializer
Posted by Reynold Xin <rx...@databricks.com>.
Thanks. Do you mind playing with chill-scala a little bit and see if it
actually works well for closures? One way to try is to hard code the
serializer to use Kryo with chill-scala, and then run through all the unit
tests.
If it works well, we can incorporate that in the next release (probably not
1.0, but after that).
On Sun, May 4, 2014 at 9:08 PM, Soren Macbeth <so...@yieldbot.com> wrote:
> fwiw, it seems like it wouldn't be very difficult to integrate chill-scala,
> since you're already chill-java and probably get kryo serialization of
> closures and all sorts of other scala stuff for free. All that would be
> needed would be to include the dependency and then update KryoSerializer to
> register the stuff in chill-scala.
>
> In that case, you could probably safely make kryo the default serializer,
> which I think would be desirable in general.
>
>
> On Sun, May 4, 2014 at 8:48 PM, Reynold Xin <rx...@databricks.com> wrote:
>
> > Good idea. I submitted a pull request for the doc update here:
> > https://github.com/apache/spark/pull/642
> >
> >
> > On Sun, May 4, 2014 at 3:54 PM, Soren Macbeth <so...@yieldbot.com>
> wrote:
> >
> > > Thanks for the reply!
> > >
> > > Ok, if that's the case, I'd recommend a note to that affect in the docs
> > at
> > > least.
> > >
> > > Just to give some more context here, I'm working on a Clojure DSL for
> > Spark
> > > called Flambo, which I plan to open source shortly. If I could I'd like
> > to
> > > focus on the initial bug that I hit.
> > >
> > > Exception in thread "main" org.apache.spark.SparkException: Job
> aborted:
> > > Exception while deserializing and fetching task:
> > > com.esotericsoftware.kryo.KryoException:
> > > java.lang.IllegalArgumentException: Can not set final
> > > scala.collection.convert.Wrappers field
> > > scala.collection.convert.Wrappers$SeqWrapper.$outer to
> > > clojure.lang.PersistentVector
> > > Serialization trace:
> > > $outer (scala.collection.convert.Wrappers$SeqWrapper)
> > > at
> > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1020)
> > > at
> > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1018)
> > > at
> > >
> > >
> >
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> > > at
> > > scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> > > at org.apache.spark.scheduler.DAGScheduler.org
> > >
> $apache$spark$scheduler$DAGScheduler$$abortStage(DAGScheduler.scala:1018)
> > > at
> > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > > at
> > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > > at scala.Option.foreach(Option.scala:236)
> > > at
> > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler.processEvent(DAGScheduler.scala:604)
> > > at
> > >
> > >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$start$1$$anon$2$$anonfun$receive$1.applyOrElse(DAGScheduler.scala:190)
> > > at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
> > > at akka.actor.ActorCell.invoke(ActorCell.scala:456)
> > > at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
> > > at akka.dispatch.Mailbox.run(Mailbox.scala:219)
> > > at
> > >
> > >
> >
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
> > > at
> > > scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> > > at
> > >
> > >
> >
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
> > > at
> > >
> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> > > at
> > >
> > >
> >
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> > >
> > > This happens immediately after all the tasks of a reduce stage complete
> > > successfully. Here is the function throwing the exception:
> > >
> > >
> > >
> >
> https://github.com/apache/spark/blob/4bc07eebbf5e2ea0c0b6f1642049515025d88d07/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala#L43
> > >
> > > This is where I get lost. From googling around, it seems that scala is
> > > trying to wrap the result of my task, which contain
> > > clojure.lang.PersistentVector objects in a scala collection, but I
> don't
> > > know why it's doing that. I have a registered kryo serializer for
> > > clojure.lang.PersistentVector.
> > >
> > > based on this line is looks like it's trying to use the closure
> > serializer,
> > > yet the expection thrown is from
> com.esotericsoftware.kryo.KryoException:
> > >
> > >
> > >
> >
> https://github.com/apache/spark/blob/4bc07eebbf5e2ea0c0b6f1642049515025d88d07/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala#L39
> > >
> > > Would storing my RDD as MEMORY_ONLY_SER prevent the closure serializer
> > from
> > > trying to deal with my clojure.lang.PeristentVector class?
> > >
> > > Where do I go from here?
> > >
> > >
> > > On Sun, May 4, 2014 at 12:50 PM, Reynold Xin <rx...@databricks.com>
> > wrote:
> > >
> > > > I added the config option to use the non-default serializer. However,
> > at
> > > > the time, Kryo fails serializing pretty much any closures so that
> > option
> > > > was never really used / recommended.
> > > >
> > > > Since then the Scala ecosystem has developed, and some other projects
> > are
> > > > starting to use Kryo to serialize more Scala data structures, so I
> > > wouldn't
> > > > be surprised if there is a way to work around this now. However, I
> > don't
> > > > have enough time to look into it at this point. If you do, please do
> > post
> > > > your findings. Thanks.
> > > >
> > > >
> > > >
> > > > On Sun, May 4, 2014 at 10:35 AM, Soren Macbeth <so...@yieldbot.com>
> > > wrote:
> > > >
> > > > > apologies for the cross-list posts, but I've gotten zero response
> in
> > > the
> > > > > user list and I guess this list is probably more appropriate.
> > > > >
> > > > > According to the documentation, using the KryoSerializer for
> closures
> > > is
> > > > > supported. However, when I try to set `spark.closure.serializer` to
> > > > > `org.apache.spark.serializer.KryoSerializer` thing fail pretty
> > > miserably.
> > > > >
> > > > > The first thing that happens it that is throws exceptions over and
> > over
> > > > > that it cannot locate my registrator class, which is located in my
> > > > assembly
> > > > > jar like so:
> > > > >
> > > > > 14/05/04 12:03:20 ERROR serializer.KryoSerializer: Failed to run
> > > > > spark.kryo.registrator
> > > > > java.lang.ClassNotFoundException: pickles.kryo.PicklesRegistrator
> > > > > at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> > > > > at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> > > > > at java.security.AccessController.doPrivileged(Native Method)
> > > > > at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> > > > > at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> > > > > at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> > > > > at java.lang.Class.forName0(Native Method)
> > > > > at java.lang.Class.forName(Class.java:270)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:63)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:61)
> > > > > at scala.Option.foreach(Option.scala:236)
> > > > > at
> > > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer.newKryo(KryoSerializer.scala:61)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializerInstance.<init>(KryoSerializer.scala:116)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer.newInstance(KryoSerializer.scala:79)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:180)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> > > > > at java.security.AccessController.doPrivileged(Native Method)
> > > > > at javax.security.auth.Subject.doAs(Subject.java:415)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> > > > > at
> > > > >
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> > > > > at
> > > org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> > > > > at java.lang.Thread.run(Thread.java:724)
> > > > >
> > > > > Now, I would expect it not to be able to find this class since it
> > > hasn't
> > > > > yet fetched my assembly jar to the executors. Once it does fetch my
> > > jar,
> > > > > those expections stop. Next, all the executor task die with the
> > > following
> > > > > exception:
> > > > >
> > > > > java.nio.ReadOnlyBufferException
> > > > > at java.nio.ByteBuffer.array(ByteBuffer.java:961)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializerInstance.deserialize(KryoSerializer.scala:136)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:193)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> > > > > at java.security.AccessController.doPrivileged(Native Method)
> > > > > at javax.security.auth.Subject.doAs(Subject.java:415)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> > > > > at
> > > > >
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> > > > > at
> > > org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> > > > > at
> > > > >
> > > > >
> > > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> > > > > at java.lang.Thread.run(Thread.java:724)
> > > > >
> > > > > AFAIK, I'm not doing anything out of the ordinary, just turning on
> > kryo
> > > > and
> > > > > using the registrator mechanism to register a couple custom
> > > serializers.
> > > > >
> > > > > The reason I tried turning on kryo for closure in the first place
> is
> > > > > because of a different bug that I was hitting during fetching and
> > > > > deserializing of tasks from my executors, which I detailed here:
> > > > >
> > > > >
> > > > >
> > > >
> > >
> >
> http://apache-spark-user-list.1001560.n3.nabble.com/Crazy-Kryo-Exception-td5257.html
> > > > >
> > > > > Here's hoping some on this list can help me track down what's
> > happening
> > > > as
> > > > > I didn't get a single reply on the user list.
> > > > >
> > > >
> > >
> >
>
Re: bug using kryo as closure serializer
Posted by Soren Macbeth <so...@yieldbot.com>.
fwiw, it seems like it wouldn't be very difficult to integrate chill-scala,
since you're already chill-java and probably get kryo serialization of
closures and all sorts of other scala stuff for free. All that would be
needed would be to include the dependency and then update KryoSerializer to
register the stuff in chill-scala.
In that case, you could probably safely make kryo the default serializer,
which I think would be desirable in general.
On Sun, May 4, 2014 at 8:48 PM, Reynold Xin <rx...@databricks.com> wrote:
> Good idea. I submitted a pull request for the doc update here:
> https://github.com/apache/spark/pull/642
>
>
> On Sun, May 4, 2014 at 3:54 PM, Soren Macbeth <so...@yieldbot.com> wrote:
>
> > Thanks for the reply!
> >
> > Ok, if that's the case, I'd recommend a note to that affect in the docs
> at
> > least.
> >
> > Just to give some more context here, I'm working on a Clojure DSL for
> Spark
> > called Flambo, which I plan to open source shortly. If I could I'd like
> to
> > focus on the initial bug that I hit.
> >
> > Exception in thread "main" org.apache.spark.SparkException: Job aborted:
> > Exception while deserializing and fetching task:
> > com.esotericsoftware.kryo.KryoException:
> > java.lang.IllegalArgumentException: Can not set final
> > scala.collection.convert.Wrappers field
> > scala.collection.convert.Wrappers$SeqWrapper.$outer to
> > clojure.lang.PersistentVector
> > Serialization trace:
> > $outer (scala.collection.convert.Wrappers$SeqWrapper)
> > at
> >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1020)
> > at
> >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1018)
> > at
> >
> >
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> > at
> > scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> > at org.apache.spark.scheduler.DAGScheduler.org
> > $apache$spark$scheduler$DAGScheduler$$abortStage(DAGScheduler.scala:1018)
> > at
> >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > at
> >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> > at scala.Option.foreach(Option.scala:236)
> > at
> >
> >
> org.apache.spark.scheduler.DAGScheduler.processEvent(DAGScheduler.scala:604)
> > at
> >
> >
> org.apache.spark.scheduler.DAGScheduler$$anonfun$start$1$$anon$2$$anonfun$receive$1.applyOrElse(DAGScheduler.scala:190)
> > at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
> > at akka.actor.ActorCell.invoke(ActorCell.scala:456)
> > at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
> > at akka.dispatch.Mailbox.run(Mailbox.scala:219)
> > at
> >
> >
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
> > at
> > scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> > at
> >
> >
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
> > at
> > scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> > at
> >
> >
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> >
> > This happens immediately after all the tasks of a reduce stage complete
> > successfully. Here is the function throwing the exception:
> >
> >
> >
> https://github.com/apache/spark/blob/4bc07eebbf5e2ea0c0b6f1642049515025d88d07/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala#L43
> >
> > This is where I get lost. From googling around, it seems that scala is
> > trying to wrap the result of my task, which contain
> > clojure.lang.PersistentVector objects in a scala collection, but I don't
> > know why it's doing that. I have a registered kryo serializer for
> > clojure.lang.PersistentVector.
> >
> > based on this line is looks like it's trying to use the closure
> serializer,
> > yet the expection thrown is from com.esotericsoftware.kryo.KryoException:
> >
> >
> >
> https://github.com/apache/spark/blob/4bc07eebbf5e2ea0c0b6f1642049515025d88d07/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala#L39
> >
> > Would storing my RDD as MEMORY_ONLY_SER prevent the closure serializer
> from
> > trying to deal with my clojure.lang.PeristentVector class?
> >
> > Where do I go from here?
> >
> >
> > On Sun, May 4, 2014 at 12:50 PM, Reynold Xin <rx...@databricks.com>
> wrote:
> >
> > > I added the config option to use the non-default serializer. However,
> at
> > > the time, Kryo fails serializing pretty much any closures so that
> option
> > > was never really used / recommended.
> > >
> > > Since then the Scala ecosystem has developed, and some other projects
> are
> > > starting to use Kryo to serialize more Scala data structures, so I
> > wouldn't
> > > be surprised if there is a way to work around this now. However, I
> don't
> > > have enough time to look into it at this point. If you do, please do
> post
> > > your findings. Thanks.
> > >
> > >
> > >
> > > On Sun, May 4, 2014 at 10:35 AM, Soren Macbeth <so...@yieldbot.com>
> > wrote:
> > >
> > > > apologies for the cross-list posts, but I've gotten zero response in
> > the
> > > > user list and I guess this list is probably more appropriate.
> > > >
> > > > According to the documentation, using the KryoSerializer for closures
> > is
> > > > supported. However, when I try to set `spark.closure.serializer` to
> > > > `org.apache.spark.serializer.KryoSerializer` thing fail pretty
> > miserably.
> > > >
> > > > The first thing that happens it that is throws exceptions over and
> over
> > > > that it cannot locate my registrator class, which is located in my
> > > assembly
> > > > jar like so:
> > > >
> > > > 14/05/04 12:03:20 ERROR serializer.KryoSerializer: Failed to run
> > > > spark.kryo.registrator
> > > > java.lang.ClassNotFoundException: pickles.kryo.PicklesRegistrator
> > > > at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> > > > at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> > > > at java.security.AccessController.doPrivileged(Native Method)
> > > > at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> > > > at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> > > > at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> > > > at java.lang.Class.forName0(Native Method)
> > > > at java.lang.Class.forName(Class.java:270)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:63)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:61)
> > > > at scala.Option.foreach(Option.scala:236)
> > > > at
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer.newKryo(KryoSerializer.scala:61)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializerInstance.<init>(KryoSerializer.scala:116)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer.newInstance(KryoSerializer.scala:79)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:180)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> > > > at java.security.AccessController.doPrivileged(Native Method)
> > > > at javax.security.auth.Subject.doAs(Subject.java:415)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> > > > at
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> > > > at
> > org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> > > > at
> > > >
> > > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> > > > at
> > > >
> > > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> > > > at java.lang.Thread.run(Thread.java:724)
> > > >
> > > > Now, I would expect it not to be able to find this class since it
> > hasn't
> > > > yet fetched my assembly jar to the executors. Once it does fetch my
> > jar,
> > > > those expections stop. Next, all the executor task die with the
> > following
> > > > exception:
> > > >
> > > > java.nio.ReadOnlyBufferException
> > > > at java.nio.ByteBuffer.array(ByteBuffer.java:961)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.serializer.KryoSerializerInstance.deserialize(KryoSerializer.scala:136)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:193)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> > > > at java.security.AccessController.doPrivileged(Native Method)
> > > > at javax.security.auth.Subject.doAs(Subject.java:415)
> > > > at
> > > >
> > > >
> > >
> >
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> > > > at
> > > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> > > > at
> > org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> > > > at
> > > >
> > > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> > > > at
> > > >
> > > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> > > > at java.lang.Thread.run(Thread.java:724)
> > > >
> > > > AFAIK, I'm not doing anything out of the ordinary, just turning on
> kryo
> > > and
> > > > using the registrator mechanism to register a couple custom
> > serializers.
> > > >
> > > > The reason I tried turning on kryo for closure in the first place is
> > > > because of a different bug that I was hitting during fetching and
> > > > deserializing of tasks from my executors, which I detailed here:
> > > >
> > > >
> > > >
> > >
> >
> http://apache-spark-user-list.1001560.n3.nabble.com/Crazy-Kryo-Exception-td5257.html
> > > >
> > > > Here's hoping some on this list can help me track down what's
> happening
> > > as
> > > > I didn't get a single reply on the user list.
> > > >
> > >
> >
>
Re: bug using kryo as closure serializer
Posted by Reynold Xin <rx...@databricks.com>.
Good idea. I submitted a pull request for the doc update here:
https://github.com/apache/spark/pull/642
On Sun, May 4, 2014 at 3:54 PM, Soren Macbeth <so...@yieldbot.com> wrote:
> Thanks for the reply!
>
> Ok, if that's the case, I'd recommend a note to that affect in the docs at
> least.
>
> Just to give some more context here, I'm working on a Clojure DSL for Spark
> called Flambo, which I plan to open source shortly. If I could I'd like to
> focus on the initial bug that I hit.
>
> Exception in thread "main" org.apache.spark.SparkException: Job aborted:
> Exception while deserializing and fetching task:
> com.esotericsoftware.kryo.KryoException:
> java.lang.IllegalArgumentException: Can not set final
> scala.collection.convert.Wrappers field
> scala.collection.convert.Wrappers$SeqWrapper.$outer to
> clojure.lang.PersistentVector
> Serialization trace:
> $outer (scala.collection.convert.Wrappers$SeqWrapper)
> at
>
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1020)
> at
>
> org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1018)
> at
>
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> at
> scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> at org.apache.spark.scheduler.DAGScheduler.org
> $apache$spark$scheduler$DAGScheduler$$abortStage(DAGScheduler.scala:1018)
> at
>
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> at
>
> org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
> at scala.Option.foreach(Option.scala:236)
> at
>
> org.apache.spark.scheduler.DAGScheduler.processEvent(DAGScheduler.scala:604)
> at
>
> org.apache.spark.scheduler.DAGScheduler$$anonfun$start$1$$anon$2$$anonfun$receive$1.applyOrElse(DAGScheduler.scala:190)
> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
> at akka.actor.ActorCell.invoke(ActorCell.scala:456)
> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
> at akka.dispatch.Mailbox.run(Mailbox.scala:219)
> at
>
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
> at
> scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> at
>
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
> at
> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> at
>
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
>
> This happens immediately after all the tasks of a reduce stage complete
> successfully. Here is the function throwing the exception:
>
>
> https://github.com/apache/spark/blob/4bc07eebbf5e2ea0c0b6f1642049515025d88d07/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala#L43
>
> This is where I get lost. From googling around, it seems that scala is
> trying to wrap the result of my task, which contain
> clojure.lang.PersistentVector objects in a scala collection, but I don't
> know why it's doing that. I have a registered kryo serializer for
> clojure.lang.PersistentVector.
>
> based on this line is looks like it's trying to use the closure serializer,
> yet the expection thrown is from com.esotericsoftware.kryo.KryoException:
>
>
> https://github.com/apache/spark/blob/4bc07eebbf5e2ea0c0b6f1642049515025d88d07/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala#L39
>
> Would storing my RDD as MEMORY_ONLY_SER prevent the closure serializer from
> trying to deal with my clojure.lang.PeristentVector class?
>
> Where do I go from here?
>
>
> On Sun, May 4, 2014 at 12:50 PM, Reynold Xin <rx...@databricks.com> wrote:
>
> > I added the config option to use the non-default serializer. However, at
> > the time, Kryo fails serializing pretty much any closures so that option
> > was never really used / recommended.
> >
> > Since then the Scala ecosystem has developed, and some other projects are
> > starting to use Kryo to serialize more Scala data structures, so I
> wouldn't
> > be surprised if there is a way to work around this now. However, I don't
> > have enough time to look into it at this point. If you do, please do post
> > your findings. Thanks.
> >
> >
> >
> > On Sun, May 4, 2014 at 10:35 AM, Soren Macbeth <so...@yieldbot.com>
> wrote:
> >
> > > apologies for the cross-list posts, but I've gotten zero response in
> the
> > > user list and I guess this list is probably more appropriate.
> > >
> > > According to the documentation, using the KryoSerializer for closures
> is
> > > supported. However, when I try to set `spark.closure.serializer` to
> > > `org.apache.spark.serializer.KryoSerializer` thing fail pretty
> miserably.
> > >
> > > The first thing that happens it that is throws exceptions over and over
> > > that it cannot locate my registrator class, which is located in my
> > assembly
> > > jar like so:
> > >
> > > 14/05/04 12:03:20 ERROR serializer.KryoSerializer: Failed to run
> > > spark.kryo.registrator
> > > java.lang.ClassNotFoundException: pickles.kryo.PicklesRegistrator
> > > at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> > > at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> > > at java.security.AccessController.doPrivileged(Native Method)
> > > at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> > > at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> > > at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> > > at java.lang.Class.forName0(Native Method)
> > > at java.lang.Class.forName(Class.java:270)
> > > at
> > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:63)
> > > at
> > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:61)
> > > at scala.Option.foreach(Option.scala:236)
> > > at
> > >
> >
> org.apache.spark.serializer.KryoSerializer.newKryo(KryoSerializer.scala:61)
> > > at
> > >
> > >
> >
> org.apache.spark.serializer.KryoSerializerInstance.<init>(KryoSerializer.scala:116)
> > > at
> > >
> > >
> >
> org.apache.spark.serializer.KryoSerializer.newInstance(KryoSerializer.scala:79)
> > > at
> > >
> > >
> >
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:180)
> > > at
> > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> > > at
> > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> > > at java.security.AccessController.doPrivileged(Native Method)
> > > at javax.security.auth.Subject.doAs(Subject.java:415)
> > > at
> > >
> > >
> >
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> > > at
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> > > at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> > > at
> > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> > > at
> > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> > > at java.lang.Thread.run(Thread.java:724)
> > >
> > > Now, I would expect it not to be able to find this class since it
> hasn't
> > > yet fetched my assembly jar to the executors. Once it does fetch my
> jar,
> > > those expections stop. Next, all the executor task die with the
> following
> > > exception:
> > >
> > > java.nio.ReadOnlyBufferException
> > > at java.nio.ByteBuffer.array(ByteBuffer.java:961)
> > > at
> > >
> > >
> >
> org.apache.spark.serializer.KryoSerializerInstance.deserialize(KryoSerializer.scala:136)
> > > at
> > >
> > >
> >
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:193)
> > > at
> > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> > > at
> > >
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> > > at java.security.AccessController.doPrivileged(Native Method)
> > > at javax.security.auth.Subject.doAs(Subject.java:415)
> > > at
> > >
> > >
> >
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> > > at
> > >
> >
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> > > at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> > > at
> > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> > > at
> > >
> > >
> >
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> > > at java.lang.Thread.run(Thread.java:724)
> > >
> > > AFAIK, I'm not doing anything out of the ordinary, just turning on kryo
> > and
> > > using the registrator mechanism to register a couple custom
> serializers.
> > >
> > > The reason I tried turning on kryo for closure in the first place is
> > > because of a different bug that I was hitting during fetching and
> > > deserializing of tasks from my executors, which I detailed here:
> > >
> > >
> > >
> >
> http://apache-spark-user-list.1001560.n3.nabble.com/Crazy-Kryo-Exception-td5257.html
> > >
> > > Here's hoping some on this list can help me track down what's happening
> > as
> > > I didn't get a single reply on the user list.
> > >
> >
>
Re: bug using kryo as closure serializer
Posted by Soren Macbeth <so...@yieldbot.com>.
Thanks for the reply!
Ok, if that's the case, I'd recommend a note to that affect in the docs at
least.
Just to give some more context here, I'm working on a Clojure DSL for Spark
called Flambo, which I plan to open source shortly. If I could I'd like to
focus on the initial bug that I hit.
Exception in thread "main" org.apache.spark.SparkException: Job aborted:
Exception while deserializing and fetching task:
com.esotericsoftware.kryo.KryoException:
java.lang.IllegalArgumentException: Can not set final
scala.collection.convert.Wrappers field
scala.collection.convert.Wrappers$SeqWrapper.$outer to
clojure.lang.PersistentVector
Serialization trace:
$outer (scala.collection.convert.Wrappers$SeqWrapper)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1020)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$org$apache$spark$scheduler$DAGScheduler$$abortStage$1.apply(DAGScheduler.scala:1018)
at
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at
scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at org.apache.spark.scheduler.DAGScheduler.org
$apache$spark$scheduler$DAGScheduler$$abortStage(DAGScheduler.scala:1018)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$processEvent$10.apply(DAGScheduler.scala:604)
at scala.Option.foreach(Option.scala:236)
at
org.apache.spark.scheduler.DAGScheduler.processEvent(DAGScheduler.scala:604)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$start$1$$anon$2$$anonfun$receive$1.applyOrElse(DAGScheduler.scala:190)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
at akka.actor.ActorCell.invoke(ActorCell.scala:456)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
at akka.dispatch.Mailbox.run(Mailbox.scala:219)
at
akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
at
scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at
scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
This happens immediately after all the tasks of a reduce stage complete
successfully. Here is the function throwing the exception:
https://github.com/apache/spark/blob/4bc07eebbf5e2ea0c0b6f1642049515025d88d07/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala#L43
This is where I get lost. From googling around, it seems that scala is
trying to wrap the result of my task, which contain
clojure.lang.PersistentVector objects in a scala collection, but I don't
know why it's doing that. I have a registered kryo serializer for
clojure.lang.PersistentVector.
based on this line is looks like it's trying to use the closure serializer,
yet the expection thrown is from com.esotericsoftware.kryo.KryoException:
https://github.com/apache/spark/blob/4bc07eebbf5e2ea0c0b6f1642049515025d88d07/core/src/main/scala/org/apache/spark/scheduler/TaskResultGetter.scala#L39
Would storing my RDD as MEMORY_ONLY_SER prevent the closure serializer from
trying to deal with my clojure.lang.PeristentVector class?
Where do I go from here?
On Sun, May 4, 2014 at 12:50 PM, Reynold Xin <rx...@databricks.com> wrote:
> I added the config option to use the non-default serializer. However, at
> the time, Kryo fails serializing pretty much any closures so that option
> was never really used / recommended.
>
> Since then the Scala ecosystem has developed, and some other projects are
> starting to use Kryo to serialize more Scala data structures, so I wouldn't
> be surprised if there is a way to work around this now. However, I don't
> have enough time to look into it at this point. If you do, please do post
> your findings. Thanks.
>
>
>
> On Sun, May 4, 2014 at 10:35 AM, Soren Macbeth <so...@yieldbot.com> wrote:
>
> > apologies for the cross-list posts, but I've gotten zero response in the
> > user list and I guess this list is probably more appropriate.
> >
> > According to the documentation, using the KryoSerializer for closures is
> > supported. However, when I try to set `spark.closure.serializer` to
> > `org.apache.spark.serializer.KryoSerializer` thing fail pretty miserably.
> >
> > The first thing that happens it that is throws exceptions over and over
> > that it cannot locate my registrator class, which is located in my
> assembly
> > jar like so:
> >
> > 14/05/04 12:03:20 ERROR serializer.KryoSerializer: Failed to run
> > spark.kryo.registrator
> > java.lang.ClassNotFoundException: pickles.kryo.PicklesRegistrator
> > at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> > at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> > at java.security.AccessController.doPrivileged(Native Method)
> > at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> > at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> > at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> > at java.lang.Class.forName0(Native Method)
> > at java.lang.Class.forName(Class.java:270)
> > at
> >
> >
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:63)
> > at
> >
> >
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:61)
> > at scala.Option.foreach(Option.scala:236)
> > at
> >
> org.apache.spark.serializer.KryoSerializer.newKryo(KryoSerializer.scala:61)
> > at
> >
> >
> org.apache.spark.serializer.KryoSerializerInstance.<init>(KryoSerializer.scala:116)
> > at
> >
> >
> org.apache.spark.serializer.KryoSerializer.newInstance(KryoSerializer.scala:79)
> > at
> >
> >
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:180)
> > at
> >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> > at
> >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> > at java.security.AccessController.doPrivileged(Native Method)
> > at javax.security.auth.Subject.doAs(Subject.java:415)
> > at
> >
> >
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> > at
> >
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> > at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> > at
> >
> >
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> > at
> >
> >
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> > at java.lang.Thread.run(Thread.java:724)
> >
> > Now, I would expect it not to be able to find this class since it hasn't
> > yet fetched my assembly jar to the executors. Once it does fetch my jar,
> > those expections stop. Next, all the executor task die with the following
> > exception:
> >
> > java.nio.ReadOnlyBufferException
> > at java.nio.ByteBuffer.array(ByteBuffer.java:961)
> > at
> >
> >
> org.apache.spark.serializer.KryoSerializerInstance.deserialize(KryoSerializer.scala:136)
> > at
> >
> >
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:193)
> > at
> >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> > at
> >
> >
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> > at java.security.AccessController.doPrivileged(Native Method)
> > at javax.security.auth.Subject.doAs(Subject.java:415)
> > at
> >
> >
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> > at
> >
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> > at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> > at
> >
> >
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> > at
> >
> >
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> > at java.lang.Thread.run(Thread.java:724)
> >
> > AFAIK, I'm not doing anything out of the ordinary, just turning on kryo
> and
> > using the registrator mechanism to register a couple custom serializers.
> >
> > The reason I tried turning on kryo for closure in the first place is
> > because of a different bug that I was hitting during fetching and
> > deserializing of tasks from my executors, which I detailed here:
> >
> >
> >
> http://apache-spark-user-list.1001560.n3.nabble.com/Crazy-Kryo-Exception-td5257.html
> >
> > Here's hoping some on this list can help me track down what's happening
> as
> > I didn't get a single reply on the user list.
> >
>
Re: bug using kryo as closure serializer
Posted by Reynold Xin <rx...@databricks.com>.
I added the config option to use the non-default serializer. However, at
the time, Kryo fails serializing pretty much any closures so that option
was never really used / recommended.
Since then the Scala ecosystem has developed, and some other projects are
starting to use Kryo to serialize more Scala data structures, so I wouldn't
be surprised if there is a way to work around this now. However, I don't
have enough time to look into it at this point. If you do, please do post
your findings. Thanks.
On Sun, May 4, 2014 at 10:35 AM, Soren Macbeth <so...@yieldbot.com> wrote:
> apologies for the cross-list posts, but I've gotten zero response in the
> user list and I guess this list is probably more appropriate.
>
> According to the documentation, using the KryoSerializer for closures is
> supported. However, when I try to set `spark.closure.serializer` to
> `org.apache.spark.serializer.KryoSerializer` thing fail pretty miserably.
>
> The first thing that happens it that is throws exceptions over and over
> that it cannot locate my registrator class, which is located in my assembly
> jar like so:
>
> 14/05/04 12:03:20 ERROR serializer.KryoSerializer: Failed to run
> spark.kryo.registrator
> java.lang.ClassNotFoundException: pickles.kryo.PicklesRegistrator
> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> at java.security.AccessController.doPrivileged(Native Method)
> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Native Method)
> at java.lang.Class.forName(Class.java:270)
> at
>
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:63)
> at
>
> org.apache.spark.serializer.KryoSerializer$$anonfun$newKryo$2.apply(KryoSerializer.scala:61)
> at scala.Option.foreach(Option.scala:236)
> at
> org.apache.spark.serializer.KryoSerializer.newKryo(KryoSerializer.scala:61)
> at
>
> org.apache.spark.serializer.KryoSerializerInstance.<init>(KryoSerializer.scala:116)
> at
>
> org.apache.spark.serializer.KryoSerializer.newInstance(KryoSerializer.scala:79)
> at
>
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:180)
> at
>
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> at
>
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> at java.security.AccessController.doPrivileged(Native Method)
> at javax.security.auth.Subject.doAs(Subject.java:415)
> at
>
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> at
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> at
>
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
>
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:724)
>
> Now, I would expect it not to be able to find this class since it hasn't
> yet fetched my assembly jar to the executors. Once it does fetch my jar,
> those expections stop. Next, all the executor task die with the following
> exception:
>
> java.nio.ReadOnlyBufferException
> at java.nio.ByteBuffer.array(ByteBuffer.java:961)
> at
>
> org.apache.spark.serializer.KryoSerializerInstance.deserialize(KryoSerializer.scala:136)
> at
>
> org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:193)
> at
>
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
> at
>
> org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
> at java.security.AccessController.doPrivileged(Native Method)
> at javax.security.auth.Subject.doAs(Subject.java:415)
> at
>
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1438)
> at
> org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
> at
>
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
> at
>
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> at java.lang.Thread.run(Thread.java:724)
>
> AFAIK, I'm not doing anything out of the ordinary, just turning on kryo and
> using the registrator mechanism to register a couple custom serializers.
>
> The reason I tried turning on kryo for closure in the first place is
> because of a different bug that I was hitting during fetching and
> deserializing of tasks from my executors, which I detailed here:
>
>
> http://apache-spark-user-list.1001560.n3.nabble.com/Crazy-Kryo-Exception-td5257.html
>
> Here's hoping some on this list can help me track down what's happening as
> I didn't get a single reply on the user list.
>