如何让Amazon AppFlow覆盖SalesFlow运行生成的历史输出文件？

一、使用AppFlow自身配置解决（优先推荐） #

你遇到的Execution ID自动分区问题，可通过修改目标存储的文件命名规则直接解决，无需依赖额外工具：

• 在AppFlow流的目标配置（如S3存储）中，找到「文件命名模式」设置，将默认的包含Execution ID的动态模式替换为固定文件名，例如salesforce_daily_latest.csv。
• 每次流运行时，新生成的CSV文件会直接覆盖桶中同名的旧文件，Lambda和QuickSight只需固定引用这个文件名，就能始终获取最新数据。
• 注意事项：
  • 确保AppFlow拥有目标S3桶的写入/覆盖权限。
  • 如果S3桶开启了版本控制，旧文件版本会被保留，但Lambda和QuickSight默认读取最新版本，不影响使用；若需彻底清除旧版本，可关闭版本控制或配置生命周期规则自动清理。

二、使用Lambda函数实现（灵活扩展） #

如果AppFlow的固定命名模式无法满足特殊需求，可通过Lambda函数自动清理旧文件：

实现逻辑 #

1. 配置Lambda触发条件：将S3桶的PutObject事件作为触发源，当AppFlow上传新CSV文件时自动执行Lambda。
2. Lambda核心操作：
   • 识别最新上传的文件。
   • 删除桶内所有历史导出文件（或复制最新文件为固定名称后删除其他文件）。

Python代码示例 #

import boto3

s3_client = boto3.client('s3')
TARGET_BUCKET = 'your-target-bucket'
FIXED_LATEST_FILENAME = 'salesforce_latest_data.csv'

def lambda_handler(event, context):
    # 获取新上传的文件路径
    new_file_key = event['Records'][0]['s3']['object']['key']
    
    # 将新文件复制为固定名称的文件（覆盖旧的固定文件）
    s3_client.copy_object(
        Bucket=TARGET_BUCKET,
        Key=FIXED_LATEST_FILENAME,
        CopySource=f"{TARGET_BUCKET}/{new_file_key}"
    )
    
    # 列出所有历史导出文件（假设文件名前缀为'salesforce_export_'）
    response = s3_client.list_objects_v2(Bucket=TARGET_BUCKET, Prefix='salesforce_export_')
    if 'Contents' in response:
        for obj in response['Contents']:
            # 删除除固定文件名和新上传文件外的所有旧文件
            if obj['Key'] not in [FIXED_LATEST_FILENAME, new_file_key]:
                s3_client.delete_object(Bucket=TARGET_BUCKET, Key=obj['Key'])
    
    # 可选：删除原始的带Execution ID的文件
    s3_client.delete_object(Bucket=TARGET_BUCKET, Key=new_file_key)
    
    return {'statusCode': 200, 'message': 'Latest file updated and old files cleaned up'}

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