java.lang.ClassCastException 在远程服务器上的 spark 作业中使用 lambda 表达式

你在这里看到的是一个后续错误，它掩盖了原始错误。

当 lambda 实例被序列化时，它们用于将其 JRE 特定实现从持久形式（即序列化Lambda 实例）中解体。还原实例后，将调用其方法以重新构建相应的 lambda 实例。正如文档所说，它将通过调用定义原始lambda的类的特殊方法来做到这一点（另请参阅此答案）。重要的一点是需要原始类，这就是你的情况所缺少的。writeReplaceSerializedLambdareadResolve

但是有一个...特殊。。。的行为。当它遇到异常时，它不会立即救助。它将记录异常并继续该过程，将当前正在读取的所有对象标记为错误，从而根据错误对象也将错误。只有在该过程结束时，它才会引发它遇到的原始异常。使它如此奇怪的是，它还将继续尝试设置这些对象的字段。但是当你看一下方法行 1806 时：ObjectInputStreamObjectInputStream.readOrdinaryObject

…
    if (obj != null &&
        handles.lookupException(passHandle) == null &&
        desc.hasReadResolveMethod())
    {
        Object rep = desc.invokeReadResolve(obj);
        if (unshared && rep.getClass().isArray()) {
            rep = cloneArray(rep);
        }
        if (rep != obj) {
            handles.setObject(passHandle, obj = rep);
        }
    }

    return obj;
}

您会看到，当报告非异常时，它不会调用该方法。但是，当替换没有发生时，继续尝试设置引荐来源网址的字段值不是一个好主意，但这正是这里发生的事情，因此会产生.readResolvelookupExceptionnullClassCastException

您可以轻松重现问题：

public class Holder implements Serializable {
    Runnable r;
}
public class Defining {
    public static Holder get() {
        final Holder holder = new Holder();
        holder.r=(Runnable&Serializable)()->{};
        return holder;
    }
}
public class Writing {
    static final File f=new File(System.getProperty("java.io.tmpdir"), "x.ser");
    public static void main(String... arg) throws IOException {
        try(FileOutputStream os=new FileOutputStream(f);
            ObjectOutputStream   oos=new ObjectOutputStream(os)) {
            oos.writeObject(Defining.get());
        }
        System.out.println("written to "+f);
    }
}
public class Reading {
    static final File f=new File(System.getProperty("java.io.tmpdir"), "x.ser");
    public static void main(String... arg) throws IOException, ClassNotFoundException {
        try(FileInputStream is=new FileInputStream(f);
            ObjectInputStream ois=new ObjectInputStream(is)) {
            Holder h=(Holder)ois.readObject();
            System.out.println(h.r);
            h.r.run();
        }
        System.out.println("read from "+f);
    }
}

编译这四个类并运行 。然后删除类文件并运行 。然后你会得到一个WritingDefining.classReading

Exception in thread "main" java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field test.Holder.r of type java.lang.Runnable in instance of test.Holder
    at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2089)
    at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1261)

（使用 1.8.0_20 进行测试）

底线是，一旦了解了发生了什么，您可能会忘记此序列化问题，解决您的问题所要做的就是确保定义lambda表达式的类在lambda被反序列化的运行时中也可用。

直接从 IDE 运行的 Spark Job 示例（默认情况下，spark-submit 分发 jar）：

SparkConf sconf = new SparkConf()
  .set("spark.eventLog.dir", "hdfs://nn:8020/user/spark/applicationHistory")
  .set("spark.eventLog.enabled", "true")
  .setJars(new String[]{"/path/to/jar/with/your/class.jar"})
  .setMaster("spark://spark.standalone.uri:7077");

我遇到了同样的错误，我用内部类替换了lambda，然后它工作了。我真的不明白为什么，并且重现这个错误是非常困难的（我们有一个服务器表现出这种行为，而不是其他地方）。

不起作用

this.variable = () -> { ..... }

收益 率java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field MyObject.val$variable

工程

this.variable = new MyInterface() {
    public void myMethod() {
       .....
    }
};