I'm using Spark with Scala and Play2.4. First of all, I saw this thread: Task not serializable: java.io.NotSerializableException when calling function outside closure only on classes not objects
I'm ok with the fact that if we pass a method to Spark, it tries to serialize the whole class.
My code, it will be clearer:
class GPSSparkServiceImpl @Inject() (val stepService: StepDbService, val coordinateService: CoordinateService) extends GPSSparkService with Serializable{
/**
* Spark config
* Set Master node and AppName
*/
val conf = new SparkConf().setAppName("Editus GPS").setMaster("local[2]")
/**
* Initialize Spark Context
*/
val sc = new SparkContext(conf)
override def execute() = {
val logData = sc.textFile("file://C://work/csv/gps.csv").cache()
val numAs = logData.filter(line => line.contains("a")).count()
println("Lines with a: %s".format(numAs))
}
override def generateUserToStep(): Unit = {
val futureSteps = stepService.findAll()
futureSteps onSuccess{
case steps =>
val data = sc.textFile("file://C://work/csv/gps.csv").cache()
val result = data.flatMap(line => steps.map(step => (line, step))).filter { tuple =>
coordinateService.checkProximity(
coordinateService.coordinateToDistanceInMeters(
tuple._1.split(";")(1).toDouble, tuple._1.split(";")(2).toDouble, tuple._2.gpsCoordinate.latitude, tuple._2.gpsCoordinate.longitude
), tuple._2
)
}.count()
println("result: " + result + " for " + steps.length + " steps")
}
}
}
As you can see I use 2 methods in spark: checkProximity and CoordinateToDistanceInMeters from CoordinateService which will be injected as CoordinateServiceImpl and this class is serializable. All objects in the class are serializable.
CoordinateServiceImpl:
class CoordinateServiceImpl @Inject() (val config: Configuration) extends CoordinateService with Serializable{
override def coordinateToDistanceInMeters(lat1: Double, lng1: Double, lat2: Double, lng2: Double): Double = {
val earthRadius: Double = 6371000
val dLat: Double = Math.toRadians(lat2-lat1)
val dLng: Double = Math.toRadians(lng2-lng1)
val a: Double = Math.sin(dLat/2) * Math.sin(dLat/2) +
Math.cos(Math.toRadians(lat1)) * Math.cos(Math.toRadians(lat2)) *
Math.sin(dLng/2) * Math.sin(dLng/2)
val c: Double = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1-a))
earthRadius * c
}
override def checkProximity(distance: Double, step: Step): Boolean =
distance < step.acceptableProximity
}
Why am I facing SparkConf not serializable, which is in the actual class ? I'm not even using any method of it. Maybe I'm missing something ....
The stacktrace:
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158)
at org.apache.spark.SparkContext.clean(SparkContext.scala:1623)
at org.apache.spark.rdd.RDD.filter(RDD.scala:303)
at services.gps.spark.GPSSparkServiceImpl$$anonfun$generateUserToStep$1.applyOrElse(GPSSparkServiceImpl.scala:
41)
at services.gps.spark.GPSSparkServiceImpl$$anonfun$generateUserToStep$1.applyOrElse(GPSSparkServiceImpl.scala:
38)
at scala.concurrent.Future$$anonfun$onSuccess$1.apply(Future.scala:117)
at scala.concurrent.Future$$anonfun$onSuccess$1.apply(Future.scala:115)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32)
at scala.concurrent.impl.ExecutionContextImpl$AdaptedForkJoinTask.exec(ExecutionContextImpl.scala:121)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.pollAndExecAll(ForkJoinPool.java:1253)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1346)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
Caused by: java.io.NotSerializableException: org.apache.spark.SparkConf
Serialization stack:
- object not serializable (class: org.apache.spark.SparkConf, value: org.apache.spark.SparkConf@1b0321ae)
- field (class: services.gps.spark.GPSSparkServiceImpl, name: conf, type: class org.apache.spark.SparkConf)
- object (class services.gps.spark.GPSSparkServiceImpl, services.gps.spark.GPSSparkServiceImpl@458c4049)
- field (class: services.gps.spark.GPSSparkServiceImpl$$anonfun$generateUserToStep$1, name: $outer, type: clas
s services.gps.spark.GPSSparkServiceImpl)
- object (class services.gps.spark.GPSSparkServiceImpl$$anonfun$generateUserToStep$1, <function1>)
- field (class: services.gps.spark.GPSSparkServiceImpl$$anonfun$generateUserToStep$1$$anonfun$3, name: $outer,
type: class services.gps.spark.GPSSparkServiceImpl$$anonfun$generateUserToStep$1)
- object (class services.gps.spark.GPSSparkServiceImpl$$anonfun$generateUserToStep$1$$anonfun$3, <function1>)
at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:38)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:80)
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164)
... 14 more
