Spark任务序列化失败异常排查求助

问题原因 #

调用finalEventsDF.first()时，代码在Driver端执行，不需要将TestModel实例序列化传输；但map算子是分布式执行逻辑，会把闭包内的TestModel实例序列化后发送到各个Executor节点，若TestModel未实现序列化接口，就会触发Task not serializable异常。

解决方案 #

1. 让TestModel实现序列化接口 #

直接修改TestModel类，实现java.io.Serializable接口，这是最直接的解决方式：

public class TestModel implements java.io.Serializable {
    // 原有类逻辑
}

2. 在Executor端实例化TestModel #

如果无法修改TestModel类，或不想让整个类序列化，可在map/mapPartitions算子内部创建TestModel实例，避免序列化Driver端的对象：

• 用map（每条数据创建一次实例，适合轻量实例）：

finalEventsDF.map { row =>
    val model = new TestModel()
    model.scoreModelRequest(row)
}

• 用mapPartitions（每个分区创建一次实例，适合创建代价高的实例，如加载模型文件）：

finalEventsDF.mapPartitions { iter =>
    val model = new TestModel()
    iter.map(row => model.scoreModelRequest(row))
}

3. 将scoreModelRequest改为静态方法 #

如果scoreModelRequest不依赖TestModel的实例状态，可将其改为静态方法，调用时无需持有实例：

public class TestModel {
    public static ResultType scoreModelRequest(Row row) {
        // 方法逻辑
    }
}

在Spark代码中直接调用静态方法：

finalEventsDF.map(row => TestModel.scoreModelRequest(row))

4. 使用广播变量复用Driver端实例 #

若必须在Driver端创建TestModel实例（如加载全局配置），可将其包装为广播变量，自动处理序列化且每个Executor仅接收一次：

val broadcastModel = spark.sparkContext.broadcast(new TestModel())
finalEventsDF.map { row =>
    broadcastModel.value.scoreModelRequest(row)
}

注意：此方式仍要求TestModel实现Serializable接口。

内容的提问来源于stack exchange，提问作者Vijayr32