这个错误是最难追踪的。我不确定发生了什么。我在我的位置机器上运行Spark集群。所以整个spark集群在一个主机下,这个主机是127.0.0.1
,我在独立模式下运行
JavaPairRDD<byte[], Iterable<CassandraRow>> cassandraRowsRDD= javaFunctions(sc).cassandraTable("test", "hello" )
.select("rowkey", "col1", "col2", "col3", )
.spanBy(new Function<CassandraRow, byte[]>() {
@Override
public byte[] call(CassandraRow v1) {
return v1.getBytes("rowkey").array();
}
}, byte[].class);
Iterable<Tuple2<byte[], Iterable<CassandraRow>>> listOftuples = cassandraRowsRDD.collect(); //ERROR HAPPENS HERE
Tuple2<byte[], Iterable<CassandraRow>> tuple = listOftuples.iterator().next();
byte[] partitionKey = tuple._1();
for(CassandraRow cassandraRow: tuple._2()) {
System.out.println("************START************");
System.out.println(new String(partitionKey));
System.out.println("************END************");
}
这个错误是最难跟踪的。它显然发生在cassandraRowsRDD.collect()
,我不知道为什么?
16/10/09 23:36:21 ERROR Executor: Exception in task 2.3 in stage 0.0 (TID 21)
java.lang.ClassCastException: cannot assign instance of scala.collection.immutable.List$SerializationProxy to field org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$dependencies_ of type scala.collection.Seq in instance of org.apache.spark.rdd.MapPartitionsRDD
at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2133)
at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1305)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2006)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000)
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924)
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:75)
at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:114)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:85)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
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)
以下是我使用的版本
Scala code runner version 2.11.8 // when I run scala -version or even ./spark-shell
compile group: 'org.apache.spark' name: 'spark-core_2.11' version: '2.0.0'
compile group: 'org.apache.spark' name: 'spark-streaming_2.11' version: '2.0.0'
compile group: 'org.apache.spark' name: 'spark-sql_2.11' version: '2.0.0'
compile group: 'com.datastax.spark' name: 'spark-cassandra-connector_2.11' version: '2.0.0-M3':
我的gradle文件看起来像这样后,介绍了一些所谓的“提供”,实际上似乎并不存在,但谷歌说要创建一个,所以我的build.gradle看起来像这样
group 'com.company'
version '1.0-SNAPSHOT'
apply plugin: 'java'
apply plugin: 'idea'
repositories {
mavenCentral()
mavenLocal()
}
configurations {
provided
}
sourceSets {
main {
compileClasspath += configurations.provided
test.compileClasspath += configurations.provided
test.runtimeClasspath += configurations.provided
}
}
idea {
module {
scopes.PROVIDED.plus += [ configurations.provided ]
}
}
dependencies {
compile 'org.slf4j:slf4j-log4j12:1.7.12'
provided group: 'org.apache.spark', name: 'spark-core_2.11', version: '2.0.0'
provided group: 'org.apache.spark', name: 'spark-streaming_2.11', version: '2.0.0'
provided group: 'org.apache.spark', name: 'spark-sql_2.11', version: '2.0.0'
provided group: 'com.datastax.spark', name: 'spark-cassandra-connector_2.11', version: '2.0.0-M3'
}
jar {
from { configurations.provided.collect { it.isDirectory() ? it : zipTree(it) } }
// with jar
from sourceSets.test.output
manifest {
attributes 'Main-Class': "com.company.batchprocessing.Hello"
}
exclude 'META-INF/.RSA', 'META-INF/.SF', 'META-INF/*.DSA'
zip64 true
}
6条答案
按热度按时间mqxuamgl1#
我遇到了同样的问题,可以通过将我的应用程序的jar添加到spark的类路径中来解决
7fhtutme2#
我遇到了相同的异常,并深入研究了多个相关的Jira(9219、12675、18075)。
我认为异常名称容易混淆,真实的的问题是spark集群和驱动程序应用程序之间的不一致的环境设置**。
例如,我在
conf/spark-defaults.conf
中使用以下行启动了Spark集群:当我启动我的驱动程序时(即使程序是以
spark-submit
启动的),我使用了一行:其中
<master ip>
是节点master
的正确IP地址,但程序将由于这种简单的不一致而失败。因此,我建议所有驱动程序都使用
spark-submit
启动,并且不要在驱动程序代码中复制任何配置(除非您需要覆盖某些配置)。也就是说,让spark-submit
在运行的Spark集群中以相同的方式设置您的环境。baubqpgj3#
在我的例子中,我必须添加
spark-avro
jar(我把它放在主jar旁边的/lib
文件夹中):fivyi3re4#
你调用的()方法应该像下面这样返回byte[]。
如果您仍然遇到此问题,请检查Jira https://issues.apache.org/jira/browse/SPARK-9219中提到的依赖项的版本
brtdzjyr5#
检查您的代码-在Intellij中:分析... -〉检查代码。如果你有与序列化相关的被弃用的方法,请修复它。或者简单地尝试降低Scala的Spark版本。在我的例子中,我将Scala版本降低到2.10,所有的工作都正常。
sgtfey8w6#
当我在spark集群的一个节点上运行eclipse时,我遇到了同样的问题,这个节点是ubuntu box。我创建了UDF作为一个单独的java类。当在本地运行spark时,一切都很好,但是转向yarn时,抛出了与问题中相同的异常。
我通过将生成的类的路径放到spark classpath中来解决这个问题,spark classpath包含类似于Holger Brandl adviced的UDF类。
我为类路径创建了一个变量:
并作为配置变量添加到Spark:
编辑:
将路径添加到classpath只解决了驱动程序节点的问题,集群中的其他节点仍然可能会出现同样的错误。我得到的最终解决方案是在每次构建后将生成的类放入hdfs,并将classpath设置为用于spark的hdfs文件夹,如下所示。
请参见TheMP的答案