logo标签列表

远程服务器上使用Lambda表达式的Spark作业出现java.lang.ClassCastException

javaapache-sparklambdajava-8spark-java
31

我正在尝试使用sparkjava.com框架为我的Apache Spark作业构建Web API。 我的代码如下：

@Override
public void init() {
    get("/hello",
            (req, res) -> {
                String sourcePath = "hdfs://spark:54310/input/*";

                SparkConf conf = new SparkConf().setAppName("LineCount");
                conf.setJars(new String[] { "/home/sam/resin-4.0.42/webapps/test.war" });
                File configFile = new File("config.properties");

                String sparkURI = "spark://hamrah:7077";

                conf.setMaster(sparkURI);
                conf.set("spark.driver.allowMultipleContexts", "true");
                JavaSparkContext sc = new JavaSparkContext(conf);

                @SuppressWarnings("resource")
                JavaRDD<String> log = sc.textFile(sourcePath);

                JavaRDD<String> lines = log.filter(x -> {
                    return true;
                });

                return lines.count();
            });
}

如果我删除lambda表达式或将它放在一个普通的jar包中而不是一个Web服务（某种方式是一个servlet），它将不会产生任何错误。但在servlet中使用lambda表达式将导致这个异常：

15/01/28 10:36:33 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, hamrah): java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.api.java.JavaRDD$$anonfun$filter$1.f$1 of type org.apache.spark.api.java.function.Function in instance of org.apache.spark.api.java.JavaRDD$$anonfun$filter$1
at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2089)
at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1261)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1999)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1918)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1993)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1918)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1993)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1918)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)
at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:62)
at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:87)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:57)
at org.apache.spark.scheduler.Task.run(Task.scala:56)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

顺便说一下：我尝试了jersey和javaspark与jetty、tomcat和resin的组合，它们都导致了相同的结果。

1
@pzecevic 我正在尝试使用Apache Spark和Spark Java的组合，但只有在它们的组合中才会出现上述错误。如果没有这个组合，一切都正常。 - Mehraban
2
@Khajavi：可序列化的lambda实例使用writeReplace在流中用SerializedLambda实例替换自身。 SerializedLambda具有readResolve方法，以在恢复时执行相反操作。 writeReplace和readResolve是序列化的内置功能。 - Holger
你是否在根据你所构建的东西使用不同的编译器？我知道Eclipse JDT编译器在lambda反序列化方面存在一些问题。我创建这篇文章是基于我遇到的问题：http://stackoverflow.com/questions/27488567/invalid-lambda-deserialization/27502424#27502424 - Robert Bain
所以，你的意思是用一个普通的匿名类替换 lambda 表达式就可以像预期的那样工作？ - Olimpiu POP
1
我认为这个 bug 与此有关。如果我将我的 lambda 替换为匿名类，它对我起作用了。https://bugs.openjdk.java.net/browse/JDK-8024931 - Gábor Lipták
显示剩余4条评论
4个回答
64

您在这里遇到的是一个跟进错误，它掩盖了原始错误。

当lambda实例被序列化时，它们使用writeReplace方法将其JRE特定的实现从持久化形式SerializedLambda实例中溶解出来。当SerializedLambda实例已经被恢复时，它的readResolve方法将被调用以重建适当的lambda实例。正如文档所说，它将通过调用定义原始lambda的类的特殊方法来实现（请参见this answer）。重要的是需要原始类，而这在您的情况下是缺失的。

但是ObjectInputStream有一种特殊的行为。当它遇到异常时，它不会立即退出。它会记录异常并继续处理，标记当前正在读取的所有对象，因此将错误对象视为错误。只有在整个过程结束时，它才会抛出它遇到的原始异常。让它如此奇怪的是，它还会继续尝试设置这些对象的字段。但是当您查看ObjectInputStream.readOrdinaryObject方法的第1806行时：
…
    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;
}

