Apache Sparkjava.lang.ClassCastException在远程主节点中运行forEach分区时[重复]

Question

我有一个Java的微服务，它连接到Apache Spark集群，并使用Datastax Spark-Cassandra连接器将数据持久化到apache Cassandra DB集群。

我编写了以下方法来从Cassandra表中删除特定日期范围的数据。

具体代码如下所示：

public void deleteData(String fromDate, String toDate) {


    SparkConf conf = sparkSession.sparkContext().getConf();
    CassandraConnector connector = CassandraConnector.apply(conf);

    Dataset<Row> df = sparkSession.read().format("org.apache.spark.sql.cassandra").options(new HashMap<String, String>() {{
        put("keyspace", CassandraProperties.KEYSPACE);
        put("table", CassandraProperties.ENERGY_FORECASTS);
    }}).load()
            .filter(col("timestamp")
                    .substr(1, 10)
                    .between(fromDate, toDate))
            .select("nodeid");


    df.foreachPartition(partition -> {
        Session session = connector.openSession();
        while (partition.hasNext()) {
            Row row = partition.next();
            session.execute("DELETE FROM " + CassandraProperties.KEYSPACE + "." + CassandraProperties.ENERGY_FORECASTS + " WHERE nodeid = '" + row.mkString() + "' AND timestamp >= '" + fromDate + "' AND timestamp <= '" + toDate + "'");
        }
        session.close();
    });
}

}

    @Bean
public SparkSession sparkSession() {
    return SparkSession
            .builder()
            .appName("SparkCassandraApp")
            .config("spark.cassandra.connection.host", host)
            .config("spark.cassandra.connection.port", port)
            .config("spark.sql.caseSensitive", false)
            .master(master)
            .getOrCreate();

使用本地spark master节点（.master（“local[*]”）选项）运行时，代码执行良好。

但是，当我在连接到远程火花主节点时尝试执行相同的代码时，会发生以下错误：

驱动程序堆栈跟踪：]与根本原因tream.java:2069ClassCastException：无法将java.io.SerializedLambda的实例分配到字段tream.readDataset$$anonfuns$foreach分区$tream.java:1573类型的4美元java.io.ForeachPartionFunction在实例tream.defaultDataset$$anonfuns$foreachPartion2美元在tream.java:2287ObjectStreamClass$FieldRjava.io.ObjFieldValue（ObjectStreamCtream.read）在java.io.ObjectStreamClass.setObjFieldValue（ObjectStreamClass.java:1417）在java.io.ObjectInputStream.defaultReadFields（ObjectInputStream.java:2293）在java.io.ObjectInputStream.readSerialData（ObjectInputStream.java:2211）在java.io.ObjectInputStream.read普通对象（ObjectInputSjava.lang.）在java.lang.invoke.ObjectInputSorg.apache.spark.sql.Object0（ObjectInputS2.func）在org.apache.spark.api.java.function.ObjectInputSorg.apache.spark.sql.ReadFields（ObjectInputSjava.io.）在eflector.setObjectInputSlass.java:2287Serjava. io处的ObjectInputStream. read普通对象（ObjectInputStream. java： 2069）。java. io处的ObjectInputStream. readObject0（ObjectInputStream. java： 1573）。org. apache. spak.序列化器处的ObjectInputStream. readObject（ObjectInputStream. java： 431）。org. apache. spak.序列化器处的JavaDeserializationStream. readObject（JavaSerializer. scala： 75）。org. apache. spak.调度器处的JavaSerializerInstance.反序列化（JavaSerializer. scala： 114）。org. apache. spak.调度器处的结果任务.运行任务（任务. scala： 83）。org. apache. spak.执行器处的任务.运行（任务. scala： 123）。执行器$任务

更新1

似乎对我有用的技巧是在spik会话配置中添加以下行：

.config("spark.jars", "meter-service-1.0.jar")

这似乎提供了缺失的依赖项，这些依赖项阻止Spark在远程节点上正确反序列化lamda表达式。

这在这里解释得更好




Answer 1

我的JAVA很脆弱，但您能否尝试将lambda提取到方法中？

public void deleteData(String fromDate, String toDate) {
    SparkConf conf = sparkSession.sparkContext().getConf();
    CassandraConnector connector = CassandraConnector.apply(conf);

    Dataset<Row> df = sparkSession.read().format("org.apache.spark.sql.cassandra").options(new HashMap<String, String>() {{
        put("keyspace", CassandraProperties.KEYSPACE);
        put("table", CassandraProperties.ENERGY_FORECASTS);
    }}).load()
        .filter(col("timestamp")
                .substr(1, 10)
                .between(fromDate, toDate))
        .select("nodeid");


    df.foreachPartition(new ForeachPartitionFunction<Row>() {
        public void call(Iterator<Row> partition) {
            Session session = connector.openSession();
            while (partition.hasNext()) {
                Row row = partition.next();
                session.execute("DELETE FROM " + CassandraProperties.KEYSPACE + "." + CassandraProperties.ENERGY_FORECASTS + " WHERE nodeid = '" + row.mkString() + "' AND timestamp >= '" + fromDate + "' AND timestamp <= '" + toDate + "'");
            }
            session.close();
        }
    });
}