从 S3 读取文件 - Pyspark

Question

我整个上午都在寻找这个问题的明确答案，但找不到任何可以理解的东西。我刚开始使用 pyspark（用 pip 安装），有一个简单的 .py 文件从本地存储读取数据，在本地做一些处理和写入结果。我目前正在使用：python my_file.py

我想要做的是： 使用 AWS S3 中的文件作为输入，将结果写入 AWS3 上的存储桶

我能够使用“boto3”创建一个存储桶和加载文件，但看到一些使用“spark.read.csv”的选项，我想使用它。

我尝试过的： 我尝试使用以下方法设置凭据：

spark = SparkSession.builder \
            .appName("my_app") \
            .config('spark.sql.codegen.wholeStage', False) \
            .getOrCreate()\

spark._jsc.hadoopConfiguration().set("fs.s3a.awsAccessKeyId", "my_key_id")
spark._jsc.hadoopConfiguration().set("fs.s3a.awsSecretAccessKey", "my_secret_key")

然后 ：

df = spark.read.option("delimiter", ",").csv("s3a://bucket/key/filename.csv", header = True)

但得到错误：

java.io.IOException: No FileSystem for scheme: s3a

问题 ：

• 我是否需要特别安装一些东西才能启用 pyspark S3？
• 我应该以某种方式打包我的代码并使用 pyspark 控制台运行一个特殊的命令吗？

谢谢大家，抱歉重复的问题

---

解决了 ：

解决方案如下：

要将本地 spark 实例链接到 S3，您必须将 aws-sdk 和 hadoop-sdk 的 jar 文件添加到您的类路径并使用以下命令运行您的应用程序：spark-submit --jars my_jars.jar

请注意您用于 SDK 的版本，并非所有版本都兼容：aws-java-sdk-1.7.4、hadoop-aws-2.7.4 对我有用。

我使用的配置是：

spark = SparkSession.builder \
            .appName("my_app") \
            .config('spark.sql.codegen.wholeStage', False) \
            .getOrCreate()

spark._jsc.hadoopConfiguration().set("fs.s3a.access.key", "mykey")
spark._jsc.hadoopConfiguration().set("fs.s3a.secret.key", "mysecret")
spark._jsc.hadoopConfiguration().set("fs.s3a.impl","org.apache.hadoop.fs.s3a.S3AFileSystem")
spark._jsc.hadoopConfiguration().set("com.amazonaws.services.s3.enableV4", "true")
spark._jsc.hadoopConfiguration().set("fs.s3a.aws.credentials.provider","org.apache.hadoop.fs.s3a.BasicAWSCredentialsProvider")
spark._jsc.hadoopConfiguration().set("fs.s3a.endpoint", "eu-west-3.amazonaws.com")