title: Spark scala task not serializable for closure
body: <p>I have an RDD of Rows and I want to filter based on a closure. Ultimately I want to pass the closure in as a parameter to my method which is doing the filter, but I've simplified it and I can reproduce the error with something simple like this.</p>
  
  <pre><code>def fn(l: Long): Boolean = true
  rdd.filter{ row =&gt; fn(row.getAs[Long]("field")) }
  </code></pre>
  
  <p>I tried putting fn into a case object, an object that extends a serializable trait, defining fn inside and outside of the method calling filter.  I'm trying to figure out what I need to do without getting these errors.  I know there are a lot of questions about this already on stack overflow and I've been looking for a suitable answer but I can't find it.</p>
  
  <pre><code>Name: org.apache.spark.SparkException
  Message: Task not serializable
  StackTrace: org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:304)
  org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:294)
  org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:122)
  org.apache.spark.SparkContext.clean(SparkContext.scala:2058)
  org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:341)
  org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:340)
  org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
  org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)
  org.apache.spark.rdd.RDD.withScope(RDD.scala:316)
  org.apache.spark.rdd.RDD.filter(RDD.scala:340)
  $line131.$read$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:43)
  $line131.$read$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:48)
  $line131.$read$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:50)
  $line131.$read$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:52)
  $line131.$read$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:54)
  $line131.$read$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:56)
  $line131.$read$$iwC.&lt;init&gt;(&lt;console&gt;:58)
  $line131.$read.&lt;init&gt;(&lt;console&gt;:60)
  $line131.$read$.&lt;init&gt;(&lt;console&gt;:64)
  $line131.$read$.&lt;clinit&gt;(&lt;console&gt;)
  $line131.$eval$.&lt;init&gt;(&lt;console&gt;:7)
  $line131.$eval$.&lt;clinit&gt;(&lt;console&gt;)
  $line131.$eval.$print(&lt;console&gt;)
  sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
  sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  java.lang.reflect.Method.invoke(Method.java:601)
  org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:1065)
  org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1346)
  org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:840)
  org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:871)
  org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:819)
  org.apache.toree.kernel.interpreter.scala.ScalaInterpreter$$anonfun$interpretAddTask$1$$anonfun$apply$3.apply(ScalaInterpreter.scala:356)
  org.apache.toree.kernel.interpreter.scala.ScalaInterpreter$$anonfun$interpretAddTask$1$$anonfun$apply$3.apply(ScalaInterpreter.scala:351)
  org.apache.toree.global.StreamState$.withStreams(StreamState.scala:81)
  org.apache.toree.kernel.interpreter.scala.ScalaInterpreter$$anonfun$interpretAddTask$1.apply(ScalaInterpreter.scala:350)
  org.apache.toree.kernel.interpreter.scala.ScalaInterpreter$$anonfun$interpretAddTask$1.apply(ScalaInterpreter.scala:350)
  org.apache.toree.utils.TaskManager$$anonfun$add$2$$anon$1.run(TaskManager.scala:140)
  java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
  java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
  java.lang.Thread.run(Thread.java:722)
  </code></pre>
  
  <p>UPDATE:</p>
  
  <p>A more complete example.  I am running Jupyter with Toree and executing code from a jar file in my cells.  Here are three of the things I tried which fail</p>
  
  <pre><code>import org.apache.spark.SparkContext
  import org.apache.spark.rdd.RDD
  import org.apache.spark.sql.{Row, SQLContext}
  
  class NotWorking1(sc: SparkContext, sqlContext: SQLContext, fn: Long=&gt;Boolean) {
    def myFilterer(rdd:RDD[Row], longField: String): RDD[Row] = rdd.filter{ row =&gt; fn(row.getAs[Long](longField)) }
  }
  
  object NotWorking1 {
    def apply(sc: SparkContext, sqlContext: SQLContext) = {
      def myFn(l: Long): Boolean = true
      new NotWorking1(sc, sqlContext, myFn)
    }
  }
  
  class NotWorking2(sc: SparkContext, sqlContext: SQLContext) {
    def myFn(l: Long): Boolean = true
  
    def myFilterer(rdd:RDD[Row], longField: String): RDD[Row] = {
      rdd.filter{ row =&gt; myFn(row.getAs[Long](longField)) }
    }
  }
  
  object NotWorking2 {
    def apply(sc: SparkContext, sqlContext: SQLContext) = {
      new NotWorking2(sc, sqlContext)
    }
  }
  
  class NotWorking3(sc: SparkContext, sqlContext: SQLContext) {
    def myFilterer(rdd:RDD[Row], longField: String): RDD[Row] = {
      def myFn(l: Long): Boolean = true
      rdd.filter{ row =&gt; myFn(row.getAs[Long](longField)) }
    }
  }
  
  object NotWorking3 {
    def apply(sc: SparkContext, sqlContext: SQLContext) = {
      new NotWorking3(sc, sqlContext)
    }
  }
  </code></pre>
  
  <p>from the Jupyter cell, I import the appropriate class and run </p>
  
  <pre><code>val nw1 = NotWorking1(sc, sqlContext)
  val nw2 = NotWorking2(sc, sqlContext)
  val nw3 = NotWorking3(sc, sqlContext)
  nw1.myFilterer(rdd, "field")
  nw2.myFilterer(rdd, "field")
  nw3.myFilterer(rdd, "field")
  </code></pre>
  
  <p>All three fail.  NotWorking3 is particularly surprising.  It their anything I can do to isolate the function and not try to serialize the whole object (which I believe will get me into trouble anyway since I'm keeping a reference to the spark and sql contexts)</p>