Could you test serializing and deserializing the MyClassReporter class
separately?
On Mon, Dec 14, 2015 at 8:57 AM, Bartłomiej Alberski <albers...@gmail.com>
wrote:
> Below is the full stacktrace(real names of my classes were changed) with
> short description of entries from my code:
>
> rdd.mapPartitions{ partition => //this is the line to which second
> stacktrace entry is pointing
> val sender = broadcastedValue.value // this is the maing place to which
> first stacktrace entry is pointing
> }
>
> java.lang.ClassCastException:
> org.apache.spark.util.SerializableConfiguration cannot be cast to
> com.example.sender.MyClassReporter
> at com.example.flow.Calculator
> $$anonfun$saveAndLoadHll$1$$anonfun$apply$14.apply(AnomalyDetection.scala:87)
> at com.example.flow.Calculator
> $$anonfun$saveAndLoadHll$1$$anonfun$apply$14.apply(Calculator.scala:82)
> at
> org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$17.apply(RDD.scala:706)
> at
> org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$17.apply(RDD.scala:706)
> at
> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
> at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
> at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
> at
> org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
> at
> org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
> at org.apache.spark.scheduler.Task.run(Task.scala:88)
> at
> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
>
> 2015-12-14 17:10 GMT+01:00 Ted Yu <yuzhih...@gmail.com>:
>
>> Can you show the complete stack trace for the ClassCastException ?
>>
>> Please see the following thread:
>> http://search-hadoop.com/m/q3RTtgEUHVmJA1T1
>>
>> Cheers
>>
>> On Mon, Dec 14, 2015 at 7:33 AM, alberskib <albers...@gmail.com> wrote:
>>
>>> Hey all,
>>>
>>> When my streaming application is restarting from failure (from
>>> checkpoint) I
>>> am receiving strange error:
>>>
>>> java.lang.ClassCastException:
>>> org.apache.spark.util.SerializableConfiguration cannot be cast to
>>> com.example.sender.MyClassReporter.
>>>
>>> Instance of B class is created on driver side (with proper config passed
>>> as
>>> constructor arg) and broadcasted to the executors in order to ensure
>>> that on
>>> each worker there will be only single instance. Everything is going well
>>> up
>>> to place where I am getting value of broadcasted field and executing
>>> function on it i.e.
>>> broadcastedValue.value.send(...)
>>>
>>> Below you can find definition of MyClassReporter (with trait):
>>>
>>> trait Reporter{
>>> def send(name: String, value: String, timestamp: Long) : Unit
>>> def flush() : Unit
>>> }
>>>
>>> class MyClassReporter(config: MyClassConfig, flow: String) extends
>>> Reporter
>>> with Serializable {
>>>
>>> val prefix = s"${config.senderConfig.prefix}.$flow"
>>>
>>> var counter = 0
>>>
>>> @transient
>>> private lazy val sender : GraphiteSender = initialize()
>>>
>>> @transient
>>> private lazy val threadPool =
>>> ExecutionContext.fromExecutorService(Executors.newSingleThreadExecutor())
>>>
>>> private def initialize() = {
>>> val sender = new Sender(
>>> new InetSocketAddress(config.senderConfig.hostname,
>>> config.senderConfig.port)
>>> )
>>> sys.addShutdownHook{
>>> sender.close()
>>> }
>>> sender
>>> }
>>>
>>> override def send(name: String, value: String, timestamp: Long) : Unit
>>> = {
>>> threadPool.submit(new Runnable {
>>> override def run(): Unit = {
>>> try {
>>> counter += 1
>>> if (!sender.isConnected)
>>> sender.connect()
>>> sender.send(s"$prefix.$name", value, timestamp)
>>> if (counter % graphiteConfig.batchSize == 0)
>>> sender.flush()
>>> }catch {
>>> case NonFatal(e) => {
>>> println(s"Problem with sending metric to graphite
>>> $prefix.$name:
>>> $value at $timestamp: ${e.getMessage}", e)
>>> Try{sender.close()}.recover{
>>> case NonFatal(e) => println(s"Error closing graphite
>>> ${e.getMessage}", e)
>>> }
>>> }
>>> }
>>> }
>>> })
>>> }
>>>
>>> Do you have any idea how I can solve this issue? Using broadcasted
>>> variable
>>> helps me keeping single socket open to the service on executor.
>>>
>>>
>>>
>>> --
>>> View this message in context:
>>> http://apache-spark-user-list.1001560.n3.nabble.com/Spark-streaming-java-lang-ClassCastException-org-apache-spark-util-SerializableConfiguration-on-restt-tp25698.html
>>> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>>>
>>> ---------------------------------------------------------------------
>>> To unsubscribe, e-mail: user-unsubscr...@spark.apache.org
>>> For additional commands, e-mail: user-h...@spark.apache.org
>>>
>>>
>>
>
