当前位置: 首页 > 面试题库 >

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

郭子航
2023-03-14
问题内容

我正在尝试使用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)

PS:我尝试过将jersey和javaspark与码头,tomcat和resin结合使用,所有这些使我得出了相同的结果。


问题答案:

您在这里遇到的是一个后续错误,该错误掩盖了原始错误。

序列化lambda实例时,它们用于writeReplace从作为SerializedLambda
实例的持久形式中分解其JRE特定实现。当SerializedLambda实例已恢复,它的readResolve方法将被调用来重建适当的拉姆达实例。如文档所述,它将通过调用定义原始lambda的类的特殊方法来做到这一点。重要的一点是,需要原始类,而这正是您的情况所缺少的。

但是,……有……的特殊行为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;
}

您会看到报告异常readResolve时它不会调用该方法。但是,如果没有发生替换,那么继续尝试设置引荐来源网址的字段值不是一个好主意,但这正是此处发生的情况,因此生成了一个。lookupException``null``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");


 类似资料:
  • 我使用Spring批处理管理来管理和监视作业和执行。如何调用一个作业并从一个独立的java应用程序启动它,该应用程序具有到包含Spring Batch Admin WebApp的服务器的给定HTTP连接。 谢谢你的帮助

  • 当在远程服务器上运行脚本时,我需要防止远程服务器的nohup输出显示在ssh服务器上。 我正在命令行下面运行,以便在远程服务器上运行脚本。 但是当它运行时,它不可预测地显示“远程服务器脚本”中的命令的nohup输出。下面是脚本的节选,您可以看到该命令应该在后台运行。 我如何使它在远程服务器上的运行方式与在家庭服务器上的运行方式相同?我真的很感谢你的帮助。

  • 问题内容: 我正在开发一个用Swift编写的iOS应用程序,该应用程序可与本地网络上的HTTP服务器通信,并且我正在使用Apple的Reachability类来确定运行HTTP服务器的远程计算机是否在线。这是代码: 问题是无论远程计算机是联机还是脱机, 只要我连接到Wifi网络,总是如此。但是,当我关闭Wifi时,它会导致错误而不是真实。我在这里做错什么了吗,还是Reachability类与Swi

  • 我们有现有的Spring批处理应用程序,我们希望使其可扩展以在多个节点上运行。 例如,我希望在不同的工作服务器上运行作业。在这种情况下,我有一个作业a和一个作业B,在worker server的一个实例上运行作业a,在worker server的另一个实例中运行作业B。 在spring batch integration文档中,我找到了一个关于SETP远程分区的解释。然而,我不知道如何从主服务器启

  • 问题内容: 我的链接服务器设置正确,我可以执行以下查询。 插入[RemoteServer]。[表] SELECT * FROM [LocalServer]。[表] 但是,当我 在事务内* 执行相同的操作时 * 开始将TRAN插入[RemoteServer]。[Table] SELECT * FROM [LocalServer]。[Table] COMMIT TRAN 我收到类似的错误 链接服务器“

  • 我已经从(“)创建了一个cron作业quartz://myscheduler?cron=20 ?bean(LdapBean.class,“开始”);(间隔时间=20s) 现在从“start”方法开始,我想调用ldap服务器来获取特定用户的数据。从(“直接:开始”)。到(“ldap:ldapserver?base=ou=group,dc=dc1,dc=dc2”)。到(“bean:ldapBean”)