python-3.x - 对 S3 中的多个文件执行 lambda 函数

Question

我试图找出在 S3 中一次处理多个文件的逻辑，因为文件是随机添加的。为了便于讨论，这里有一个例子:

1. 文件随机添加到S3 bucket；通过突发或随机间隔
2. 一旦 S3 存储桶中有 9 个文件，就会触发 Lambda 函数； lambda 函数后处理这些文件或将这些文件组合在一起。
3. 处理后，文件将被移动到另一个存储桶或被删除。

这是我尝试过的:

• 我有适用于所有 S3 放置的 S3 触发器
• 在我的 lambda 函数中，我忽略文件名本身并根据键列出 S3 存储桶以计算存在的文件数
• 问题是当流量突发或稳定到达但速度很快时，很难识别 9 个文件的独特组
• 出于性能原因，我在文件名上使用了 uuid 前缀，因此不存在顺序文件名。
• 我考虑过将元数据写入 nosql 数据库，但还没有走这条路。

Answer 1

一种可能的解决方案是使用计划的 lambda(根据您的流量可以尽可能频繁或尽可能稀疏)从由 S3 put 事件填充的 SQS 队列中提取事件。假定您专注于一次批处理 n 文件，并且顺序无关紧要(给定 uuid 命名)。

要创建此工作流，将如下所示:

1. 创建用于保存 S3 PUT 事件的 SQS 队列
2. 将触发器添加到 PUT 上的 S3 存储桶，以从 1 在 SQS 队列中创建事件。
3. 使用环境变量创建 Lambda(用于存储桶和队列)
   1. lambda 应该检查队列是否有任何正在处理的消息并只使用桶
   2. 如果有，停止运行(防止一个文件被多次处理)
   3. 如果没有动态消息，则列出来自 S3 的对象，限制为 n(您的批量大小)
   4. 如果返回了足够多的对象(可能小于 n)，则运行您的流程逻辑
   5. 删除文件
4. 创建每 n 秒/分钟/小时运行 lambda 的 CloudWatch 规则

根据您的具体情况，需要记住的其他一些事项:

• 如果有很多文件正在快速发送并且 n 非常小，则单跟踪处理(步骤 3.2 会导致处理时间过长)。这还要看处理时间的长短，数据是否可以多次处理等等……
• ListObjectsV2 可以返回小于 MaxKeys 参数，如果这是一个问题，可以有一个更大的 MaxKeys 并且只处理第一个 n.