如果您使用SparkSession和SparkContext同时读取文件，然后使用**wholeTextFiles**方法循环访问s3目录，会怎么样？您可以在url中使用s3a连接器，它允许通过Hadoop从s3读取。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('S3Example').getOrCreate()

s3_bucket = 'your-bucket'
s3_path = f's3a://{s3_bucket}/my-directory/'

# List files S3
file_list = spark.sparkContext.wholeTextFiles(s3_path).map(lambda x: x[0]).collect()

for file_path in file_list:
    print(file_path)

请注意，上面我只检索了文件路径。如果你想要两者，你可以避免只提取文件路径（lambda中的x[0]），而得到两者。

file_tuple = spark.sparkContext.wholeTextFiles(s3_path)

您可以使用s3fs-pip install s3fs
可以试试下面的代码-

import s3fs

fs = s3fs.S3FileSystem(anon=True)
fs.ls('my-bucket')

也可以看到这个doc。
另一种方法是使用hadoopFile：

SparkContext.hadoopFile("s3a://bucket_name/prefix)

请参阅此doc。

除了所有其他的答案;
您仍然可以使用boto 3 lib来获取具有完整路径的对象列表，如下所示：

response = boto3.client('s3').list_objects_v2(Bucket=bucket, Prefix=prefix)
full_paths_list = list()
for obj in response["Contents"]:
    full_path = f's3://full_paths_list bucket}/{obj["Key"]}'
    full_paths_list.append(full_path)

现在，你有了文件列表，所以你可以像下面这样用pyspark阅读它们：

df = spark.read.parquet(*full_paths_list)

使用Spark/Hadoop，您可以处理和管理不同文件系统中的文件（s3，adls gen 1/2，本地文件系统，hdfs...）
您只需要在设置系统参数后使用HadoopConfiguration创建特定的fileSystem。对于S3，您必须设置端点、访问密钥和密钥

fileSystem = FileSystem.get(URI("s3_host:port"), HadoopConfiguration)

之后，您只需使用定义的方法listStatus、create、copyToLocalFile、exists等。

fileSystem.listStatus(Path('path in your s3 bucket'))

重要注意事项：为此，您必须使用JVM网关并导入必要的库：

sc._gateway.jvm.org.apache.hadoop.fs.FileSystem
sc._gateway.jvm.org.apache.hadoop.fs.Path
and the needed ones

你可以在pyspark中使用boto3。列出S3存储桶中的对象，并将它们读入PySpark DataFrame。
1.使用www.example.com方法遍历对象列表并将每个文件读入PySpark DataFramespark.read。
1.将每个DataFrame追加到列表中，然后使用reduce函数将所有 Dataframe 合并为一个。

import boto3
import pyspark.sql.functions as F
from pyspark.sql import SparkSession

# Create a SparkSession
spark = SparkSession.builder.appName("Read from S3").getOrCreate()

# Set up a boto3 client
s3 = boto3.client('s3')

# List all objects in the S3 bucket
bucket_name = "your-bucket-name"
prefix = "path-to-folder-within-bucket/"  # optional: if you want to read only files in a particular folder
objects = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)

# Read each file into a PySpark DataFrame
dataframes = []
for obj in objects['Contents']:
    file_name = obj['Key']
    s3_path = f's3://{bucket_name}/{file_name}'
    df = spark.read.option("header", True).csv(s3_path)
    dataframes.append(df)

# Union all dataframes into one
final_df = reduce(lambda a, b: a.union(b), dataframes)

# Show the final dataframe
final_df.show()