Spark 3.4.1中Dataset reduce操作出现ClassCastException问题求助

将Apache Spark从3.2.1版本升级至3.4.1后，执行自定义case类的Dataset.reduce()操作时抛出java.lang.ClassCastException，该问题在旧版本中未出现。错误提示为CustomCaseClass cannot be cast to CustomCaseClass，属于同类型转换异常，非常令人困惑。

复现代码 #

定义case类：

case class CustomCaseClass(id: String, body: Map[String, Int])

执行逻辑：

val result = someDataset
  .groupByKey(ds => ds.id)(Encoders.STRING)
  .mapGroups((key, iter) => CustomCaseClass(key, Map(key -> iter.length)))(Encoders.product)
  .reduce((before, next) => before.copy(body = before.body ++ next.body))

错误信息 #

class threw exception: java.lang.ClassCastException: CustomCaseClass cannot be cast to CustomCaseClass

已尝试的无效方案 #

• 使用显式编码器
• 禁用AQE（Adaptive Query Execution）

问题解答 #

1. Spark 3.4.1中导致该ClassCastException的原因是什么？

这种同类型转换异常本质是类加载器不一致导致的。Spark 3.4.1在序列化/反序列化逻辑上有变更，尤其是在Dataset算子链中，当自定义case类被不同类加载器（如Spark任务类加载器与应用程序类加载器）加载时，JVM会判定它们为不同类型，从而抛出转换异常。

具体到你的代码，mapGroups算子生成的Dataset与后续reduce算子处理时，可能出现类加载上下文切换，导致CustomCaseClass被重复加载，实例无法跨类加载器转换。

2. 是否需要修改特定的序列化器选项？

需要。默认Java序列化无法处理类加载器不一致的问题，建议切换到Kryo序列化，并显式注册自定义case类，确保序列化/反序列化使用统一类定义。

修改配置的方式：

spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
spark.conf.set("spark.kryo.registrationRequired", "true")
spark.sessionState.conf.registerKryoClasses(Array(classOf[CustomCaseClass]))

3. 是否应该使用比Encoders.product更精确的编码器？如果是，该如何操作？

Encoders.product本身是适配case类的通用编码器，但Spark 3.4.1中编码器的类加载逻辑变更，可能导致与后续算子的类加载上下文不匹配。可以尝试显式创建case类的编码器，结合Kryo序列化解决类加载问题。

显式创建编码器的方式：

import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
val customEncoder: ExpressionEncoder[CustomCaseClass] = ExpressionEncoder()

在mapGroups中指定该编码器：

.mapGroups((key, iter) => CustomCaseClass(key, Map(key -> iter.length)))(customEncoder)

4. Spark 3.4.1中有哪些已知问题或变更可能导致此问题？

Spark 3.4.1的以下变更可能引发此类问题：

• Dataset算子代码生成优化：3.4.x版本对Dataset代码生成逻辑进行升级，可能导致类加载路径变化，尤其是算子链较长时。
• 类加载器隔离机制增强：为提升任务隔离性，Spark 3.4.1强化了任务类加载器的隔离策略，导致应用程序类加载器与任务类加载器加载的类无法互相转换。
• 相关已知问题：Spark社区JIRA中，SPARK-40012、SPARK-39858等问题均涉及类加载器导致的类型转换异常，集中在Dataset算子序列化环节。

5. 应如何修复此错误？

综合分析，推荐修复步骤：

1. 切换到Kryo序列化并注册自定义类：这是解决类加载不一致问题最直接的方案，配置方式参考问题2的代码。
2. 统一编码器的使用：在所有涉及自定义case类的Dataset算子中，显式指定同一个编码器，避免隐式编码器带来的类加载差异。
3. 检查类加载上下文：确保自定义case类的字节码正确打包到应用JAR中，避免Spark任务节点重复加载不同版本的类文件。
4. 调整Spark类加载模式：可设置spark.driver.userClassPathFirst=true和spark.executor.userClassPathFirst=true，让应用程序类加载器优先加载自定义类，但此配置可能引发依赖冲突，需谨慎使用。

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