You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Josh Rosen (JIRA)" <ji...@apache.org> on 2014/12/01 02:26:12 UTC
[jira] [Commented] (SPARK-3926) result of JavaRDD collectAsMap() is
not serializable
[ https://issues.apache.org/jira/browse/SPARK-3926?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14229318#comment-14229318 ]
Josh Rosen commented on SPARK-3926:
-----------------------------------
Just saw a bug report on the mailing list that looks possily related to this: http://apache-spark-user-list.1001560.n3.nabble.com/java-io-InvalidClassException-org-apache-spark-api-java-JavaUtils-SerializableMapWrapper-no-valid-cor-td20034.html
{code}
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in
stage 815081.0 failed 4 times, most recent failure: Lost task 0.3 in stage
815081.0 (TID 4751, ns2.xxxxx.net): java.io.InvalidClassException:
org.apache.spark.api.java.JavaUtils$SerializableMapWrapper; no valid
constructor
at
java.io.ObjectStreamClass$ExceptionInfo.newInvalidClassException(ObjectStreamClass.java:150)
at
java.io.ObjectStreamClass.checkDeserialize(ObjectStreamClass.java:768)
at
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1775)
at
java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)
at
org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:62)
at
org.apache.spark.broadcast.TorrentBroadcast$.unBlockifyObject(TorrentBroadcast.scala:216)
at
org.apache.spark.broadcast.TorrentBroadcast$$anonfun$readBroadcastBlock$1.apply(TorrentBroadcast.scala:177)
at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1000)
at
org.apache.spark.broadcast.TorrentBroadcast.readBroadcastBlock(TorrentBroadcast.scala:164)
at
org.apache.spark.broadcast.TorrentBroadcast._value$lzycompute(TorrentBroadcast.scala:64)
at
org.apache.spark.broadcast.TorrentBroadcast._value(TorrentBroadcast.scala:64)
at
org.apache.spark.broadcast.TorrentBroadcast.getValue(TorrentBroadcast.scala:87)
at
com.xxxxxxx.common.chores.kafka.kafkaListenerChore$4$1.call(kafkaListenerChore.java:418)
at
com.xxxxxxx.common.chores.kafka.kafkaListenerChore$4$1.call(kafkaListenerChore.java:406)
at
org.apache.spark.api.java.JavaRDDLike$$anonfun$foreachPartition$1.apply(JavaRDDLike.scala:195)
at
org.apache.spark.api.java.JavaRDDLike$$anonfun$foreachPartition$1.apply(JavaRDDLike.scala:195)
at
org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1.apply(RDD.scala:775)
at
org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1.apply(RDD.scala:775)
at
org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1314)
at
org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1314)
at
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
at org.apache.spark.scheduler.Task.run(Task.scala:56)
at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
{code}
Any idea what's going on here?
> result of JavaRDD collectAsMap() is not serializable
> ----------------------------------------------------
>
> Key: SPARK-3926
> URL: https://issues.apache.org/jira/browse/SPARK-3926
> Project: Spark
> Issue Type: Bug
> Components: Java API
> Affects Versions: 1.0.2, 1.1.0, 1.2.0
> Environment: CentOS / Spark 1.1 / Hadoop Hortonworks 2.4.0.2.1.2.0-402
> Reporter: Antoine Amend
> Assignee: Sean Owen
> Fix For: 1.1.1, 1.2.0
>
>
> Using the Java API, I want to collect the result of a RDD<String, String> as a HashMap using collectAsMap function:
> Map<String, String> map = myJavaRDD.collectAsMap();
> This works fine, but when passing this map to another function, such as...
> myOtherJavaRDD.mapToPair(new CustomFunction(map))
> ...this leads to the following error:
> Exception in thread "main" org.apache.spark.SparkException: Task not serializable
> at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166)
> at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158)
> at org.apache.spark.SparkContext.clean(SparkContext.scala:1242)
> at org.apache.spark.rdd.RDD.map(RDD.scala:270)
> at org.apache.spark.api.java.JavaRDDLike$class.mapToPair(JavaRDDLike.scala:99)
> at org.apache.spark.api.java.JavaPairRDD.mapToPair(JavaPairRDD.scala:44)
> ../.. MY CLASS ../..
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
> at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:606)
> at org.apache.spark.deploy.SparkSubmit$.launch(SparkSubmit.scala:328)
> at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:75)
> at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
> Caused by: java.io.NotSerializableException: scala.collection.convert.Wrappers$MapWrapper
> at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1183)
> at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
> at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
> at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
> at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
> at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
> at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
> at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
> at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
> at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347)
> at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:42)
> at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:73)
> at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164)
> This seems to be due to WrapAsJava.scala being non serializable
> ../..
> implicit def mapAsJavaMap[A, B](m: Map[A, B]): ju.Map[A, B] = m match {
> //case JConcurrentMapWrapper(wrapped) => wrapped
> case JMapWrapper(wrapped) => wrapped.asInstanceOf[ju.Map[A, B]]
> case _ => new MapWrapper(m)
> }
> ../..
> The workaround is to manually wrapper this map into another one (serialized)
> Map<String, String> map = myJavaRDD.collectAsMap();
> Map<String, String> tmp = new HashMap<String, String>(map);
> myOtherJavaRDD.mapToPair(new CustomFunction(tmp))
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org