K8s多Executor下保存Spark PipelineModel到S3报ClassFormatError

在Kubernetes集群中运行Spark作业，尝试将Spark PipelineModel保存至S3存储时出现稳定性问题：

• 提交作业时仅配置1个Executor，任务运行完全正常
• 配置2个及以上Executor时作业直接失败，抛出ClassFormatError异常，核心报错信息为Unknown constant tag 83 in class file org/apache/spark/ml/param/DoubleArrayParam，完整异常栈如下：

22/07/01 07:07:48 ERROR Instrumentation: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 4 times, most recent failure: Lost task 0.3 in stage 1.0 (TID 6) (10.0.2.190 executor 2): java.lang.ClassFormatError: Unknown constant tag 83 in class file org/apache/spark/ml/param/DoubleArrayParam
    at java.base/java.lang.ClassLoader.defineClass1(Native Method)
    at java.base/java.lang.ClassLoader.defineClass(Unknown Source)
    at java.base/java.security.SecureClassLoader.defineClass(Unknown Source)
    at java.base/jdk.internal.loader.BuiltinClassLoader.defineClass(Unknown Source)
    at java.base/jdk.internal.loader.BuiltinClassLoader.findClassOnClassPathOrNull(Unknown Source)
    at java.base/jdk.internal.loader.BuiltinClassLoader.loadClassOrNull(Unknown Source)
    at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(Unknown Source)
    at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(Unknown Source)
    at java.base/java.lang.ClassLoader.loadClass(Unknown Source)
    at java.base/java.lang.Class.getDeclaredFields0(Native Method)
    at java.base/java.lang.Class.privateGetDeclaredFields(Unknown Source)
    at java.base/java.lang.Class.getDeclaredField(Unknown Source)
    at java.base/java.io.ObjectStreamClass.getDeclaredSUID(Unknown Source)
    at java.base/java.io.ObjectStreamClass$2.run(Unknown Source)
    at java.base/java.io.ObjectStreamClass$2.run(Unknown Source)
    at java.base/java.security.AccessController.doPrivileged(Native Method)
    at java.base/java.io.ObjectStreamClass.<init>(Unknown Source)
    at java.base/java.io.ObjectStreamClass.lookup(Unknown Source)
    at java.base/java.io.ObjectStreamClass.initNonProxy(Unknown Source)
    at java.base/java.io.ObjectInputStream.readNonProxyDesc(Unknown Source)
    at java.base/java.io.ObjectInputStream.readClassDesc(Unknown Source)
    at java.base/java.io.ObjectInputStream.readNonProxyDesc(Unknown Source)
    at java.base/java.io.ObjectInputStream.readClassDesc(Unknown Source)
    at java.base/java.io.ObjectInputStream.readClass(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.defaultReadFields(Unknown Source)
    at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.readArray(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.defaultReadFields(Unknown Source)
    at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.defaultReadFields(Unknown Source)
    at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.defaultReadFields(Unknown Source)
    at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject(Unknown Source)
    at scala.collection.immutable.List$SerializationProxy.readObject(List.scala:527)
    at jdk.internal.reflect.GeneratedMethodAccessor3.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.base/java.lang.reflect.Method.invoke(Unknown Source)
    at java.base/java.io.ObjectStreamClass.invokeReadObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.defaultReadFields(Unknown Source)
    at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.defaultReadFields(Unknown Source)
    at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject(Unknown Source)
    at scala.collection.immutable.List$SerializationProxy.readObject(List.scala:527)
    at jdk.internal.reflect.GeneratedMethodAccessor3.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.base/java.lang.reflect.Method.invoke(Unknown Source)
    at java.base/java.io.ObjectStreamClass.invokeReadObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.defaultReadFields(Unknown Source)
    at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.defaultReadFields(Unknown Source)
    at java.base/java.io.ObjectInputStream.readSerialData(Unknown Source)
    at java.base/java.io.ObjectInputStream.readOrdinaryObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject0(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject(Unknown Source)
    at java.base/java.io.ObjectInputStream.readObject(Unknown Source)
    at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:76)
    at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:115)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:83)
    at org.apache.spark.scheduler.Task.run(Task.scala:131)
    at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1462)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.base/java.lang.Thread.run(Unknown Source)

使用Spark官方ML Pipeline示例代码运行时，可稳定复现该错误，示例代码如下：

from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import HashingTF, Tokenizer
from pyspark.sql import SparkSession
from pyspark import SparkContext, SparkConf

conf = SparkConf()

spark = SparkSession \
    .builder \
    .config(conf=conf) \
    .getOrCreate()

# 准备训练数据
training = spark.createDataFrame([
    (0, "a b c d e spark", 1.0),
    (1, "b d", 0.0),
    (2, "spark f g h", 1.0),
    (3, "hadoop mapreduce", 0.0)
], ["id", "text", "label"])

# 构建Pipeline：分词器、特征哈希、逻辑回归
tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="features")
lr = LogisticRegression(maxIter=10, regParam=0.001)
pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])

# 训练模型
model = pipeline.fit(training)

# 保存模型到S3
model.write().session(spark).overwrite().save('s3a://path/to/save')

额外测试验证：如果移除Pipeline中的Estimator组件（即示例中的逻辑回归分类器），仅保存预处理Transformer步骤，保存操作可正常执行。

运行环境 #

• Spark版本：3.2.1
• 额外引入Jar包：hadoop-aws 3.2.2、aws-java-sdk-bundle 1.11.375

该问题属于典型的Spark类加载冲突，触发逻辑如下：

1. 单Executor场景下，所有任务调度、模型序列化/反序列化、类加载操作都在同一个JVM进程内完成，类只会被加载一次，即使类路径存在重复类也不会出现跨进程的类版本不一致问题，因此作业可正常运行。
2. 多Executor场景下，模型保存是分布式执行的，Driver会将模型元数据序列化后分发到各个Executor，Executor需要加载DoubleArrayParam等MLlib核心类完成反序列化后再执行写S3操作。如果用户作业FatJar中重复打包了Spark内置的MLlib依赖（且打包过程中类文件被裁剪、损坏，或者版本与集群原生Spark版本不完全一致），加上类加载顺序配置为优先加载用户Jar，就会出现Executor加载到损坏/不匹配的类文件，抛出类格式错误。
3. 仅保存预处理步骤时不涉及模型训练生成的参数类（如DoubleArrayParam是逻辑回归等算法模型存储系数的核心参数类），不会触发对应类的跨节点反序列化，因此可以正常执行。

• 作业打包时严格排除所有Spark集群内置的provided依赖，包括spark-core、spark-sql、spark-mllib等，不要将这些依赖打入用户作业Jar，避免类文件重复或损坏。
• 检查Spark配置，将spark.driver.userClassPathFirst、spark.executor.userClassPathFirst设置为false，确保JVM优先加载Spark镜像原生自带的官方类，再加载用户上传的第三方依赖，避免用户Jar中的类覆盖原生类。
• 校验类路径下的S3依赖：当前使用的hadoop-aws 3.2.2、aws-java-sdk-bundle 1.11.375与Spark 3.2.1版本兼容，不需要调整版本，但要保证类路径下不存在其他版本的AWS SDK或hadoop-aws包，避免依赖冲突。

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