java.lang.ClassNotFoundException：用于 Spark 3.0.0 的 org.apache.spark.sql.sources.v2.DataSourceV2

Posted

技术标签:

【中文标题】java.lang.ClassNotFoundException：用于 Spark 3.0.0 的 org.apache.spark.sql.sources.v2.DataSourceV2【英文标题】：java.lang.ClassNotFoundException: org.apache.spark.sql.sources.v2.DataSourceV2 for Spark 3.0.0 【发布时间】：2020-07-29 08:14:53 【问题描述】：

简介

有哪些可能的路径可以让我通过 pyspark 3.0.0 从纯 pip 安装成功处理数据，至少在不降级 Spark 版本的情况下加载数据？ p>

当我尝试加载 parquet 和 csv 的数据集时，我会收到异常消息，因为 Exception Message 下面的内容会显示。 Spark会话的初始化很好，但是当我想加载数据集时，它就出错了。

一些信息

Java：openjdk 11 Python：3.8.5 模式：本地模式 操作系统：Ubuntu 16.04.6 LTS 注意事项：
    我执行了python3.8 -m pip install pyspark 来安装 Spark。 当我查找spark-sql_2.12-3.0.0.jar的jar（在Python站点包路径下，即~/.local/lib/python3.8/site-packages/pyspark/jars在我的情况下）时，spark.sql.sources下没有v2，我最相似的一个在同一个包下找到了一个叫DatSourceRegister的接口。 我在 *** 上发现的最相似的问题是 PySpark structured Streaming + Kafka Error (Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.sources.v2.StreamWriteSupport )，在该页面上的信息中建议降级 Spark 版本。

异常信息

Py4JJavaError: An error occurred while calling o94.csv.
: java.lang.NoClassDefFoundError: org/apache/spark/sql/sources/v2/DataSourceV2
    at java.base/java.lang.ClassLoader.defineClass1(Native Method)
    at java.base/java.lang.ClassLoader.defineClass(ClassLoader.java:1016)
    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:575)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)
    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:1209)
    at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.hasNextService(ServiceLoader.java:1220)
    at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.hasNext(ServiceLoader.java:1264)
    at java.base/java.util.ServiceLoader$2.hasNext(ServiceLoader.java:1299)
    at java.base/java.util.ServiceLoader$3.hasNext(ServiceLoader.java:1384)
    at scala.collection.convert.Wrappers$JIteratorWrapper.hasNext(Wrappers.scala:43)
    at scala.collection.Iterator.foreach(Iterator.scala:941)
    at scala.collection.Iterator.foreach$(Iterator.scala:941)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
    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:255)
    at scala.collection.TraversableLike.filterImpl$(TraversableLike.scala:249)
    at scala.collection.AbstractTraversable.filterImpl(Traversable.scala:108)
    at scala.collection.TraversableLike.filter(TraversableLike.scala:347)
    at scala.collection.TraversableLike.filter$(TraversableLike.scala:347)
    at scala.collection.AbstractTraversable.filter(Traversable.scala:108)
    at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:644)
    at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSourceV2(DataSource.scala:728)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:230)
    at org.apache.spark.sql.DataFrameReader.csv(DataFrameReader.scala:705)
    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:357)
    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.GatewayConnection.run(GatewayConnection.java:238)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.sources.v2.DataSourceV2
    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:521)
    ... 45 more

【问题讨论】：

pyspark 与 Spark 本身捆绑在一起，您可能不需要单独安装它。另外，请使用pip list 命令检查您没有其他版本 @AlexOtt，谢谢提醒。我知道 Python 只是一个包装器，因为 Spark 的核心是由 Scala 编写的，而 pyspark 实际上是通过 py4j 操作 Java 对象来工作的。我在帖子中提到的查找 jar 的一点是通过在 $SPARK_HOME/jars 下检查它，并且 $SPARK_HOME 在 Python 站点包路径下，在我的例子中是 ~/.local/lib/python3.8/site-packages/pyspark。 您是否使用了没有 hadoop 的 spark 3.0.0 二进制文件？我能够在能够读取文件的 macbook 上 brew 安装本地 spark 3.0.0。如果我在 Ubuntu 集群上使用没有 hadoop 的 spark 二进制调用 spark_session.read.parquet()，我会遇到同样的问题。虽然我找不到的 java 类是java.lang.NoClassDefFoundError: org/apache/spark/sql/sources/v2/ReadSupport。所以我想知道这是否是spark-3.0.0-bin-without-hadoop.tgz 二进制文件的可重现问题。 @bricard 仅尝试通过 pip 安装并从源代码构建。也许我可以尝试下载你提到的二进制文件，通过运行独立模式检查我是否会遇到相同或类似的问题 【参考方案1】：

