Pyspark无法覆盖S3中的CSV文件问题求助

问题根源 #

S3没有真正意义上的文件夹，所谓的“文件夹”只是对象键的前缀。当你删除某个前缀下的唯一文件时，这个前缀对应的“文件夹”标记（如果存在）会被自动移除。Spark的overwrite模式在处理这类场景时，可能会先删除现有文件，导致后续写入时找不到对应的前缀路径，从而抛出错误。

可行解决方案 #

1. 手动删除目标文件后再写入

不要依赖Spark的overwrite模式，先通过AWS SDK（比如boto3）删除S3上的目标文件，再执行写入操作：

import boto3

# 初始化S3客户端
s3 = boto3.client('s3')
bucket_name = '你的存储桶名称'
target_file_key = '目标文件的完整键，比如folder/your_file.csv'

# 删除目标文件
s3.delete_object(Bucket=bucket_name, Key=target_file_key)

# 执行Spark写入（用append模式即可，因为文件已删除）
df.coalesce(1) \
  .write \
  .mode("append") \
  .option("header", "true") \
  .csv("s3://bucket_name/folder/")

如果要清理整个前缀下的所有文件，可以遍历删除：

# 获取前缀下的所有对象
response = s3.list_objects_v2(Bucket=bucket_name, Prefix='folder/')
if 'Contents' in response:
    for obj in response['Contents']:
        s3.delete_object(Bucket=bucket_name, Key=obj['Key'])

2. 直接指定单个文件路径写入

Spark默认写入文件夹生成part文件，但如果明确要写入单个文件，可以直接指定具体的S3文件路径，配合overwrite模式：

df.coalesce(1) \
  .write \
  .mode("overwrite") \
  .option("header", "true") \
  .csv("s3://bucket_name/folder/your_file.csv")

注意：这种方式需要确保Spark版本和Hadoop AWS客户端版本兼容，部分旧版本可能不支持直接写入单个文件路径。

3. 临时写入后替换目标文件

先将数据写入临时文件夹，再把临时文件夹中的part文件复制（覆盖）到目标路径，最后清理临时文件：

import boto3

# 写入临时文件夹
temp_path = "s3://bucket_name/temp_folder/"
df.coalesce(1) \
  .write \
  .mode("overwrite") \
  .option("header", "true") \
  .csv(temp_path)

# 获取临时文件夹中的CSV文件
s3 = boto3.client('s3')
bucket_name = '你的存储桶名称'
response = s3.list_objects_v2(Bucket=bucket_name, Prefix=temp_path)
part_file_key = [obj['Key'] for obj in response['Contents'] if obj['Key'].endswith('.csv')][0]

# 复制并覆盖目标文件
target_key = 'folder/your_file.csv'
s3.copy_object(
    Bucket=bucket_name,
    Key=target_key,
    CopySource={'Bucket': bucket_name, 'Key': part_file_key},
    MetadataDirective='REPLACE'
)

# 清理临时文件
for obj in response['Contents']:
    s3.delete_object(Bucket=bucket_name, Key=obj['Key'])

4. 调整Spark S3相关配置

更新Hadoop AWS客户端到最新版本，并添加以下Spark配置，优化S3路径处理逻辑：

spark.conf.set("fs.s3a.create.dir.permission", "777")
spark.conf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
spark.conf.set("fs.s3a.path.style.access", "true")

内容的提问来源于stack exchange，提问作者t1808