AWS Glue PySpark作业写入CSV到S3无法覆盖同名文件如何实现

你可以通过写入时开启覆盖模式 + 写入后调用S3 API重命名生成的单分区文件的方式实现每次运行覆盖同一份固定名称文件的需求，具体修改如下：

核心修改点 #

• 写入DynamicFrame时添加mode参数为overwrite，每次写入前清空目标路径下的历史文件，避免文件堆积
• 导入boto3操作S3，写入完成后扫描目标路径下生成的part开头的CSV文件，重命名为你需要的固定文件名（例如customer_data.csv）

修改后完整代码 #

import sys
import boto3
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

s3_bucket = "auroratos3dataimport"
s3_prefix = "customerdata2/"
target_file_name = "customer_data.csv"

datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "test_db", table_name = "test_dash_data", transformation_ctx = "datasource0")

applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("id", "int", "id", "int"), ("value", "int", "value", "int"), ("email", "string", "email", "string"), ("age", "int", "age", "int")], transformation_ctx = "applymapping1")

repartitioned1 = applymapping1.repartition(1)
# 写入时添加覆盖模式
datasink2 = glueContext.write_dynamic_frame.from_options(
    frame = repartitioned1, 
    connection_type = "s3", 
    connection_options = {
        "path": f"s3://{s3_bucket}/{s3_prefix}",
        "mode": "overwrite"
    }, 
    format = "csv", 
    transformation_ctx = "datasink2"
)
job.commit()

# 写入完成后重命名文件
s3 = boto3.client('s3')
# 列出前缀下的所有对象
response = s3.list_objects_v2(Bucket=s3_bucket, Prefix=s3_prefix)
for obj in response['Contents']:
    file_key = obj['Key']
    # 匹配Spark生成的part开头的CSV文件
    if file_key.startswith(f"{s3_prefix}part-") and file_key.endswith(".csv"):
        # 复制文件到目标文件名
        s3.copy_object(
            Bucket=s3_bucket,
            CopySource={'Bucket': s3_bucket, 'Key': file_key},
            Key=f"{s3_prefix}{target_file_name}"
        )
        # 删除原生成的part文件
        s3.delete_object(Bucket=s3_bucket, Key=file_key)
        break

注意事项 #

• 你已开启S3版本控制，上述重命名覆盖操作会自动生成新版本的customer_data.csv，历史版本会按你的S3版本控制规则保留，符合你的使用需求
• repartition(1)仅适合中小数据量场景，如果你的数据量超过单节点处理能力，需要评估性能风险
• 如果需要输出CSV带表头，可以在写入参数中添加format_options = {"header": True}

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