如何在函数中用Boto3重命名Databricks PySpark输出的S3单CSV文件

方案1：使用Databricks原生dbutils.fs操作（推荐） #

在Databricks中直接用dbutils.fs无需额外配置存储权限（若已挂载存储），可快速完成文件重命名：

# 配置路径参数
source_dir = "/dbfs/mnt/your_mount/output_dir/"  # 支持挂载路径或直接s3://bucket/path/
new_file_name = "desired_name.csv"
new_file_path = f"{source_dir}{new_file_name}"

# 获取目录下所有文件
file_list = dbutils.fs.ls(source_dir)

# 筛选出以part-00000开头的CSV文件（coalesce(1)仅生成一个此类文件）
target_file = next(file for file in file_list if file.name.startswith("part-00000") and file.name.endswith(".csv"))

# 复制为新名称
dbutils.fs.cp(target_file.path, new_file_path)

# 删除原文件
dbutils.fs.rm(target_file.path)

补充校验逻辑（可选） #

若不确定目录下是否存在唯一符合条件的文件，可添加校验避免报错：

matching_files = [file for file in file_list if file.name.startswith("part-00000") and file.name.endswith(".csv")]
if len(matching_files) != 1:
    raise Exception("找到多个或零个目标文件，请检查路径")
target_file = matching_files[0]

方案2：使用Boto3操作S3存储 #

若需直接操作未挂载的S3桶，可通过Boto3遍历并筛选目标文件：

import boto3

# 配置S3参数
bucket_name = "your-bucket-name"
source_prefix = "path/to/output_dir/"  # 前缀末尾需带/，确保仅匹配目标目录
new_file_key = f"{source_prefix}desired_name.csv"

# 初始化S3客户端
s3_client = boto3.client("s3")

# 获取前缀下所有对象
response = s3_client.list_objects_v2(Bucket=bucket_name, Prefix=source_prefix)

# 筛选目标文件
matching_objs = [
    obj for obj in response["Contents"]
    if obj["Key"].startswith(f"{source_prefix}part-00000") and obj["Key"].endswith(".csv")
]
if len(matching_objs) != 1:
    raise Exception("找到多个或零个目标文件，请检查前缀")
old_file_key = matching_objs[0]["Key"]

# 复制并重命名
s3_client.copy_object(
    Bucket=bucket_name,
    Key=new_file_key,
    CopySource={"Bucket": bucket_name, "Key": old_file_key}
)

# 删除原文件
s3_client.delete_object(Bucket=bucket_name, Key=old_file_key)

注意事项 #

• 需确保Databricks集群实例角色拥有S3的ListBucket、CopyObject、DeleteObject权限。
• 目标目录层级较深时，需确认source_prefix的准确性，避免匹配错误对象。

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