Getting strange behavior when calling function outside of a closure:

• when function is in a object everything is working
• when function is in a class get :

Task not serializable: java.io.NotSerializableException: testing

The problem is I need my code in a class and not an object. Any idea why this is happening? Is a Scala object serialized (default?)?

This is a working code example:

object working extends App {
    val list = List(1,2,3)

    val rddList = Spark.ctx.parallelize(list)
    //calling function outside closure 
    val after = rddList.map(someFunc(_))

    def someFunc(a:Int)  = a+1

    after.collect().map(println(_))
}

This is the non-working example :

object NOTworking extends App {
  new testing().doIT
}

//adding extends Serializable wont help
class testing {  
  val list = List(1,2,3)  
  val rddList = Spark.ctx.parallelize(list)

  def doIT =  {
    //again calling the fucntion someFunc 
    val after = rddList.map(someFunc(_))
    //this will crash (spark lazy)
    after.collect().map(println(_))
  }

  def someFunc(a:Int) = a+1
}

I solved this problem using a different approach. You simply need to serialize the objects before passing through the closure, and de-serialize afterwards. This approach just works, even if your classes aren't Serializable, because it uses Kryo behind the scenes. All you need is some curry. ;)

Here's an example of how I did it:

def genMapper(kryoWrapper: KryoSerializationWrapper[(Foo => Bar)])
               (foo: Foo) : Bar = {
    kryoWrapper.value.apply(foo)
}
val mapper = genMapper(KryoSerializationWrapper(new Blah(abc))) _
rdd.flatMap(mapper).collectAsMap()

object Blah(abc: ABC) extends (Foo => Bar) {
    def apply(foo: Foo) : Bar = { //This is the real function }
}

import org.apache.spark.{SparkContext,SparkConf}

object Spark {
val ctx = new SparkContext(new SparkConf().setAppName("test").setMaster("local
"))
}

object NOTworking extends App {
new Test().doIT
}

class Test extends java.io.Serializable {
val rddList = Spark.ctx.parallelize(List(1,2,3))

def doIT() = {
val after = rddList.map(someFunc)
after.collect().foreach(println)
}

def someFunc(a: Int) = a + 1
}​

import org.apache.spark.{SparkContext,SparkConf}

object Spark {
val ctx = new SparkContext(new SparkConf().setAppName("test").setMaster("local