Hari Shreedharan created SPARK-1785:
---------------------------------------
Summary: Streaming requires receivers to be serializable
Key: SPARK-1785
URL: https://issues.apache.org/jira/browse/SPARK-1785
Project: Spark
Issue Type: Bug
Components: Streaming
Affects Versions: 0.9.0
Reporter: Hari Shreedharan
When the ReceiverTracker starts the receivers it creates a temporary RDD to
send the receivers over to the workers. Then they are started on the workers
using a the startReceivers method.
Looks like this means that the receivers have to really be serializable. In
case of the Flume receiver, the Avro IPC components are not serializable
causing an error that looks like this:
{code}
Exception in thread "Thread-46" org.apache.spark.SparkException: Job aborted
due to stage failure: Task not serializable: java.io.NotSerializableException:
org.apache.avro.ipc.specific.SpecificResponder
- field (class "org.apache.spark.streaming.flume.FlumeReceiver", name:
"responder", type: "class org.apache.avro.ipc.specific.SpecificResponder")
- object (class "org.apache.spark.streaming.flume.FlumeReceiver",
org.apache.spark.streaming.flume.FlumeReceiver@5e6bbb36)
- element of array (index: 0)
- array (class "[Lorg.apache.spark.streaming.receiver.Receiver;", size:
1)
- field (class "scala.collection.mutable.WrappedArray$ofRef", name:
"array", type: "class [Ljava.lang.Object;")
- object (class "scala.collection.mutable.WrappedArray$ofRef",
WrappedArray(org.apache.spark.streaming.flume.FlumeReceiver@5e6bbb36))
- field (class "org.apache.spark.rdd.ParallelCollectionPartition",
name: "values", type: "interface scala.collection.Seq")
- custom writeObject data (class
"org.apache.spark.rdd.ParallelCollectionPartition")
- object (class "org.apache.spark.rdd.ParallelCollectionPartition",
org.apache.spark.rdd.ParallelCollectionPartition@691)
- writeExternal data
- root object (class "org.apache.spark.scheduler.ResultTask",
ResultTask(0, 0))
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)
{code}
A way out of this is to simply send the class name (or .class) to the workers
in the tempRDD and have the workers instantiate and start the receiver.
My analysis maybe wrong. but if it makes sense, I will submit a PR to fix this.
--
This message was sent by Atlassian JIRA
(v6.2#6252)
