如何用Python编写AWS S3 Lambda 复制指定最新文件到同桶其他目录?
实现方案
一、Lambda Python 函数实现
1. 前置依赖配置
- Lambda执行角色需要添加如下IAM权限:
- 针对目标S3桶的
s3:ListBucket权限 - 针对源备份路径的
s3:GetObject权限 - 针对目标存放路径的
s3:PutObject权限
- 针对目标S3桶的
- Lambda运行时选择Python 3.9及以上版本即可,无需额外层依赖,boto3为Lambda内置库
2. 函数代码
import boto3 # 配置参数,可根据实际情况修改 S3_BUCKET_NAME = "你的存储桶名称" SOURCE_PREFIX = "源备份子文件夹路径/" # 示例:backup/prod/ TARGET_PREFIX = "目标子文件夹路径/" # 示例:latest_backup/ MATCH_KEYWORD = "Database5" TARGET_FILENAME = "Database5_backup.bak" s3 = boto3.client('s3') def lambda_handler(event, context): # 列出源路径下的所有对象 response = s3.list_objects_v2( Bucket=S3_BUCKET_NAME, Prefix=SOURCE_PREFIX ) if "Contents" not in response: print("源路径下无匹配文件") return # 筛选文件名包含指定关键字的备份文件 matched_files = [ obj for obj in response["Contents"] if MATCH_KEYWORD in obj["Key"] and obj["Key"].endswith(".bak") ] if not matched_files: print(f"未找到包含{MATCH_KEYWORD}的备份文件") return # 按文件最后修改时间倒序排序,取最新的一个 # 若需按文件名中的时间戳排序,可自行修改sorted的key规则 latest_file = sorted(matched_files, key=lambda x: x["LastModified"], reverse=True)[0] source_key = latest_file["Key"] target_key = f"{TARGET_PREFIX.rstrip('/')}/{TARGET_FILENAME}" # 同桶内直接调用复制接口,无需下载后再上传,效率更高 s3.copy_object( Bucket=S3_BUCKET_NAME, Key=target_key, CopySource={ "Bucket": S3_BUCKET_NAME, "Key": source_key } ) print(f"已成功复制最新备份文件:{source_key} 到 {target_key}") return
3. 触发方式推荐
- 定时触发:配置EventBridge定时规则,每天备份任务完成后的固定时间执行即可
- 事件触发:配置S3桶的对象创建事件通知,仅当源路径下有新的
.bak文件上传时触发函数,实时更新最新备份
二、S3 Replication 适配性分析
S3 Replication不适用于当前场景,原因如下:
- S3 Replication仅支持固定前缀/后缀的规则匹配,无法自动筛选出最新的单个文件
- S3 Replication不支持复制过程中动态修改文件名、去除时间戳的自定义逻辑
- 同桶复制属于S3 Replication的额外限制场景,需要额外配置且成本高于Lambda方案
内容的提问来源于stack exchange,提问作者J. Outlaw
相关产品推荐
相关产品推荐