你会发现当 lookupException 报告非 null 异常时，它不会调用 readResolve 方法。但是当替换没有发生时，继续尝试设置引用对象的字段值并不是一个好主意，但这正是此处发生的情况，因此产生了 ClassCastException。

你可以轻松地重现这个问题：

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

编译这四个类并运行Writing。然后删除类文件Defining.class并运行Reading。然后你会得到一个

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表达式的类在反序列化lambda的运行时环境中也可用。”，你能告诉我如何做到吗？能给个例子吗？ - ѕтƒ
6
如果您在像Idea这样的IDE中运行代码，则只需在SparkConf实例上调用setJars(new String[]{"/path/to/jar/with/your/class.jar"})。如果您使用spark-submit提交作业，则默认情况下会分发您的jar包，因此不会出现此类问题。请注意，本人的任务是翻译，因此无法提供任何解释或其他内容。 - morsik
1
@Holger - 太棒了！ :) 出于好奇，对于在Spark shell/repl中定义的lambda函数，它是如何工作的？我想这些函数的定义类是匿名的，但它们也会被反序列化吗？ - George
2
@George，我不参与Spark的开发，所以我不知道。但是我可以想象，shell会记住你输入的定义，这是一种可转移的形式，足以在另一端复制代码。因此，只需要在另一端提供shell或其后端的代码即可。 - Holger
优秀的分析，干得好！在我看来，这种“特殊”的行为实际上会暗中吞噬异常，这是一个非常糟糕的反模式，绝对不应该潜藏在JVM的代码中。也许我们应该发布一个错误报告？ - Martin Vysny
1
@MartinVysny 这么久以前了，我都记不起为什么我没有这样做。也许是因为我意识到了这个根本问题，并且认为它已经损坏了。但既然这个答案的问题是可以避免的，发布一个错误报告是有意义的。我会检查是否已经存在相应的报告，如果没有，我会提交一个新的报告。 - Holger
5

我遇到了同样的错误，将lambda替换为内部类后，它就能正常工作了。我不太明白为什么，而且再现此错误非常困难（我们只有一个服务器出现了这种行为，其他地方都没有）。

无法正常工作

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

出现了 java.lang.ClassCastException: 无法将 java.lang.invoke.SerializedLambda 分配给 MyObject.val$variable 字段

正常工作

this.variable = new MyInterface() {
    public void myMethod() {
       .....
    }
};
同样的问题，我用一个匿名类替换了它，现在它可以工作了。我不认为这是因为 JAR 包的问题，因为 JAR 包已经在配置中发送给 Spark 了。不知道是什么原因导致了这种 lambda 表达式的行为异常。 - Danilow
在我的情况下非常奇怪，因为它只与特定类型的FilterFunction lambda不起作用，而同一类中的其他lambda却可以正常工作。 - Danilow
这对我也解决了问题。 - undefined
3

我猜你的问题是自动装箱失败。在代码中

x -> {
      return true;
}

在使用过滤方法时，需要传入(String->Boolean)类型的lambda表达式（即Function<String,Boolean>），而你传入了一个(String->boolean)类型的lambda表达式（即Predicate<String>）。因此，建议你将代码改为：

x -> {
      return Boolean.TRUE;
}

请在您的问题中提供详细信息。非常感谢您提供uname -a和java -version的输出。如果可能，请提供sscce。

我认为你忽略了这个注释：“如果我将它放在一个简单的jar包中而不是一个Web服务（某种方式是一个servlet），它将不会出现任何错误”。 - Mehraban
@SAM 我已经阅读了这部分内容。我认为这可能是servlet或Java中的错误。我建议您更改代码，而不是修复代码中的问题，而是解决环境问题。 - Sergey Fedorov
0

你可以用 spark.scala.Function 更简单地替换你的 Java8 Lambda 函数。

替换

output = rdds.map(x->this.function(x)).collect()

使用：

output = rdds.map(new Function<Double,Double>(){

   public Double call(Double x){
       return MyClass.this.function(x);
   }

}).collect();
网页内容由stack overflow 提供, 点击上面的
可以查看英文原文，
原文链接