PySpark Spark + s3 – 错误 – java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found

在本文中，我们将解决在使用PySpark时遇到的一个常见错误：java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found。我们将介绍该错误的原因，并给出解决方案和示例代码。

阅读更多：PySpark 教程

错误原因

在使用PySpark与Amazon S3进行集成时，我们可能会遇到java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found的错误。这是因为在运行PySpark应用程序时，与Amazon S3集成所需的依赖项未正确加载。

解决方案

要解决这个错误，我们需要确保以下几点：

1. 添加必要的依赖项

为了与Amazon S3进行集成，我们需要添加几个必要的依赖项。在PySpark应用程序的代码中，我们可以通过通过SparkSession对象的config方法来添加这些依赖项。

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MyApp") \
    .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.2.0") \
    .getOrCreate()

以上代码中，我们通过配置spark.jars.packages属性来添加hadoop-aws依赖项。这个属性值的格式为groupId:artifactId:version。

2. 添加Hadoop配置

为了正确加载Amazon S3的文件系统类，我们还需要添加一些Hadoop配置。可以通过设置spark.hadoop.前缀的配置项来实现。

spark.conf.set("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
spark.conf.set("spark.hadoop.fs.s3a.access.key", "<your_access_key>")
spark.conf.set("spark.hadoop.fs.s3a.secret.key", "<your_secret_key>")

以上代码中，我们设置了spark.hadoop.fs.s3a.impl配置项为org.apache.hadoop.fs.s3a.S3AFileSystem，设置了spark.hadoop.fs.s3a.access.key和spark.hadoop.fs.s3a.secret.key配置项为你的访问密钥和密钥。

3. 添加AWS SDK依赖项

除了上述的配置项外，我们还需要添加AWS SDK的依赖项。我们可以通过通过spark.jars.packages属性添加该依赖项。

spark = SparkSession.builder \
    .appName("MyApp") \
    .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.2.0,com.amazonaws:aws-java-sdk:1.11.999") \
    .getOrCreate()

以上代码中，我们通过配置spark.jars.packages属性来同时添加了hadoop-aws和aws-java-sdk依赖项。

示例代码

下面是一个使用了上述解决方案的示例代码：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("S3Integration") \
    .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.2.0,com.amazonaws:aws-java-sdk:1.11.999") \
    .getOrCreate()

spark.conf.set("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
spark.conf.set("spark.hadoop.fs.s3a.access.key", "<your_access_key>")
spark.conf.set("spark.hadoop.fs.s3a.secret.key", "<your_secret_key>")

df = spark.read.csv("s3a://bucket/path/to/file.csv")
df.show()

以上示例代码通过SparkSession对象创建了一个Spark应用程序，并使用了我们提到的解决方案。它从Amazon S3的CSV文件中读取数据，并展示了该数据。