3
$\begingroup$
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.sql.cassandra.CassandraSQLContext
object Test {
  val sparkConf = new SparkConf(true).set("spark.cassandra.connection.host", <Cassandra Server IP>)
  val sc = new SparkContext(sparkConf)
  val cassandraSQLContext = new CassandraSQLContext(sc)
  val numberAsString = cassandraSQLContext.sql("select * from testing.test").first().getAs[Int]("number").toString()
  val testRDD = sc.parallelize(List(0, 0))
  val newRDD = testRDD.map { x => numberAsString }
}

This is the code that I have written in Spark. I was expecting it to work properly because i am evaluating the numberAsString value and then using it in the map function but it gives me task not serializable error. I am running the job in local mode.

Error on spark-shell

Error on spark-shell

$\endgroup$
1
  • $\begingroup$ I looked at the answer by @PierceHardman and I got the answer to my problem. $\endgroup$ – Credosam Sep 15 '16 at 6:54
3
$\begingroup$

Looking at another question on Stack Overflow about serialization exceptions in Spark, it says that anonymous functions serialize their containing class, and if that class contains the SparkContext -- which is not serializable -- then an error is thrown.

Perhaps this is happening to you?

$\endgroup$
0
3
$\begingroup$

I changed my code to this:

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.sql.cassandra.CassandraSQLContext
import org.apache.spark.rdd.RDD
object Test2 {
  def calculate(numberAsString: String, testRDD: RDD[Int]): RDD[String] = {
    val newRDD = testRDD.map { x => numberAsString }
    newRDD
  }
}

object Test {
  val sparkConf = new SparkConf(true).set("spark.cassandra.connection.host", <Cassandra Server IP>)
  val sc = new SparkContext(sparkConf)
  val cassandraSQLContext = new CassandraSQLContext(sc)
  val numberAsString = cassandraSQLContext.sql("select * from hdfc.test").first().getAs[Int]("number").toString()
  val testRDD = sc.parallelize(List(0, 0))
  val newRDD = Test2.calculate(numberAsString, testRDD)
}

Now when it will go to serialize the containing class then it will get serialized.

$\endgroup$
0
$\begingroup$

You should broadcast your variable.

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.sql.cassandra.CassandraSQLContext
object Test {
  val sparkConf = new SparkConf(true).set("spark.cassandra.connection.host", <Cassandra Server IP>)
  val sc = new SparkContext(sparkConf)
  val cassandraSQLContext = new CassandraSQLContext(sc)
  val numberAsString = cassandraSQLContext.sql("select * from testing.test").first().getAs[Int]("number").toString()
  val numberAsStringBC = sc.broadcast(numberAsString)
  val testRDD = sc.parallelize(List(0, 0))
  val newRDD = testRDD.map { x => numberAsStringBC.value }
}
$\endgroup$
1
  • $\begingroup$ I tried broadcasting the variable. There is no change in the error. It still shows me Task not serializable error. I am running the job in local mode. $\endgroup$ – Credosam Sep 14 '16 at 13:47

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.