title: Spark TaskNotSerializable when using anonymous function
body: <p><strong>Background</strong></p>
  
  <p>Here's my situation: I'm trying to create a class that filters an RDD based on some feature of the contents, but that feature can be different in different scenarios so I'd like to parameterize that with a function. Unfortunately, I seem to be running into issues with the way Scala captures its closures. Even though my function is serializable, the class is not.</p>
  
  <p>From the example in the <a href="https://github.com/apache/spark/blob/c1bc4f439f54625c01a585691e5293cd9961eb0c/core/src/main/scala/org/apache/spark/util/ClosureCleaner.scala#L148" rel="nofollow">spark source on closure cleaning</a>, it seems to suggest my situation can't be solved, but I'm convinced there's a way to achieve what I'm trying to do by creating the right (smaller) closure.</p>
  
  <p><strong>My Code</strong></p>
  
  <pre><code>class MyFilter(getFeature: Element =&gt; String, other: NonSerializable) {
    def filter(rdd: RDD[Element]): RDD[Element] = {
      // All my complicated logic I want to share
      rdd.filter { elem =&gt; getFeature(elem) == "myTargetString" }     
  }
  </code></pre>
  
  <p><strong>Simplified Example</strong></p>
  
  <pre><code>class Foo(f: Int =&gt; Double, rdd: RDD[Int]) { 
    def go(data: RDD[Int]) = data.map(f) 
  }
  
  val works = new Foo(_.toDouble, otherRdd)
  works.go(myRdd).collect() // works
  
  val myMap = Map(1 -&gt; 10d)
  val complicatedButSerializableFunc: Int =&gt; Double = x =&gt; myMap.getOrElse(x, 0)
  val doesntWork = new Foo(complicatedButSerializableFunc, otherRdd)
  doesntWork.go(myRdd).collect() // craps out
  
  org.apache.spark.SparkException: Task not serializable
  Caused by: java.io.NotSerializableException: $iwC$$iwC$Foo
  Serialization stack:
      - object not serializable (class: $iwC$$iwC$Foo, value: $iwC$$iwC$Foo@61e33118)
      - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: foo, type: class $iwC$$iwC$Foo)
      - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@47d6a31a)
      - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$1, name: $outer, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
      - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$1, &lt;function1&gt;)
  
  // Even though
  val out = new ObjectOutputStream(new FileOutputStream("test.obj"))
  out.writeObject(complicatedButSerializableFunc) // works
  </code></pre>
  
  <p><strong>Questions</strong></p>
  
  <ol>
  <li>Why does the first simplied example not attempt to serialize all of <code>Foo</code> but the second one does?</li>
  <li>How can I get the reference to my serializable function without including a reference to <code>Foo</code> in my closure?</li>
  </ol>