SparkDriverExecutionException when using actorStream

[Juan Rodríguez Hortalá <ju...@gmail.com>]

Hi,

I'm trying to create a Spark Streaming actor stream but I'm having several
problems. First of all the guide from
https://spark.apache.org/docs/latest/streaming-custom-receivers.html refers
to the code
https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/ActorWordCount.scala,
which uses AkkaUtils and org.apache.spark.SecurityManager which are now
private[spark]. So I've tried with the example from
http://www.typesafe.com/activator/template/spark-streaming-scala-akka, but
I get the following exception as soon as I store some data in Spark
Streaming

org.apache.spark.SparkDriverExecutionException: Execution error
at
org.apache.spark.scheduler.DAGScheduler.handleTaskCompletion(DAGScheduler.scala:1025)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1411)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
Caused by: java.lang.ArrayStoreException: [Ljava.lang.Object;
at scala.runtime.ScalaRunTime$.array_update(ScalaRunTime.scala:88)
at
org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1750)
at
org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1750)
at org.apache.spark.scheduler.JobWaiter.taskSucceeded(JobWaiter.scala:56)
at
org.apache.spark.scheduler.DAGScheduler.handleTaskCompletion(DAGScheduler.scala:1021)
... 3 more

My code is basically the same as in that example, and it is available at
https://gist.github.com/juanrh/139af20fd2060cb1a9d1 . If I comment
receiverActor
! msg then there is no exception, but also no data is received in the
stream. Any thoughts on this?

Thanks a lot for you help.

Greetings,

Juan

Re: SparkDriverExecutionException when using actorStream

[Juan Rodríguez Hortalá <ju...@gmail.com>]

Hi,

I've finally fixed this. The problem was that I wasn't providing a type for
the DStream in ssc.actorStream

 /* with this inputDStream : ReceiverInputDStream[Nothing] and we get
SparkDriverExecutionException: Execution error
   * Caused by: java.lang.ArrayStoreException: [Ljava.lang.Object;
   *
   * val inputDStream = ssc.actorStream(Props(new
ProxyReceiverActor[String]), receiverActorName)
   */
  // with this inputDStream : ReceiverInputDStream[String]
  val inputDStream = ssc.actorStream[String](Props(new
ProxyReceiverActor[String]), receiverActorName)


Now it works ok both in Spark 1.3.1 and Spark 1.4.0. Here
https://gist.github.com/juanrh/eaf34cf0a308a87db32c you have the corrected
example in case someone is interested.

Greetings,

Juan


2015-07-10 18:27 GMT+02:00 Juan Rodríguez Hortalá <
juan.rodriguez.hortala@gmail.com>:

> Hi,
>
> I'm trying to create a Spark Streaming actor stream but I'm having several
> problems. First of all the guide from
> https://spark.apache.org/docs/latest/streaming-custom-receivers.html
> refers to the code
> https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/ActorWordCount.scala,
> which uses AkkaUtils and org.apache.spark.SecurityManager which are now
> private[spark]. So I've tried with the example from
> http://www.typesafe.com/activator/template/spark-streaming-scala-akka,
> but I get the following exception as soon as I store some data in Spark
> Streaming
>
> org.apache.spark.SparkDriverExecutionException: Execution error
> at
> org.apache.spark.scheduler.DAGScheduler.handleTaskCompletion(DAGScheduler.scala:1025)
> at
> org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1447)
> at
> org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1411)
> at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
> Caused by: java.lang.ArrayStoreException: [Ljava.lang.Object;
> at scala.runtime.ScalaRunTime$.array_update(ScalaRunTime.scala:88)
> at
> org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1750)
> at
> org.apache.spark.SparkContext$$anonfun$runJob$4.apply(SparkContext.scala:1750)
> at org.apache.spark.scheduler.JobWaiter.taskSucceeded(JobWaiter.scala:56)
> at
> org.apache.spark.scheduler.DAGScheduler.handleTaskCompletion(DAGScheduler.scala:1021)
> ... 3 more
>
> My code is basically the same as in that example, and it is available at
> https://gist.github.com/juanrh/139af20fd2060cb1a9d1 . If I comment receiverActor
> ! msg then there is no exception, but also no data is received in the
> stream. Any thoughts on this?
>
> Thanks a lot for you help.
>
> Greetings,
>
> Juan
>
>
>
>
>
>