如何配置AWS Lambda处理S3中带可变前缀的目标文件

一、S3触发规则的通配符配置 #

你完全可以通过通配符逻辑配置S3触发规则，不需要在测试事件里写通配符。在Lambda的S3触发器配置中：

• 设置前缀为mcu/，后缀为marvel_report.csv：只要marvel-analysis桶里，任何mcu/下子路径中的marvel_report.csv被上传，都会触发Lambda。
• 如果要更精准匹配（比如只针对dx-开头的中间前缀），可以把前缀设为mcu/dx-，后缀保持marvel_report.csv，这样只有mcu/dx-xxx/marvel_report.csv这类路径的文件才会触发函数，避免无关文件干扰。

二、测试事件的灵活处理 #

测试事件里不能用通配符，但可以通过两种方式适配可变前缀：

1. 创建多个测试事件模板，分别对应不同的前缀场景（比如dx-hero-230211、dx-villain-230223），测试时切换使用。
2. 修改测试事件的key值快速替换前缀，示例如下：

{
  "Records": [
    {
      "eventVersion": "2.0",
      "eventSource": "aws:s3",
      "awsRegion": "us-east-1", // 替换为你的函数所在区域，不要用Global
      "eventTime": "2024-05-20T00:00:00.000Z",
      "eventName": "ObjectCreated:Put",
      "userIdentity": {
        "principalId": "EXAMPLE"
      },
      "requestParameters": {
        "sourceIPAddress": "127.0.0.1"
      },
      "responseElements": {
        "x-amz-request-id": "EXAMPLE123456789",
        "x-amz-id-2": "EXAMPLE123/5678abcdefghijklambdaisawesome/mnopqrstuvwxyzABCDEFGH"
      },
      "s3": {
        "s3SchemaVersion": "1.0",
        "configurationId": "testConfigRule",
        "bucket": {
          "name": "marvel-analysis",
          "ownerIdentity": {
            "principalId": "EXAMPLE"
          },
          "arn": "arn:aws:s3:::marvel-analysis"
        },
        "object": {
          "key": "mcu/dx-villain-230223/marvel_report.csv", // 替换为需要测试的前缀
          "size": 1024,
          "eTag": "0123456789abcdef0123456789abcdef",
          "sequencer": "0A1B2C3D4E5F678901"
        }
      }
    }
  ]
}

三、代码优化：替换硬编码的前缀提取 #

你当前用split("/")[1]提取前缀的方法不够稳健，若路径结构变化（比如新增目录层级）会直接出错，推荐两种更可靠的方式：

方法1：按固定文件名反向截取 #

利用目标文件名为marvel_report.csv的固定特征，反向提取前缀：

import json
import urllib.parse
import boto3
import os

print('Loading function')

s3 = boto3.client('s3')

def lambda_handler(event, context):
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
    
    try:
        target_file = "marvel_report.csv"
        if key.endswith(target_file):
            # 去掉文件名后，截取最后一层目录作为run_id
            prefix_part = key[:-len(target_file)].rstrip("/")
            run_id = prefix_part.split("/")[-1]
            print(f"Extracted run_id: {run_id}")
        else:
            print("Skipping non-target file")
    except Exception as e:
        print(e)
        print(f'Error processing object {key} from bucket {bucket}')
        raise e

方法2：正则表达式精准匹配 #

如果前缀有固定格式（比如dx-xxx-6位日期），用正则匹配更严谨：

import json
import urllib.parse
import boto3
import re

print('Loading function')

s3 = boto3.client('s3')

def lambda_handler(event, context):
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
    
    try:
        # 匹配mcu/[dx-xxx-YYYYMMDD]/marvel_report.csv格式
        pattern = r"mcu/(dx-[a-z]+-\d{6})/marvel_report.csv"
        match = re.fullmatch(pattern, key)
        if match:
            run_id = match.group(1)
            print(f"Extracted run_id: {run_id}")
        else:
            print("Path does not match expected pattern")
    except Exception as e:
        print(e)
        print(f'Error processing object {key} from bucket {bucket}')
        raise e

四、新手学习建议 #

• 先研读AWS Lambda S3触发器官方文档：掌握触发规则的前缀/后缀过滤逻辑，以及事件结构细节
• 学习S3对象键命名规范：了解路径分隔符、特殊字符处理等注意事项
• 多练习Lambda事件测试：创建不同场景的测试事件，熟悉事件字段含义
• 优化错误处理逻辑：添加对非目标文件的判断，避免函数因无关文件触发报错

内容的提问来源于stack exchange，提问作者R.T. Canterbury