title: Configure function/lambda serialization in Spark
body: <p>How do I configure Spark to use KryoSerializer for lambdas? Or have I found a bug in Spark? We have no issue with the serialization of the data elsewhere, just these lambdas where it uses the default instead of Kryo.</p>
  
  <p>Here's the code:</p>
  
  <pre><code>JavaPairRDD&lt;String, IonValue&gt; rdd; // provided
  IonSexp filterExpression; // provided
  Function&lt;Tuple2&lt;String, IonValue&gt;, Boolean&gt; filterFunc = record -&gt; myCustomFilter(filterExpression, record);
  rdd = rdd.filter(filterFunc);
  </code></pre>
  
  <p>Exception thrown:</p>
  
  <pre><code>org.apache.spark.SparkException: Task not serializable
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:403)
      at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:393)
      at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162)
      at org.apache.spark.SparkContext.clean(SparkContext.scala:2326)
      at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:388)
      at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:387)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
      at org.apache.spark.rdd.RDD.withScope(RDD.scala:363)
      at org.apache.spark.rdd.RDD.filter(RDD.scala:387)
      at org.apache.spark.api.java.JavaPairRDD.filter(JavaPairRDD.scala:99)
      at com.example.SomeClass.process(SomeClass.java:ABC)
      {more stuff}
  Caused by: java.io.NotSerializableException: com.amazon.ion.impl.lite.IonSexpLite
  Serialization stack:
      - object not serializable (class: com.amazon.ion.impl.lite.IonSexpLite, value: (and (equals (literal 1) (path marketplace_id)) (equals (literal 351) (path product gl_product_group))))
      - element of array (index: 1)
      - array (class [Ljava.lang.Object;, size 2)
      - field (class: java.lang.invoke.SerializedLambda, name: capturedArgs, type: class [Ljava.lang.Object;)
      - object (class java.lang.invoke.SerializedLambda, SerializedLambda[capturingClass=class com.example.SomeClass, functionalInterfaceMethod=org/apache/spark/api/java/function/Function.call:(Ljava/lang/Object;)Ljava/lang/Object;, implementation=invokeSpecial com/example/SomeClass.lambda$process$8f20a2d2$1:(Lcom/amazon/ion/IonSexp;Lscala/Tuple2;)Ljava/lang/Boolean;, instantiatedMethodType=(Lscala/Tuple2;)Ljava/lang/Boolean;, numCaptured=2])
      - writeReplace data (class: java.lang.invoke.SerializedLambda)
      - object (class com.example.SomeClass$$Lambda$36/263969036, com.example.SomeClass$$Lambda$36/263969036@31880efa)
      - field (class: org.apache.spark.api.java.JavaPairRDD$$anonfun$filter$1, name: f$1, type: interface org.apache.spark.api.java.function.Function)
      - object (class org.apache.spark.api.java.JavaPairRDD$$anonfun$filter$1, &lt;function1&gt;)
      at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
      at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
      at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:400)
      ... 18 more
  
  </code></pre>
  
  <p>The problematic <code>filterExpression</code> in this case is an <a href="https://github.com/amzn/ion-java/blob/master/src/com/amazon/ion/IonSexp.java" rel="nofollow noreferrer">Ion S-Expression</a> object, which does not implement <code>java.io.Serializable</code>. We're using the Kryo serializer and have registered &amp; configured it such that it can serialize it just fine.</p>
  
  <p>Code when we initialize our spark configuration:</p>
  
  <pre><code>sparkConf = new SparkConf().setAppName("SomeAppName").setMaster("MasterLivesHere")
          .set("spark.serializer", KryoSerializer.class.getCanonicalName())
          .set("spark.kryo.registrator", KryoRegistrator.class.getCanonicalName())
          .set("spark.kryo.registrationRequired", "false");
  </code></pre>
  
  <p>Code in our registrator:</p>
  
  <pre><code>kryo.register(com.amazon.ion.IonSexp.class);
  kryo.register(Class.forName("com.amazon.ion.impl.lite.IonSexpLite"));
  </code></pre>
  
  <p>If I try to manually serialize that lambda with the code below</p>
  
  <pre><code>SerializationUtils.serialize(filterFunc);
  </code></pre>
  
  <p>It fails with the same error, as expected, since <code>filterExpression</code> is not serializable. However, the code below works:</p>
  
  <pre><code>sparkContext.env().serializer().newInstance().serialize(filterFunc, ClassTag$.MODULE$.apply(filterFunc.getClass()));
  </code></pre>
  
  <p>Which, again, is as expected since our Kryo set-up is able to handle these objects.</p>
  
  <p>So my question/confusion is, why does Spark try to serialize that lambda with <code>org.apache.spark.serializer.JavaSerializer</code> when we've clearly configured it to use Kryo?</p>