如何配置Lambda函数使其仅在两个S3前缀均完成文件上传后触发

该方案轻量无冗余组件，运维成本、使用成本极低，适配绝大多数同场景聚合运算需求，核心逻辑为引入轻量状态存储记录两份文件的上传状态，校验通过后再执行聚合逻辑，完整实现步骤如下：

1. 配置事件触发链路 #

• 给两个目标S3前缀分别配置事件通知规则，触发条件选择s3:ObjectCreated:*类事件（可按需指定为Put/Post/多部分上传完成事件，过滤临时分片文件触发），事件目标可直接绑定到同一个处理Lambda，省略不必要的中转链路。
• 如果你需要保留SNS做消息广播的需求，可将S3事件通知的目标先设置为SNS主题，再给SNS主题配置Lambda订阅，最终效果一致。

2. 配置状态存储表 #

推荐使用DynamoDB做状态存储，单表即可支撑，表结构设计如下：

• 主键设置为两份文件共有的业务关联标识（比如同批次数据的批次ID、同日期数据的日期标识，无特殊业务标识的场景可直接用固定的聚合任务ID作为主键）
• 字段分别设置file_a_uploaded（布尔型）、file_b_uploaded（布尔型），可选新增两个文件的S3路径字段存储方便后续聚合调用。

3. 编写Lambda核心逻辑 #

Lambda每次被触发后的执行流程如下，附核心伪代码示例：

import boto3
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('替换为你的状态表名')

def lambda_handler(event, context):
    # 解析S3事件获取上传文件信息
    s3_info = event['Records'][0]['s3']
    file_key = s3_info['object']['key']
    bucket_name = s3_info['bucket']['name']
    # 提取两份文件共有的业务关联ID，可根据你的文件命名规则自行实现
    biz_id = extract_biz_id(file_key)

    # 原子更新对应文件的上传状态
    update_exp = None
    if file_key.startswith('替换为前缀A的路径/'):
        update_exp = 'SET file_a_uploaded = :val, file_a_path = :path'
    elif file_key.startswith('替换为前缀B的路径/'):
        update_exp = 'SET file_b_uploaded = :val, file_b_path = :path'
    else:
        # 非目标前缀的文件直接终止执行
        return

    # 更新状态同时返回更新后的最新数据
    response = table.update_item(
        Key={'biz_id': biz_id},
        UpdateExpression=update_exp,
        ExpressionAttributeValues={
            ':val': True,
            ':path': f"s3://{bucket_name}/{file_key}"
        },
        ReturnValues='ALL_NEW'
    )
    latest_state = response['Attributes']

    # 校验两份文件是否都已上传完成
    if latest_state.get('file_a_uploaded') == True and latest_state.get('file_b_uploaded') == True:
        # 执行你的聚合运算逻辑
        run_aggregation_logic(latest_state['file_a_path'], latest_state['file_b_path'])
        # 任务完成后可选删除对应状态记录，避免重复执行
        table.delete_item(Key={'biz_id': biz_id})

4. 补充容错与幂等配置 #

• 给DynamoDB表开启TTL字段，设置状态记录24小时后自动过期，避免异常场景下垃圾数据堆积
• 给Lambda配置死信队列（DLQ），处理偶发的事件消费失败场景
• 若需要严格避免重复执行，可在状态表中新增processed字段，执行聚合前先校验该字段状态，聚合完成后标记为已处理

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