方案没有文件系统：s3 在读取 parquet s3 文件时使用 pyspark

Question

我有一个存储桶，里面有一些小的 Parquet 文件，我想将它们合并成一个更大的文件。

为了完成这项任务，我想创建一个 spark 作业来消耗和写入一个新文件。

from pyspark import SparkContext
from pyspark.sql import SparkSession, SQLContext

spark = SparkSession.builder \
                    .master("local") \
                    .appName("Consolidated tables") \
                    .getOrCreate()

spark._jsc.hadoopConfiguration().set("fs.s3a.access.key", "access")
spark._jsc.hadoopConfiguration().set("fs.s3a.secret.key", "secret")

df = spark.read.parquet("s3://lake/bronze/appx/contextb/*")

此代码向我抛出异常：No FileSystem for scheme: s3。如果我切换到s3a://...，则会收到错误：Class org.apache.hadoop.fs.s3a.S3AFileSystem not found。

我正在尝试将此代码作为python myfile.py 运行。

有什么问题吗？

Answer 1

下载此 hadoop-aws-2.7.5.jar（或最新版本）并配置此 jar 可用于 spark

spark = SparkSession \
        .builder \
        .config("spark.jars", "/path/to/hadoop-aws-2.7.5.jar")\
        .getOrCreate()

Answer 2

from boto3.session import Session
from pyspark import SparkContext
from pyspark.sql import SparkSession, SQLContext

spark = SparkSession.builder \
                    .master("local") \
                    .appName("Consolidated tables") \
                    .getOrCreate()

ACCESS_KEY='your_access_key'
SECRET_KEY='your_secret_key'

session = Session(aws_access_key_id=ACCESS_KEY,
                  aws_secret_access_key=SECRET_KEY)
s3 = session.resource('s3')

df = spark.read.parquet("s3://lake/bronze/appx/contextb/*")