You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by _soumya_ <so...@gmail.com> on 2014/10/16 17:39:02 UTC
Running an action inside a loop across multiple RDDs +
java.io.NotSerializableException
Hi, my programming model requires me to generate multiple RDDs for various
datasets across a single run and then run an action on it - E.g.
MyFunc myFunc = ... //It implements VoidFunction
//set some extra variables - all serializable
...
for (JavaRDD<String> rdd: rddList) {
...
sc.foreach(myFunc);
}
The problem I'm seeing is that after the first run of the loop - which
succeeds on foreach, the second one fails with
java.io.NotSerializableException for a specific object I'm setting. In my
particular case, the object contains a reference to
org.apache.hadoop.conf.Configuration. Question is:
1. Why does this succeed the first time, and fail the second?
2. Any alternatives to this programming model?
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Running-an-action-inside-a-loop-across-multiple-RDDs-java-io-NotSerializableException-tp16580.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: Running an action inside a loop across multiple RDDs +
java.io.NotSerializableException
Posted by _soumya_ <so...@gmail.com>.
Excuse me - the line inside the loop should read: rdd.foreach(myFunc) - not
sc.
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Running-an-action-inside-a-loop-across-multiple-RDDs-java-io-NotSerializableException-tp16580p16581.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: Running an action inside a loop across multiple RDDs +
java.io.NotSerializableException
Posted by _soumya_ <so...@gmail.com>.
Sorry - I'll furnish some details below. However, union is not an option for
the business logic I have. The function will generate a specific file based
on a variable passed in as the setter for the function. This variable
changes with each RDD. I annotated the log line where the first run
succeeds.
Logs output to give you some context:
...
14/10/16 15:32:25 INFO SparkContext: Starting job: count at
GenerateJSONContent.java:145
14/10/16 15:32:25 INFO DAGScheduler: Got job 0 (count at
GenerateJSONContent.java:145) with 1 output partitions (allowLocal=false)
14/10/16 15:32:25 INFO DAGScheduler: Final stage: Stage 0(count at
GenerateJSONContent.java:145)
14/10/16 15:32:25 INFO DAGScheduler: Parents of final stage: List()
14/10/16 15:32:25 INFO DAGScheduler: Missing parents: List()
14/10/16 15:32:25 INFO DAGScheduler: Submitting Stage 0
(ParallelCollectionRDD[2] at parallelize at GenerateJSONContent.java:82),
which has no missing parents
14/10/16 15:32:25 INFO DAGScheduler: Submitting 1 missing tasks from Stage 0
(ParallelCollectionRDD[2] at parallelize at GenerateJSONContent.java:82)
14/10/16 15:32:25 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks
14/10/16 15:32:25 INFO TaskSetManager: Starting task 0.0:0 as TID 0 on
executor localhost: localhost (PROCESS_LOCAL)
14/10/16 15:32:25 INFO TaskSetManager: Serialized task 0.0:0 as 2048 bytes
in 3 ms
14/10/16 15:32:25 INFO Executor: Running task ID 0
14/10/16 15:32:25 INFO Executor: Fetching
http://172.16.1.204:42232/jars/rickshaw-spark-0.0.1-SNAPSHOT.jar with
timestamp 1413473544838
14/10/16 15:32:25 INFO Utils: Fetching
http://172.16.1.204:42232/jars/rickshaw-spark-0.0.1-SNAPSHOT.jar to
/tmp/fetchFileTemp7432615579770034188.tmp
14/10/16 15:32:26 INFO Executor: Adding
file:/tmp/spark-a471145f-ab44-447b-b48c-bb499024d756/rickshaw-spark-0.0.1-SNAPSHOT.jar
to class loader
14/10/16 15:32:26 INFO Executor: Serialized size of result for 0 is 597
14/10/16 15:32:26 INFO Executor: Sending result for 0 directly to driver
14/10/16 15:32:26 INFO Executor: Finished task ID 0
14/10/16 15:32:26 INFO DAGScheduler: Completed ResultTask(0, 0)
14/10/16 15:32:26 INFO DAGScheduler: Stage 0 (count at
GenerateJSONContent.java:145) finished in 0.791 s
14/10/16 15:32:26 INFO SparkContext: Job finished: count at
GenerateJSONContent.java:145, took 0.889171151 s
14/10/16 15:32:26 INFO GenerateJSONContent: Running for tag
:PublicationProductId
14/10/16 15:32:26 INFO TaskSetManager: Finished TID 0 in 778 ms on localhost
(progress: 1/1)
14/10/16 15:32:26 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks
have all completed, from pool
14/10/16 15:32:26 INFO SparkContext: Starting job: foreach at
GenerateJSONContent.java:156
14/10/16 15:32:26 INFO DAGScheduler: Got job 1 (foreach at
GenerateJSONContent.java:156) with 1 output partitions (allowLocal=false)
14/10/16 15:32:26 INFO DAGScheduler: Final stage: Stage 1(foreach at
GenerateJSONContent.java:156)
14/10/16 15:32:26 INFO DAGScheduler: Parents of final stage: List()
14/10/16 15:32:26 INFO DAGScheduler: Missing parents: List()
14/10/16 15:32:26 INFO DAGScheduler: Submitting Stage 1
(ParallelCollectionRDD[2] at parallelize at GenerateJSONContent.java:82),
which has no missing parents
14/10/16 15:32:26 INFO DAGScheduler: Submitting 1 missing tasks from Stage 1
(ParallelCollectionRDD[2] at parallelize at GenerateJSONContent.java:82)
14/10/16 15:32:26 INFO TaskSchedulerImpl: Adding task set 1.0 with 1 tasks
14/10/16 15:32:26 INFO TaskSetManager: Starting task 1.0:0 as TID 1 on
executor localhost: localhost (PROCESS_LOCAL)
14/10/16 15:32:26 INFO TaskSetManager: Serialized task 1.0:0 as 2792 bytes
in 1 ms
14/10/16 15:32:26 INFO Executor: Running task ID 1
14/10/16 15:32:35 INFO Executor: Serialized size of result for 1 is 559
14/10/16 15:32:35 INFO Executor: Sending result for 1 directly to driver
14/10/16 15:32:35 INFO Executor: Finished task ID 1
14/10/16 15:32:35 INFO DAGScheduler: Completed ResultTask(1, 0)
14/10/16 15:32:35 INFO DAGScheduler: Stage 1 (foreach at
GenerateJSONContent.java:156) finished in 9.098 s
14/10/16 15:32:35 INFO SparkContext: Job finished: foreach at
GenerateJSONContent.java:156, took 9.112736939 s
14/10/16 15:32:35 INFO TaskSetManager: Finished TID 1 in 9095 ms on
localhost (progress: 1/1)
14/10/16 15:32:35 INFO TaskSchedulerImpl: Removed TaskSet 1.0, whose tasks
have all completed, from pool
14/10/16 15:32:36 INFO SparkContext: Starting job: count at
GenerateJSONContent.java:145
14/10/16 15:32:36 INFO DAGScheduler: Got job 2 (count at
GenerateJSONContent.java:145) with 1 output partitions (allowLocal=false)
14/10/16 15:32:36 INFO DAGScheduler: Final stage: Stage 2(count at
GenerateJSONContent.java:145)
14/10/16 15:32:36 INFO DAGScheduler: Parents of final stage: List()
14/10/16 15:32:36 INFO DAGScheduler: Missing parents: List()
14/10/16 15:32:36 INFO DAGScheduler: Submitting Stage 2
(ParallelCollectionRDD[1] at parallelize at GenerateJSONContent.java:82),
which has no missing parents
14/10/16 15:32:36 INFO DAGScheduler: Submitting 1 missing tasks from Stage 2
(ParallelCollectionRDD[1] at parallelize at GenerateJSONContent.java:82)
14/10/16 15:32:36 INFO TaskSchedulerImpl: Adding task set 2.0 with 1 tasks
14/10/16 15:32:36 INFO TaskSetManager: Starting task 2.0:0 as TID 2 on
executor localhost: localhost (PROCESS_LOCAL)
14/10/16 15:32:36 INFO TaskSetManager: Serialized task 2.0:0 as 2048 bytes
in 0 ms
14/10/16 15:32:36 INFO Executor: Running task ID 2
14/10/16 15:32:36 INFO Executor: Serialized size of result for 2 is 597
14/10/16 15:32:36 INFO Executor: Sending result for 2 directly to driver
14/10/16 15:32:36 INFO Executor: Finished task ID 2
14/10/16 15:32:36 INFO DAGScheduler: Completed ResultTask(2, 0)
14/10/16 15:32:36 INFO DAGScheduler: Stage 2 (count at
GenerateJSONContent.java:145) finished in 0.008 s
14/10/16 15:32:36 INFO SparkContext: Job finished: count at
GenerateJSONContent.java:145, took 0.025463494 s <<<<<< Loop 1 succeeds
14/10/16 15:32:36 INFO GenerateJSONContent: Running for tag :CrossRevisionId
14/10/16 15:32:36 INFO SparkContext: Starting job: foreach at
GenerateJSONContent.java:156
14/10/16 15:32:36 INFO DAGScheduler: Got job 3 (foreach at
GenerateJSONContent.java:156) with 1 output partitions (allowLocal=false)
14/10/16 15:32:36 INFO DAGScheduler: Final stage: Stage 3(foreach at
GenerateJSONContent.java:156)
14/10/16 15:32:36 INFO DAGScheduler: Parents of final stage: List()
14/10/16 15:32:36 INFO DAGScheduler: Missing parents: List()
14/10/16 15:32:36 INFO DAGScheduler: Submitting Stage 3
(ParallelCollectionRDD[1] at parallelize at GenerateJSONContent.java:82),
which has no missing parents
14/10/16 15:32:36 INFO TaskSetManager: Finished TID 2 in 15 ms on localhost
(progress: 1/1)
14/10/16 15:32:36 INFO TaskSchedulerImpl: Removed TaskSet 2.0, whose tasks
have all completed, from pool
14/10/16 15:32:37 INFO DAGScheduler: Failed to run foreach at
GenerateJSONContent.java:156
Exception in thread "main" org.apache.spark.SparkException: Job aborted due
to stage failure: Task not serializable: java.io.NotSerializableException:
org.apache.hadoop.conf.Configuration
at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1033)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1017)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1015)
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.abortStage(DAGScheduler.scala:1015)
at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$submitMissingTasks(DAGScheduler.scala:770)
at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$submitStage(DAGScheduler.scala:713)
at
org.apache.spark.scheduler.DAGScheduler.handleJobSubmitted(DAGScheduler.scala:697)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1176)
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)
More code context follows:
...
Map<String, JavaRDD<String>> idLists = new HashMap<>();
//Populate the map above based on string tags
....
String filePathSuffix = null;
GenerateJSONContentFunction jsonContent;
switch (options.getType()) {
case "opinions":
jsonContent = (GenerateJSONContentFunction)
context.getBean("AssetOpinionsJsonContent");
filePathSuffix += "/reviews/product/opinions/json/";
esConf.setQueryString(
esUtil.getTagDetailsForBusinessQueryString(options.getBusinessId())
);
break;
case "summary":
jsonContent = (GenerateJSONContentFunction)
context.getBean("AssetSummaryJsonContent");
filePathSuffix += "/reviews/product/summary/json/";
break;
case "consumer":
jsonContent = (GenerateJSONContentFunction)
context.getBean("ConsumerProfileJsonContent");
filePathSuffix += "/consumers/profiles/json/";
break;
default: throw new RuntimeException("Invalid type parameter
passed. Current values allows ('opinions','summary','consumer').");
}
jsonContent.setHdfsUtil(hdfsUtil);
for (String tagName: idLists.keySet()) {
if (idLists.get(tagName).count() == 0) {
continue;
}
LOG.info("Running for tag :" + tagName);
String runId = UUID.randomUUID().toString();
String filePath = "/static_output/" + runId + "/" +
businessIdHash + "/" + locale + filePathSuffix;
jsonContent.setTagName(tagName);
jsonContent.setOutputPath(filePath);
idLists.get(tagName).foreach(
jsonContent
);
...
}
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Running-an-action-inside-a-loop-across-multiple-RDDs-java-io-NotSerializableException-tp16580p16597.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: Running an action inside a loop across multiple RDDs + java.io.NotSerializableException
Posted by Cheng Lian <li...@gmail.com>.
You can first union them into a single RDD and then call |foreach|. In
Scala:
|rddList.reduce(_.union(_)).foreach(myFunc)
|
For the serialization issue, I don’t have any clue unless more code can
be shared.
On 10/16/14 11:39 PM, /soumya/ wrote:
> Hi, my programming model requires me to generate multiple RDDs for various
> datasets across a single run and then run an action on it - E.g.
>
> MyFunc myFunc = ... //It implements VoidFunction
> //set some extra variables - all serializable
> ...
> for (JavaRDD<String> rdd: rddList) {
> ...
> sc.foreach(myFunc);
>
> }
>
> The problem I'm seeing is that after the first run of the loop - which
> succeeds on foreach, the second one fails with
> java.io.NotSerializableException for a specific object I'm setting. In my
> particular case, the object contains a reference to
> org.apache.hadoop.conf.Configuration. Question is:
>
> 1. Why does this succeed the first time, and fail the second?
> 2. Any alternatives to this programming model?
>
>
>
>
> --
> View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/Running-an-action-inside-a-loop-across-multiple-RDDs-java-io-NotSerializableException-tp16580.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>