That stack trace is quite similar to the one that is generated when trying
to do a collect within a closure. In this case, it feels "wrong" to
collect in a closure, but I wonder what's reason behind the NPE.
Curious to know whether they are related.
Here's a very simple example:
rrd1.flatMap(x=> rrd2.collect.flatMap(y=> List(y,x)))
res7: org.apache.spark.rdd.RDD[Int] = FlatMappedRDD[10] at flatMap at
<console>:17
scala> res7.collect
14/06/13 01:11:48 INFO SparkContext: Starting job: collect at <console>:19
14/06/13 01:11:48 INFO DAGScheduler: Got job 2 (collect at <console>:19)
with 3 output partitions (allowLocal=false)
14/06/13 01:11:48 INFO DAGScheduler: Final stage: Stage 4(collect at
<console>:19)
14/06/13 01:11:48 INFO DAGScheduler: Parents of final stage: List()
14/06/13 01:11:48 INFO DAGScheduler: Missing parents: List()
14/06/13 01:11:48 INFO DAGScheduler: Submitting Stage 4 (FlatMappedRDD[10]
at flatMap at <console>:17), which has no missing parents
14/06/13 01:11:48 INFO DAGScheduler: Submitting 3 missing tasks from Stage
4 (FlatMappedRDD[10] at flatMap at <console>:17)
14/06/13 01:11:48 INFO TaskSchedulerImpl: Adding task set 4.0 with 3 tasks
14/06/13 01:11:48 INFO TaskSetManager: Starting task 4.0:0 as TID 16 on
executor localhost: localhost (PROCESS_LOCAL)
14/06/13 01:11:48 INFO TaskSetManager: Serialized task 4.0:0 as 1850 bytes
in 0 ms
14/06/13 01:11:48 INFO TaskSetManager: Starting task 4.0:1 as TID 17 on
executor localhost: localhost (PROCESS_LOCAL)
14/06/13 01:11:48 INFO TaskSetManager: Serialized task 4.0:1 as 1850 bytes
in 0 ms
14/06/13 01:11:48 INFO TaskSetManager: Starting task 4.0:2 as TID 18 on
executor localhost: localhost (PROCESS_LOCAL)
14/06/13 01:11:48 INFO TaskSetManager: Serialized task 4.0:2 as 1850 bytes
in 0 ms
14/06/13 01:11:48 INFO Executor: Running task ID 16
14/06/13 01:11:48 INFO Executor: Running task ID 17
14/06/13 01:11:48 INFO Executor: Running task ID 18
14/06/13 01:11:48 ERROR Executor: Exception in task ID 18
java.lang.NullPointerException
at org.apache.spark.rdd.RDD.collect(RDD.scala:728)
at $line45.$read$$iwC$$iwC$$iwC$$iwC$$anonfun$1.apply(<console>:17)
at $line45.$read$$iwC$$iwC$$iwC$$iwC$$anonfun$1.apply(<console>:17)
at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
at scala.collection.AbstractIterator.to(Iterator.scala:1157)
at
scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
at scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
at org.apache.spark.rdd.RDD$$anonfun$15.apply(RDD.scala:728)
at org.apache.spark.rdd.RDD$$anonfun$15.apply(RDD.scala:728)
at
org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1079)
at
org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1079)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:111)
at org.apache.spark.scheduler.Task.run(Task.scala:51)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:187)
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:744)
14/06/13 01:11:48 ERROR Executor: Exception in task ID 16
... same for each partition.
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org
$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1037)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1021)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1019)
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:1019)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:637)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:637)
at scala.Option.foreach(Option.scala:236)
at
org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:637)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1211)
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)
-kr, Gerard.
On Fri, Jun 13, 2014 at 12:32 AM, bdamos <a...@adobe.com> wrote:
> Hi, I'm consistently getting NullPointerExceptions when trying to use
> String val objects defined in my main application -- even for broadcast
> vals!
> I'm deploying on a standalone cluster with a master and 4 workers on the
> same machine, which is not the machine I'm submitting from.
>
> The following example works in spark-shell, but does not when
> submitted to the cluster with spark-submit, and also does not work locally.
>
> Is there anything I can do to fix this?
> Do vals need to be explicitly synchronized for RDD operations?
> One workaround in would be to inline the vals,
> but the logic in my actual application doesn't allow for this.
>
> Thanks,
> Brandon.
>
> ---
>
> sbt-shell --master <my-server>
>
> val suffix = "-suffix"
> val l = sc.parallelize(List("a", "b", "c"))
> println(l.map(_+suffix).collect().mkString(","))
>
> Result: a-suffix,b-suffix,c-suffix
>
> ---
>
> Standalone Cluster with `submit.sh` (my script below):
>
> TestApp.scala:
>
> package com.adobe.spark
>
> // Spark.
> import org.apache.spark.{SparkConf,SparkContext}
> import org.apache.spark.broadcast._
> import org.apache.spark.SparkContext._
> import org.apache.spark.storage.StorageLevel
>
> // Scala.
> import scala.collection.mutable.ArrayBuffer
>
> object TestApp extends App {
> val memory = "1g"
> val maxCores = "1"
> val conf = new SparkConf()
> .setMaster("spark://myserver:7077")
> //.setMaster("local[4]")
> .setAppName("ValError")
> .setSparkHome("/usr/local/spark-1.0.0")
> .setJars(Seq("/tmp/val-error.jar"))
> .set("spark.executor.memory", memory)
> .set("spark.cores.max", maxCores)
> val sc = new SparkContext(conf)
>
> val suffix = "-suffix"
> val l = sc.parallelize(List("a", "b", "c"))
> println(l.map(_+suffix).collect().mkString(","))
>
> val suffix_bc = sc.broadcast(suffix)
> println(l.map(_+suffix_bc.value).collect().mkString(","))
>
> sc.stop()
> }
>
> build.sbt:
>
> import AssemblyKeys._
>
> assemblySettings
>
> jarName in assembly := "val-error.jar"
>
> // Load "provided" libraries with `sbt run`.
> run in Compile <<= Defaults.runTask(
> fullClasspath in Compile, mainClass in (Compile, run), runner in
> (Compile, run)
> )
>
> name := "TestApp"
>
> version := "1.0"
>
> scalaVersion := "2.10.3"
>
> libraryDependencies ++= Seq(
> "org.apache.spark" %% "spark-core" % "1.0.0" % "provided",
> "org.slf4j" % "slf4j-simple" % "1.7.7" // Logging.
> )
>
> resolvers ++= Seq(
> "Akka Repository" at "http://repo.akka.io/releases/";
> )
>
> submit.sh:
>
> #!/bin/bash
>
> rm -f *.log driver-id.txt
>
> JAR=val-error.jar
> CLASS=com.adobe.spark.TestApp
> SPARK=/usr/local/spark-1.0.0
>
> set -x
> sbt assembly &> assembly.log || exit 1
> scp target/scala-2.10/$JAR eagle:/tmp || exit 2
>
> $SPARK/bin/spark-submit \
> --class $CLASS \
> --master spark://myserver:7077 \
> --deploy-mode cluster \
> /tmp/$JAR | tee submit.log
> set +x
>
> DRIVER_ID=$(grep 'Driver successfully submitted' submit.log | sed
> 's/Driver successfully submitted as \(.*\)/\1/g')
> [ -z $DRIVER_ID ] && exit 3
> echo $DRIVER_ID > driver-id.txt
>
> Output:
> anull,bnull,cnull (For the first part.)
>
> Stack Trace: (For the broadcast var.)
> Caused by: org.apache.spark.SparkException: Job aborted due to stage
> failure: Task 1.0:0 failed 4 times, most recent failure: Exception failure
> in TID 8 on host eagle.corp.adobe.com: java.lang.NullPointerException
> com.adobe.spark.TestApp$$anonfun$2.apply(App.scala:29)
> com.adobe.spark.TestApp$$anonfun$2.apply(App.scala:29)
> scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
> scala.collection.Iterator$class.foreach(Iterator.scala:727)
> scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
>
> scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
>
> scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:103)
>
> scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:47)
>
> scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
> scala.collection.AbstractIterator.to(Iterator.scala:1157)
>
> scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:265)
> scala.collection.AbstractIterator.toBuffer(Iterator.scala:1157)
>
> scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:252)
> scala.collection.AbstractIterator.toArray(Iterator.scala:1157)
> org.apache.spark.rdd.RDD$$anonfun$15.apply(RDD.scala:717)
> org.apache.spark.rdd.RDD$$anonfun$15.apply(RDD.scala:717)
>
>
> org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1080)
>
>
> org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1080)
>
> org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:111)
> org.apache.spark.scheduler.Task.run(Task.scala:51)
>
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:187)
>
>
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
>
>
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
> java.lang.Thread.run(Thread.java:744)
> Driver stacktrace:
> 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$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:633)
> at
>
> org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:633)
> at scala.Option.foreach(Option.scala:236)
> at
>
> org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:633)
> at
>
> org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1207)
> 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)
>
> ---
>
> When using 'local[4]' and `sbt run` with `setJars` commented.
> This is happening when `collect()` is called on the first map.
>
> org.apache.spark.SparkException: Job aborted due to stage failure: Task
> 0.0:0 failed 1 times, most recent failure: Exception failure in TID 0 on
> host localhost: java.lang.ClassNotFoundException: scala.None$
> java.net.URLClassLoader$1.run(URLClassLoader.java:372)
> java.net.URLClassLoader$1.run(URLClassLoader.java:361)
> java.security.AccessController.doPrivileged(Native Method)
> java.net.URLClassLoader.findClass(URLClassLoader.java:360)
> java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> java.lang.Class.forName0(Native Method)
> java.lang.Class.forName(Class.java:340)
>
>
> org.apache.spark.serializer.JavaDeserializationStream$$anon$1.resolveClass(JavaSerializer.scala:60)
>
> java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1613)
>
> java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1518)
>
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1774)
>
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
>
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1993)
>
> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1918)
>
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
>
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
> java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)
>
>
> org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:63)
>
> org.apache.spark.scheduler.ResultTask$.deserializeInfo(ResultTask.scala:61)
>
> org.apache.spark.scheduler.ResultTask.readExternal(ResultTask.scala:141)
>
> java.io.ObjectInputStream.readExternalData(ObjectInputStream.java:1840)
>
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1799)
>
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
> java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)
>
>
> org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:63)
>
>
> org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:85)
>
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:169)
>
>
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
>
>
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> java.lang.Thread.run(Thread.java:745)
> Driver stacktrace:
> 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$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:633)
> at
>
> org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:633)
> at scala.Option.foreach(Option.scala:236)
> at
>
> org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:633)
> at
>
> org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1207)
> 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)
>
>
>
>
> --
> View this message in context:
> http://apache-spark-user-list.1001560.n3.nabble.com/NullPointerExceptions-when-using-val-or-broadcast-on-a-standalone-cluster-tp7524.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
