如何解决PySpark 3.5.0连接Salesforce时的NoClassDefFoundError问题

问题概述 #

使用PySpark 3.5.0（Scala 2.12版本），通过spark-salesforce_2.12-1.1.4.jar驱动连接Salesforce时，频繁抛出NoClassDefFoundError，错误指向org/apache/spark/sql/sources/v2/ReadSupport类缺失，推测为驱动版本不匹配或依赖缺失导致。

连接代码 #

from pyspark.sql import SparkSession

jdbc_driver = "/home/azminds/pyspark/spark-3.5.0-bin-hadoop3/jars/spark-salesforce_2.12-1.1.4.jar"
spark = SparkSession.builder.master("local[*]") \
    .appName('Db2Connection') \
    .config('spark.driver.extraClassPath', jdbc_driver) \
    .getOrCreate()

try:
    soql = "SELECT * FROM account"  
    df = spark \
        .read \
        .format("com.springml.spark.salesforce") \
        .option("username", "arvind.azminds-uc6a@force.com") \
        .option("password", "password+securityToken") \
        .option("soql", soql) \
        .option("version", "56.0") \
        .load()

    df.show()
except Exception as Error:
    print("Error ==",Error)

错误堆栈信息 #

Error == An error occurred while calling o37.load.
: java.lang.NoClassDefFoundError: org/apache/spark/sql/sources/v2/ReadSupport
        at java.base/java.lang.ClassLoader.defineClass1(Native Method)
        at java.base/java.lang.ClassLoader.defineClass(ClassLoader.java:1022)
        at java.base/java.security.SecureClassLoader.defineClass(SecureClassLoader.java:174)
        at java.base/jdk.internal.loader.BuiltinClassLoader.defineClass(BuiltinClassLoader.java:800)
        at java.base/jdk.internal.loader.BuiltinClassLoader.findClassOnClassPathOrNull(BuiltinClassLoader.java:698)
        at java.base/jdk.internal.loader.BuiltinClassLoader.loadClassOrNull(BuiltinClassLoader.java:621)
        at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:579)
        at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:581)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:527)
        at java.base/java.lang.Class.forName0(Native Method)
        at java.base/java.lang.Class.forName(Class.java:398)
        at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.nextProviderClass(ServiceLoader.java:1210)
        at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.hasNextService(ServiceLoader.java:1221)
        at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.hasNext(ServiceLoader.java:1265)
        at java.base/java.util.ServiceLoader$2.hasNext(ServiceLoader.java:1300)
        at java.base/java.util.ServiceLoader$3.hasNext(ServiceLoader.java:1385)
        at scala.collection.convert.Wrappers$JIteratorWrapper.hasNext(Wrappers.scala:45)
        at scala.collection.Iterator.foreach(Iterator.scala:943)
        at scala.collection.Iterator.foreach$(Iterator.scala:943)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
        at scala.collection.IterableLike.foreach(IterableLike.scala:74)
        at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
        at scala.collection.TraversableLike.filterImpl(TraversableLike.scala:303)
        at scala.collection.TraversableLike.filterImpl$(TraversableLike.scala:297)
        at scala.collection.AbstractTraversable.filterImpl(Traversable.scala:108)
        at scala.collection.TraversableLike.filter(TraversableLike.scala:395)
        at scala.collection.TraversableLike.filter$(TraversableLike.scala:395)
        at scala.collection.AbstractTraversable.filter(Traversable.scala:108)
        at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:629)
        at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSourceV2(DataSource.scala:697)
        at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:208)
        at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:172)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:566)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374)
        at py4j.Gateway.invoke(Gateway.java:282)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
        at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.sources.v2.ReadSupport
        at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:581)
        at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:527)
        ... 46 more

解决方案 #

1. 更换兼容的驱动版本 #

spark-salesforce_2.12-1.1.4.jar是为Spark 2.x版本开发的，依赖的org.apache.spark.sql.sources.v2.ReadSupport类在Spark 3.x中已被移除或重构。需要选用与Spark 3.5.x兼容的spark-salesforce版本，确保Scala版本保持2.12，比如适配Spark 3.x的最新稳定版本。

2. 补充完整依赖包 #

spark-salesforce驱动依赖Salesforce Java SDK、HTTP客户端等多个第三方jar包，仅单独添加spark-salesforce jar会导致依赖缺失。需将所有依赖包放入Spark的jars目录，或通过spark.driver.extraClassPath和spark.executor.extraClassPath配置所有依赖的路径。

3. 调整ClassPath配置 #

• 本地模式：将所有依赖jar直接复制到Spark安装目录下的jars文件夹，Spark启动时会自动加载这些jar，无需手动配置extraClassPath。
• 集群模式：同时配置spark.driver.extraClassPath和spark.executor.extraClassPath，确保驱动和执行器都能获取到所有依赖包。

修正后的代码示例 #

假设已替换为兼容Spark 3.5的驱动及依赖包，代码可调整为：

from pyspark.sql import SparkSession

# 确保所有依赖jar都在Spark的jars目录或配置了正确的classpath
spark = SparkSession.builder.master("local[*]") \
    .appName('SalesforceConnection') \
    .getOrCreate()

try:
    soql = "SELECT * FROM account"  
    df = spark \
        .read \
        .format("com.springml.spark.salesforce") \
        .option("username", "arvind.azminds-uc6a@force.com") \
        .option("password", "password+securityToken") \
        .option("soql", soql) \
        .option("version", "56.0") \
        .load()

    df.show()
except Exception as Error:
    print("Error ==",Error)
finally:
    spark.stop()

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