我在使用 spark 3 时遇到了同样的问题，终于找到了原因。我包含了一个依赖于旧数据源 v2 api 的自定义 jar。

解决方案是移除自定义 jar，然后 spark 开始正常工作。

【讨论】：

我前几天搭建了一个Spark集群，单机模式，由1个master和4个worker组成。我下载了spark-3.0.0-bin-hadoop3.2.tgz 进行构建。而且我可以从 AWS S3 检索 Spark 支持的格式的数据，而不会出现异常消息或类似消息。【参考方案2】：

目前，我已经找到了通过 Python 函数 API 为 Spark 操作数据的方法。

解决方法

1

# clone a specific branch 
git clone -b branch-3.0 --single-branch https://github.com/apache/spark.git
## could try the follwoing command
## git clone --branch v3.0.0 https://github.com/apache/spark.git

# build a Spark distribution
cd spark
./dev/make-distribution.sh --name spark3.0.1 --pip --r --tgz -e -PR -Phive -Phive-thriftserver -Pmesos -Pyarn -Dhadoop.version=3.0.0 -DskipTests -Pkubernetes
## after changing the value of SPARK_HOME in `.bashrc_profile`
source ~/.bashrc_profile

# downlaod needed additional jars into the directory
cd $SPARK_HOME/assembly/target/scala-2.12/jars
curl -O https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.0.0/hadoop-aws-3.0.0.jar
curl -O https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.11.828/aws-java-sdk-bundle-1.11.828.jar
cd $SPARK_HOME

# add related configuraionts for Spark
cp $SPARK_HOME/conf/spark-defaults.conf.template $SPARK_HOME/conf/spark-defaults.conf
## add required or desired parameters into the `spark-defaults.conf`
## as of me, I edited the configuraion file by `vi`

# launch an interactive shell
pyspark
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 3.0.1-SNAPSHOT
      /_/

Using Python version 3.8.5 (default, Jul 24 2020 05:43:01)
SparkSession available as 'spark'.
## after launching, I can read parquet and csv files without the exception

2 设置好上面的所有东西后，将$SPARK_HOME/python添加到环境变量PYTHONPATH，然后记得source相关文件（我添加到.bashrc_profile）。

from pyspark import SparkConf
from pyspark.sql import SparkSession
sc = SparkConf()
threads_max = 512
connection_max = 600
sc.set("spark.driver.memory", "10g")
sc.set('spark.hadoop.fs.s3a.threads.max', threads_max)
sc.set('spark.hadoop.fs.s3a.connection.maximum', connection_max)
sc.set('spark.hadoop.fs.s3a.aws.credentials.provider',
           'com.amazonaws.auth.EnvironmentVariableCredentialsProvider')
sc.set('spark.driver.maxResultSize', 0)
spark = SparkSession.builder.appName("cest-la-vie")\
    .master("local[*]").config(conf=sc).getOrCreate()
## after launching, I can read parquet and csv files without the exception

注释

我还尝试让 PySpark pip 可以从源代码的建筑物中安装，但我被困在将文件大小上传到 testpypi 上。这个尝试是我希望 pyspark 包出现在站点包目录下。以下是我的尝试步骤：

cd $SPARK_HOME/python
# Step 1
python3.8 -m pip install --user --upgrade setuptools wheel
# Step 2
python3.8 setup.py sdist bdist_wheel ## /opt/spark/python
# Step 3
python3.8 -m pip install --user --upgrade twine
# Step 4
python3.8 -m twine upload --repository testpypi dist/*
## have registered an account for testpypi and got a token
Uploading pyspark-3.0.1.dev0-py2.py3-none-any.whl

## stuck here
100%|██████████████████████████████████████████████████████████████████████████████████████████████████████████████| 345M/345M [00:49<00:00, 7.33MB/s]
Received "503: first byte timeout" Package upload appears to have failed.  Retry 1 of 5

【讨论】：

【参考方案3】：

我使用的是 Spark 3.1.1 的独立安装。

我尝试了很多东西。

我已经排除了很多jar文件。

在经历了很多痛苦之后，我决定删除我的 Spark 安装并安装（解压缩）一个新的。

我不知道为什么......但它正在工作。

【讨论】：

以上是关于java.lang.ClassNotFoundException：用于 Spark 3.0.0 的 org.apache.spark.sql.sources.v2.DataSourceV2的主要内容，如果未能解决你的问题，请参考以下文章