Spark保存Delta Lake数据到Minio时触发ClassCastException问题求助

核心原因 #

1. PySpark与集群Spark版本不一致
   本地安装的pyspark 3.2.3和集群运行的Apache Spark 3.3.1版本不匹配。Spark不同版本对Scala函数（scala.Function1）与Java Lambda（java.lang.invoke.SerializedLambda）的序列化/反序列化逻辑存在差异，这种版本错位会导致跨进程（Driver与Executor）的数据传输时出现类型转换异常。

2. Delta Lake版本与Spark版本不兼容
   delta-spark 2.0.2官方适配的是Spark 3.2.x系列，而集群使用的是Spark 3.3.1，版本不匹配会导致Delta内部依赖的函数接口、序列化逻辑与Spark集群环境冲突，进一步触发该类型转换错误。

解决方案 #

1. 统一PySpark与集群Spark版本 #

将本地PySpark版本升级至与集群完全一致的3.3.1：

pip install pyspark==3.3.1 --force-reinstall

2. 匹配Delta Lake与Spark版本 #

更换为适配Spark 3.3.x的Delta Lake版本（推荐delta-spark 2.2.0，官方明确支持Spark 3.3.x）：

pip install delta-spark==2.2.0 --force-reinstall

3. 提交作业时确保依赖一致性 #

使用spark-submit提交作业时，显式指定对应版本的Delta Lake核心jar包，避免集群端依赖冲突：

spark-submit \
  --packages io.delta:delta-core_2.12:2.2.0 \
  --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" \
  your_script.py

4. 验证环境配置 #

在代码初始化SparkSession时，确保Delta相关配置正确加载：

from pyspark.sql import SparkSession
from delta import *

builder = SparkSession.builder.appName("DeltaExample") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = configure_spark_with_delta_pip(builder).getOrCreate()

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