在Spark中配置函数/lambda序列化

Question

如何配置Spark使用KryoSerializer处理lambda函数？还是说我发现了Spark的一个bug？我们在其他地方对数据的序列化没有问题，只有在使用默认值而不是Kryo时才会出现与这些lambda函数相关的问题。

以下是代码：

JavaPairRDD<String, IonValue> rdd; // provided
IonSexp filterExpression; // provided
Function<Tuple2<String, IonValue>, Boolean> filterFunc = record -> myCustomFilter(filterExpression, record);
rdd = rdd.filter(filterFunc);

抛出异常：

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, <function1>)
    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

在这种情况下，有问题的filterExpression是一个Ion S-Expression对象，它没有实现java.io.Serializable。我们正在使用Kryo序列化器，并已注册和配置它，以便可以很好地进行序列化。

当我们初始化Spark配置时的代码:

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");

我们注册器中的代码：

kryo.register(com.amazon.ion.IonSexp.class);
kryo.register(Class.forName("com.amazon.ion.impl.lite.IonSexpLite"));

如果我尝试使用以下代码手动序列化该lambda函数

SerializationUtils.serialize(filterFunc);

由于filterExpression不可序列化，因此它会按预期失败并显示相同的错误。然而，下面的代码可以正常工作：

sparkContext.env().serializer().newInstance().serialize(filterFunc, ClassTag$.MODULE$.apply(filterFunc.getClass()));

这是符合预期的，因为我们的Kryo设置能够处理这些对象。

那么我的问题/困惑是，既然我们已经明确配置了使用Kryo，为什么Spark要尝试使用org.apache.spark.serializer.JavaSerializer序列化那个lambda表达式呢？

- Mohannad

Answer 1

经过进一步挖掘，发现确实有一个不同的序列化程序用于闭包。由于Kryo存在错误，闭包序列化程序被硬编码为默认程序。

这个答案做了一个不错的解释: https://dev59.com/NFkS5IYBdhLWcg3wKzyq#40261550 然而，我通过使用广播来解决了我的特定问题。

现在我的代码看起来像这样:

JavaSparkContext sparkContext; // provided
JavaPairRDD<String, IonValue> rdd; // provided
IonSexp filterExpression; // provided

Broadcast<IonSexp> filterExprBroadcast = sparkContext.broadcast(filterExpression);
rdd = rdd.filter(record -> myCustomFilter(filterExprBroadcast.value(), record));
filterExprBroadcast.destroy(false); // Only do this after an action is executed

广播处理类似于RDD，因此需要使用配置的Kryo序列化器。

- Mohannad