请求技术指导：用Python编写AWS Lambda实现S3文件复制与处理

核心依赖库 #

• boto3: AWS官方Python SDK，用于操作S3等AWS服务，Lambda运行环境默认预装，无需额外打包
• io: 用于在内存中处理文件内容（避免使用Lambda临时目录，提升效率）
• os: 读取环境变量（推荐用环境变量存储桶名，避免硬编码）

关键便捷函数 #

• boto3.client('s3').get_object(): 从S3读取文件内容
• boto3.client('s3').copy_object(): 直接在S3之间复制文件（无需下载到本地，性能更优）
• boto3.client('s3').put_object(): 将处理后的内容上传到S3
• os.getenv(): 读取Lambda配置的环境变量（比如桶B、桶C的名称）

前置配置 #

1. IAM权限: 给Lambda角色添加以下S3权限：
   • s3:GetObject（桶A）
   • s3:PutObject（桶B、桶C）
   • s3:CopyObject（桶A到桶B）
2. S3触发器: 给桶A配置"创建对象"事件，触发目标Lambda函数
3. 环境变量: 在Lambda控制台配置BUCKET_B和BUCKET_C两个环境变量，分别对应目标桶名

完整示例代码 #

import boto3
import os
from io import StringIO, BytesIO

s3_client = boto3.client('s3')

def lambda_handler(event, context):
    # 1. 解析S3触发事件，获取源桶和文件名
    record = event['Records'][0]['s3']
    source_bucket = record['bucket']['name']
    source_key = record['object']['key']
    
    # 2. 根据文件名/内容判断执行逻辑
    # 示例1：按文件名后缀判断（比如只处理.csv文件）
    if not source_key.endswith('.csv'):
        print(f"跳过非CSV文件: {source_key}")
        return
    
    # 示例2：读取文件内容判断（比如检查是否包含特定关键字）
    try:
        response = s3_client.get_object(Bucket=source_bucket, Key=source_key)
        content = response['Body'].read().decode('utf-8')
        if 'target_keyword' not in content:
            print(f"文件不包含目标关键字，跳过处理: {source_key}")
            # 即使不处理，也可以选择复制到桶B，这里根据需求调整
            # s3_client.copy_object(...)
            return
    except Exception as e:
        print(f"读取文件内容失败: {str(e)}")
        raise e
    
    # 3. 将文件复制到桶B
    try:
        copy_source = {'Bucket': source_bucket, 'Key': source_key}
        s3_client.copy_object(
            Bucket=os.getenv('BUCKET_B'),
            Key=source_key,
            CopySource=copy_source
        )
        print(f"文件已复制到桶B: {source_key}")
    except Exception as e:
        print(f"复制文件到桶B失败: {str(e)}")
        raise e
    
    # 4. 处理文件内容，上传到桶C
    try:
        # 示例处理逻辑：给CSV文件添加表头（根据你的实际需求修改）
        processed_content = "新增列1,新增列2," + content if content.startswith('列1') else content
        
        # 处理二进制文件（比如图片）的话，用BytesIO
        # processed_content = some_binary_processing_function(content_bytes)
        
        # 上传到桶C，这里可以修改输出文件名，比如加前缀
        target_key = f"processed/{source_key}"
        s3_client.put_object(
            Bucket=os.getenv('BUCKET_C'),
            Key=target_key,
            Body=processed_content.encode('utf-8')
        )
        print(f"处理后的文件已上传到桶C: {target_key}")
    except Exception as e:
        print(f"处理并上传文件到桶C失败: {str(e)}")
        raise e
    
    return {
        'statusCode': 200,
        'body': f"处理完成: {source_key}"
    }

注意事项 #

• 避免硬编码: 桶名、关键字等配置尽量用环境变量，方便后续修改
• 内存与超时: 根据文件大小调整Lambda的内存配置（内存越高，CPU性能越好），并设置足够的超时时间
• 错误处理: 增加必要的try-except块，避免单个文件处理失败导致整个Lambda报错
• 大文件处理: 如果处理GB级文件，建议使用S3 Select或分段处理，避免内存溢出

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