Dataproc集群PySpark调用DeltaTable.forPath报方法不存在错误

问题重现 #

在Dataproc集群上运行PySpark作业时，调用DeltaTable.forPath(sparkSession, path)读取Delta表抛出以下错误：

Traceback (most recent call last):
  File "/tmp/job-0eb2543e/cohort_ka.py", line 146, in <module>
    main()
  File "/tmp/job-0eb2543e/cohort_ka.py", line 128, in main
    persisted = DeltaTable.forPath(spark, destination)
  File "/opt/conda/default/lib/python3.8/site-packages/delta/tables.py", line 387, in forPath
    jdt = jvm.io.delta.tables.DeltaTable.forPath(jsparkSession, path, hadoopConf)
  File "/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 1304, in __call__
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 111, in deco
  File "/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/protocol.py", line 330, in get_return_value
py4j.protocol.Py4JError: An error occurred while calling z:io.delta.tables.DeltaTable.forPath. Trace:
py4j.Py4JException: Method forPath([class org.apache.spark.sql.SparkSession, class java.lang.String, class java.util.HashMap]) does not exist
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:339)
    at py4j.Gateway.invoke(Gateway.java:276)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:750)

环境配置：

• Dataproc集群镜像：2.0-debian10
• Delta Java jar版本：delta-core_2.12-1.0.0.jar
• Spark版本：3.1

错误原因 #

报错核心是Python端的Delta库尝试调用Java端DeltaTable.forPath(SparkSession, String, HashMap)方法，但该重载在Delta 1.0.0版本中不存在。主要原因是Python Delta包版本与Java Delta jar版本不匹配：如果Python安装的Delta包版本高于1.0.0，它会尝试调用新增的方法重载，但Java端的1.0.0 jar并没有这个方法，导致方法不存在错误。

解决方案 #

1. 对齐Python Delta包与Java jar版本 #

确保Python环境中的Delta包版本和集群使用的Java jar版本完全一致（即1.0.0）：

# 在集群的Python环境中安装指定版本的Delta包
pip install delta-spark==1.0.0

2. 验证作业提交时的依赖配置 #

提交PySpark作业时，确保正确指定Delta的Java jar，避免依赖冲突：

gcloud dataproc jobs submit pyspark your_job.py \
    --cluster=your-cluster-name \
    --jars=gs://your-bucket/delta-core_2.12-1.0.0.jar \
    --properties="spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension,spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

3. 简化DeltaTable.forPath调用 #

如果版本对齐后仍有问题，尝试直接使用仅传递sparkSession和path的重载（避免Python库自动添加多余的hadoopConf参数）：

from delta.tables import DeltaTable

# 直接调用双参数重载
delta_table = DeltaTable.forPath(spark, destination)

或者先通过Spark读取表再转为DeltaTable：

df = spark.read.format("delta").load(destination)
delta_table = DeltaTable.forDF(spark, df)

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