示例Java程序中的Spark UDF反序列化错误

Question

这个示例直接取自Spark示例代码，所以我不太清楚到底发生了什么。

import org.apache.spark.sql.*;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.types.DataTypes;

public class TestSpark {

    public static void main(String[] args) {

        SparkSession spark = SparkSession.builder()
                .appName("testspark")
                .master("spark://0.0.0.0:7077")
                 // Yes this file exists, and contains this class
                .config("spark.jars", "out/artifacts/testspark_jar/testspark.jar")
                .getOrCreate();

        // This works
        spark.sql("SELECT 5 + 1").show();

        spark.udf().register("plusOne", (UDF1<Integer, Integer>) x -> x + 1, DataTypes.IntegerType);

        // This fails
        spark.sql("SELECT plusOne(5)").show();

    }

}

我在localhost上运行的Spark独立集群上运行这个。

工人始终失败：

Caused by: java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.rdd.MapPartitionsRDD.f of type scala.Function3 in instance of org.apache.spark.rdd.MapPartitionsRDD
    at java.base/java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2205)
    at java.base/java.io.ObjectStreamClass$FieldReflector.checkObjectFieldValueTypes(ObjectStreamClass.java:2168)
    at java.base/java.io.ObjectStreamClass.checkObjFieldValueTypes(ObjectStreamClass.java:1422)
    at java.base/java.io.ObjectInputStream.defaultCheckFieldValues(ObjectInputStream.java:2450)
    at java.base/java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2357)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2166)
    at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1668)
    at java.base/java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2434)
    at java.base/java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2328)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2166)
    at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1668)
    at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:482)
    at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:440)
    at scala.collection.immutable.List$SerializationProxy.readObject(List.scala:488)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at java.base/java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1175)
    at java.base/java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2295)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2166)
    at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1668)
    at java.base/java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2434)
    at java.base/java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2328)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2166)
    at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1668)
    at java.base/java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2434)
    at java.base/java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2328)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2166)
    at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1668)
    at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:482)
    at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:440)
    at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:76)
    at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:115)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:83)
    at org.apache.spark.scheduler.Task.run(Task.scala:127)
    at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:446)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:449)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)

我运行的是Java 11，使用的是Spark 3.0.1。

我确实发现了一个非常相似的问题，看起来它就是答案：java。lang.ClassCastException在远程服务器上的spark作业中使用lambda表达式

然而，在确保将TestSpark编译到提供给SparkSession的JAR中之后，我仍然会收到相同的错误。

任何帮助都将不胜感激。似乎Java/Scala边界上发生了一些事情，但我对Scala互操作的了解还不够，无法进一步分析。

Answer 1

我已经链接的问题（在远程服务器上的spark job中使用lambda表达式的java.lang.ClassCastException）的答案是正确的。

因为我没有正确编译JAR，所以它再次被弄糊涂了（我想应该是有清单的，但不是百分之百确定）。

正确编译JAR后（使用mvn包而不是IDE），并在spark中提供JAR。jarsconfig属性，代码按预期工作。