如何创建S3上传触发的AWS Lambda函数 截取视频前30秒回存S3

核心链路 #

整个功能的执行逻辑完全在AWS托管服务内闭环，无外部依赖：
S3存储桶接收新视频上传 -> 触发Lambda执行 -> 下载源视频到Lambda本地临时目录 -> 调用内置FFmpeg截取前30秒片段 -> 将处理后的视频上传到目标存储路径 -> 自动清理本地临时文件 #

前置环境准备 #

• Lambda运行时选择
  选用Python 3.10作为运行时，必须额外配置自定义Lambda层：将静态编译的FFmpeg二进制文件打包到层内，路径对应到代码中调用的/opt/ffmpeg；打包时尽量裁剪掉不需要的编码器、格式解析功能，把层体积控制在50MB以内，减少函数冷启动耗时，不要依赖Lambda官方基础镜像内置的工具（官方镜像默认无FFmpeg环境）
• 权限配置
  为Lambda执行角色配置最小权限集，禁止开放全局S3访问权限：
  • 对源视频存储桶授予s3:GetObject权限，限定资源为桶内raw-uploads/*路径下的对象
  • 对输出结果存储桶授予s3:PutObject权限，限定资源为桶内processed/*路径下的对象
• 基础参数调整
  • 将Lambda临时磁盘容量调整为至少10GB（默认仅512MB，无法容纳大体积视频文件）
  • 将Lambda超时时间调整为60秒起步，如涉及重编码场景可延长至最长15分钟的系统上限
  • Lambda内存配置建议从1GB起步，更高内存对应更高的CPU算力，可大幅提升FFmpeg处理速度
• S3触发规则配置
  为源存储桶配置对象创建事件通知，关联目标Lambda函数；必须添加路径与后缀过滤规则：仅监听raw-uploads/路径下、后缀为.mp4/.mov/.mkv/.avi的文件变更，从根源上避免处理结果文件被重复识别为新上传对象、触发无限循环执行
• 环境变量配置
  在Lambda控制台配置环境变量TARGET_BUCKET，值为存储处理后视频的目标桶名称，避免将桶名硬编码在代码中

核心实现代码 #

import boto3
import os
import subprocess
import uuid

s3_client = boto3.client('s3')
TARGET_BUCKET = os.environ['TARGET_BUCKET']
FFMPEG_BIN_PATH = '/opt/ffmpeg'

def lambda_handler(event, context):
    # 解析S3触发事件中的源文件元数据
    event_record = event['Records'][0]
    source_bucket_name = event_record['s3']['bucket']['name']
    source_object_key = event_record['s3']['object']['key']

    # 生成带唯一标识的临时文件路径，避免并发执行时的文件冲突
    source_file_temp = f'/tmp/source_{uuid.uuid4().hex}.mp4'
    output_file_temp = f'/tmp/output_{uuid.uuid4().hex}.mp4'

    try:
        # 下载源视频到本地临时目录
        s3_client.download_file(source_bucket_name, source_object_key, source_file_temp)

        # 调用FFmpeg截取前30秒，直接复用原始音视频流避免重编码损耗
        exec_cmd = [
            FFMPEG_BIN_PATH,
            '-i', source_file_temp,
            '-t', '30',
            '-c', 'copy',
            '-y',
            output_file_temp
        ]
        subprocess.run(exec_cmd, check=True, capture_output=True, text=True)

        # 生成输出文件的存储路径，保留原文件的文件名
        output_object_key = f'processed/{source_object_key.split("/")[-1]}'
        s3_client.upload_file(output_file_temp, TARGET_BUCKET, output_object_key)

        return f"处理完成，输出路径：s3://{TARGET_BUCKET}/{output_object_key}"

    except Exception as e:
        raise RuntimeError(f"文件处理失败，源文件路径 s3://{source_bucket_name}/{source_object_key}，错误信息：{str(e)}")

    finally:
        # 无论执行成功或失败，强制清理临时文件，避免残留占满磁盘
        if os.path.exists(source_file_temp):
            os.remove(source_file_temp)
        if os.path.exists(output_file_temp):
            os.remove(output_file_temp)

常见踩坑点说明 #

• 循环触发问题：这是90%的人第一次实现这个功能都会踩的坑——如果不做S3事件的路径过滤，处理完的视频上传到同一个被监听的路径下时，会被再次识别为新上传文件，反复触发Lambda执行，平白产生大量费用
• 处理速度问题：采用-c copy参数的流复制模式时，不需要对视频重新编码，处理速度可达数十倍实时速度，100MB以内的视频仅需几秒即可完成；如果必须做转码适配，需要同步调高Lambda内存配置，否则会频繁触发超时
• 临时文件冲突：多文件同时触发Lambda时，如果所有函数实例都用固定的临时文件名，会出现不同执行实例读写同一文件的冲突，因此代码中必须给临时文件加唯一标识
• 权限最小化：不要给Lambda配置s3:*的全局权限，所有权限限定到需要的桶和对应路径，避免出现数据泄露风险

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