问题描述
我开始使用星火流处理实时数据馈送我得到。我的情况是我必须用同ActorHelper一个演员阿卡接收器的话,我有我的工作星火做一些映射和转换,然后我想将结果发送给另一个演员。
I am starting to use Spark Streaming to process a real time data feed I am getting. My scenario is I have a Akka actor receiver using "with ActorHelper", then I have my Spark job doing some mappings and transformation and then I want to send the result to another actor.
我的问题是最后一部分。当试图发送到另一个演员Spark是引发异常:
My issue is the last part. When trying to send to another actor Spark is raising an exception:
15/02/20 16时43分16秒WARN TaskSetManager:在第一阶段2.0迷失任务0.0(TID 2,本地主机):java.lang.IllegalStateException:尝试反序列化序列化ActorRef没有范围的ActorSystem。使用akka.serialization.Serialization.currentSystem.withValue(系统){...}
我创造这最后男主角的方法如下:
The way I am creating this last actor is the following:
val actorSystem = SparkEnv.get.actorSystem
val lastActor = actorSystem.actorOf(MyLastActor.props(someParam), "MyLastActor")
,然后用它是这样的:
And then using it like this:
result.foreachRDD(rdd => rdd.foreachPartition(lastActor ! _))
我不知道在哪里,怎么办劝使用'akka.serialization.Serialization.currentSystem.withValue(系统){...}'。我是否需要设置什么特别通过配置?或不同的创建我的演员?
I am not sure where or how to do the advise "Use 'akka.serialization.Serialization.currentSystem.withValue(system) { ... }'". Do I need to set anything special through configuration? Or create my actor differently?
推荐答案
我发现,如果我收集之前,我发送给它就像一个魅力的演员:
I found that if I collect before I send to the actor it works like a charm:
result.foreachRDD(rdd => rdd.collect().foreach(producer ! _))
这篇关于无法反序列化ActorRef发送结果不同的演员的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!