java.lang.ClassCastException 在远程服务器上的 spark 作业中使用 lambda 表达式
java.lang.ClassCastException using lambda expressions in spark job on remote server
我正在尝试使用 sparkjava.com 框架为我的 apache spark 作业构建一个网络 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.f of type org.apache.spark.api.java.function.Function in instance of org.apache.spark.api.java.JavaRDD$$anonfun$filter
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)
P.S：我尝试将 jersey 和 javaspark 与 jetty、tomcat 和 resin 结合使用，所有这些都让我得到了相同的结果。
我想你的问题是自动装箱失败。在代码中
x -> {
return true;
}
你传递 (String->boolean) lambda（它是 Predicate<String>）而 filter method 传递 (String->Boolean) lambda（它是 Function<String,Boolean>）。所以我建议您将代码更改为
x -> {
return Boolean.TRUE;
}
请在您的问题中包含详细信息。感谢 uname -a 和 java -version 的输出。如果可能，请提供 sscce。
你在这里看到的是一个后续错误，它掩盖了原始错误。
当 lambda 实例被序列化时，它们使用 writeReplace 来解散它们的 JRE 特定
从 SerializedLambda 的持久形式实现
实例。 SerializedLambda 实例恢复后，将调用其 readResolve 方法
重构适当的 lambda 实例。正如文档所说，它将通过调用定义原始 lambda 的 class 的特殊方法来实现（另请参见 ）。重要的一点是需要原始 class，而这正是您的情况所缺少的。
但是 ObjectInputStream 有一个……特殊的……行为。当它遇到异常时，它不会立即退出。它将记录异常并继续该过程，标记当前正在读取的所有对象，从而将错误的对象也视为错误。只有在过程结束时，它才会抛出它遇到的原始异常。奇怪的是它还会继续尝试设置这些对象的字段。但是当你查看方法 ObjectInputStream.readOrdinaryObject line 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);
}
}
编译这四个class和运行Writing。然后删除class文件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 表达式的 class 是在反序列化 lambda 的 运行 时间也可用。
直接从 IDE 到 运行 的 Spark 作业示例（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");
我遇到了同样的错误，我用内部 class 替换了 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() {
.....
}
};
您可以更简单地用 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();
我正在尝试使用 sparkjava.com 框架为我的 apache spark 作业构建一个网络 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.f of type org.apache.spark.api.java.function.Function in instance of org.apache.spark.api.java.JavaRDD$$anonfun$filter
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)
P.S：我尝试将 jersey 和 javaspark 与 jetty、tomcat 和 resin 结合使用，所有这些都让我得到了相同的结果。
我想你的问题是自动装箱失败。在代码中
x -> {
return true;
}
你传递 (String->boolean) lambda（它是 Predicate<String>）而 filter method 传递 (String->Boolean) lambda（它是 Function<String,Boolean>）。所以我建议您将代码更改为
x -> {
return Boolean.TRUE;
}
请在您的问题中包含详细信息。感谢 uname -a 和 java -version 的输出。如果可能，请提供 sscce。
你在这里看到的是一个后续错误，它掩盖了原始错误。
当 lambda 实例被序列化时，它们使用 writeReplace 来解散它们的 JRE 特定
从 SerializedLambda 的持久形式实现
实例。 SerializedLambda 实例恢复后，将调用其 readResolve 方法
重构适当的 lambda 实例。正如文档所说，它将通过调用定义原始 lambda 的 class 的特殊方法来实现（另请参见
但是 ObjectInputStream 有一个……特殊的……行为。当它遇到异常时，它不会立即退出。它将记录异常并继续该过程，标记当前正在读取的所有对象，从而将错误的对象也视为错误。只有在过程结束时，它才会抛出它遇到的原始异常。奇怪的是它还会继续尝试设置这些对象的字段。但是当你查看方法 ObjectInputStream.readOrdinaryObject line 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);
}
}
编译这四个class和运行Writing。然后删除class文件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 表达式的 class 是在反序列化 lambda 的 运行 时间也可用。
直接从 IDE 到 运行 的 Spark 作业示例（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");
我遇到了同样的错误，我用内部 class 替换了 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() {
.....
}
};
您可以更简单地用 spark.scala.Function
替换
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();