CODEWITHSUNDEEP
scalaapache-sparkakka
null
null

Reputation: 3517

Trying to deserialize a serialized ActorRef without an ActorSystem in scope - Spark/Akka/Scala

EDIT: Exception relating to serialising / deserialising Akka Actor remains after forEach is confirmed to be executing. I'm learning Akka and encountering this serialisation issue, I'm quite new to Akka and finding it quite difficult to figure out how best to approach a solution.

I have a class where the actorSystem and actor is created, that class calls a method in another class, I'd like to use my actor in that other class. So far I've tried:

Class a:
-create actor
-classB.methodX(actor)

Class b:
-methodX(actorRef)
--actorRef! message

currently the message from class b does not get sent whereas the class A ones do, this is making me think that I either need to instantiate a new Actor with the ActorRef that I'm passing in or, it's completely wrong.

I understand that the system needs to be present on all nodes but I'm unsure how to do that.

Edit: Code Snippets:

ClassA:
  private val system = ActorSystem("System")
  private val metricActor= system.actorOf(Props(new MetricActor("metrics")), name = "metrics")

metricActor ! message works fine

writeFiles(metricActor) //writeFiles is in another class


writeFiles(actor: ActorRef) {
  f.forEach(v => {
  actor ! message // this doesn't seem to work
})
}

Edited to provide more code.

full example of classB as it were

def writeFiles(bucket: String, filesToWrite: RDD[(String, String)], actor:ActorRef): Unit =
filesToWrite.foreachPartition(f => {
  f.foreach(v => {
    try {
      //actor  ! ProcessSingleItem(MetricItem("File Created"))
      logger.debug(s"Writing file:")
      writeFile()
    }
    catch {
      case e: Exception =>
        //actor ! ProcessSingleItem(CloudWatchMetricItem("File create failure")) 
        logger.error("Error creating file", e)
    }
  })
})

Upvotes: 1

Views: 503

Answers (1)

slouc
slouc

Reputation: 9698

What you are describing should work. You didn't paste the full (but minimal!) code sample so I can't say what's the problem (maybe something in your foreach logic; we don'e even know what f is), but here's what definitely works. Try building up on that.

object Test extends App {

  class MetricActor extends Actor {
    def receive = {
      case a: String => println(a)
    }
  }

  private val system = ActorSystem("System")
  private val metricActor= system.actorOf(Props(new MetricActor()), name = "metrics")

  OtherClass.writeFiles(metricActor) // prints "foo"
}

// this is in other file:
object OtherClass {
  def writeFiles(actor: ActorRef) {
    actor ! "foo"
  }
}

EDIT: Turns out after discussion in comments that the problematic part was unrelated to Akka (try block was not even executing), but I will leave this here in case it helps someone.

Upvotes: 1

Related Questions