如何将S3中A文件夹的历史批量文件传入Lambda进行补处理？

核心思路 #

通过SQS消息队列做缓冲层，结合批量遍历Lambda与原有处理Lambda，实现可靠的历史文件补处理，同时覆盖各类潜在故障场景，避免重复处理、数据丢失等问题。

步骤1：构建SQS缓冲队列（含死信机制） #

• 创建标准SQS队列，用于传递待处理的S3文件信息
• 绑定死信队列（DLQ）：设置最大重试次数（建议3次），处理失败超过阈值的消息自动进入DLQ，避免无限重试消耗资源
• 将原有Lambda配置为该SQS队列的触发器：根据Lambda性能和日常业务负载，设置合理的并发数（比如10），避免批量处理影响日常定时任务

步骤2：编写批量遍历Lambda #

该Lambda的核心职责是把A文件夹下的所有历史文件信息批量推送到SQS队列：

• 核心逻辑：
  1. 分页调用S3的list_objects_v2接口，遍历s3://your-bucket/A/下的所有实际文件（过滤掉文件夹前缀）
  2. 对每个文件构造消息体：{"s3_bucket": "your-bucket", "s3_key": "A/2023-01-01-data.csv"}
  3. 批量（每次10条左右）发送消息到SQS队列
• 关键容错细节：
  • 用DynamoDB或S3标记文件记录遍历进度（比如存储S3分页的ContinuationToken和已推送的文件前缀），如果Lambda中断，重启后可从断点继续遍历，无需重复处理已推送文件

步骤3：改造原有Lambda（确保幂等与安全） #

原有Lambda需做以下适配，满足批量补处理的可靠性要求：

• 幂等校验：处理前先检查B文件夹是否存在同名文件，或用DynamoDB记录文件处理状态（未处理/处理中/成功/失败），仅当状态为「未处理」时才启动处理流程
• 原子性操作：先将文件复制到B文件夹，确认复制成功后再删除A文件夹的原文件，避免复制失败却丢失原数据
• 异常处理：捕获处理过程中的所有异常，写入CloudWatch日志后主动抛出异常，触发SQS的重试机制

步骤4：监控与收尾 #

• 通过CloudWatch监控SQS队列的消息数、死信队列的消息数，及时发现处理失败的文件
• 处理完所有队列消息后，排查DLQ中的失败消息（比如文件损坏、权限不足等），修复后重新推送至SQS处理
• 确认A文件夹所有文件都已迁移至B文件夹后，可停止批量遍历Lambda，或保留SQS队列用于后续可能的补处理需求

关键故障场景应对 #

• 批量遍历Lambda中断：通过进度记录实现断点续跑，无需重新遍历全部文件
• 原有Lambda处理失败：SQS自动重试，失败消息进入DLQ，不影响其他文件处理，可单独排查修复
• 重复处理风险：幂等校验机制确保同一文件不会被多次处理，即使SQS重试或批量Lambda重复推送
• 并发资源冲突：SQS触发器的并发数限制，避免批量处理抢占日常定时任务的资源
• 数据丢失风险：先复制后删除的操作顺序，确保只有复制成功才删除原文件

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