AWS Elastic Transcoder批量转码：为何需为每个文件创建任务？

阿华AIGC实验室

2026-5-14

解决AWS Elastic Transcoder批量自动转码的问题

嗨，我完全懂你的困惑——一开始我也以为管道能自动监听文件并批量处理，但其实Elastic Transcoder的「任务（Job）」本身就是针对单个文件设计的，要实现你想要的「上传后自动转码」，得搭配其他AWS服务来触发任务创建，而不是只靠Transcoder本身。

下面是具体的解决方案，分自动触发和批量处理已有文件两种场景：

一、实现「上传即自动转码」的正确姿势：S3 + Lambda

这是最常用的自动化方案，步骤如下：

给输入S3桶配置事件通知
- 进入你的输入S3桶的「属性」→「事件通知」，创建一个新的通知：
  - 触发条件选「对象创建（所有）」或者更精准的「对象创建（PUT）」
  - 目标选择「Lambda函数」，关联一个你要创建的Lambda函数

编写Lambda函数自动创建转码任务
Lambda的作用就是当有新文件上传时，自动调用Elastic Transcoder的API创建对应任务。这里给你一个Python的示例代码：

import boto3
from botocore.exceptions import ClientError

def lambda_handler(event, context):
    # 配置你的参数
    TRANSCODER_PIPELINE_ID = '你的管道ID'
    TRANSCODER_PRESET_ID = '你的转码预设ID（比如系统预设的1080p、720p）'
    OUTPUT_BUCKET = '你的输出桶名称'

    # 获取上传的文件Key
    s3_event = event['Records'][0]['s3']
    input_key = s3_event['object']['key']
    # 生成输出文件名（比如去掉原扩展名加.mp4）
    output_key = f"{input_key.rsplit('.', 1)[0]}_transcoded.mp4"

    transcoder = boto3.client('elastictranscoder')
    try:
        response = transcoder.create_job(
            PipelineId=TRANSCODER_PIPELINE_ID,
            Input={'Key': input_key},
            Outputs=[
                {
                    'Key': output_key,
                    'PresetId': TRANSCODER_PRESET_ID,
                    'Bucket': OUTPUT_BUCKET
                }
            ]
        )
        print(f"转码任务创建成功：{response['Job']['Id']}")
        return {'statusCode': 200, 'body': 'Job created successfully'}
    except ClientError as e:
        print(f"创建任务失败：{e.response['Error']['Message']}")
        return {'statusCode': 500, 'body': 'Failed to create transcoder job'}

记得给Lambda配置足够的IAM权限：允许它调用elastictranscoder:CreateJob，以及读取S3桶的对象信息。

二、批量处理输入桶中已有的文件

如果你的输入桶已经有一堆视频文件，不想一个个手动创建任务，可以写个简单的脚本批量调用API：

import boto3
import time

def batch_create_transcode_jobs():
    S3_INPUT_BUCKET = '你的输入桶名称'
    TRANSCODER_PIPELINE_ID = '你的管道ID'
    TRANSCODER_PRESET_ID = '你的预设ID'
    OUTPUT_BUCKET = '你的输出桶名称'

    s3 = boto3.client('s3')
    transcoder = boto3.client('elastictranscoder')

    # 遍历输入桶中的所有文件（处理分页）
    paginator = s3.get_paginator('list_objects_v2')
    for page in paginator.paginate(Bucket=S3_INPUT_BUCKET):
        for obj in page.get('Contents', []):
            input_key = obj['Key']
            # 过滤非视频文件（可选，根据你的需求调整后缀）
            if not input_key.lower().endswith(('.mp4', '.mov', '.avi', '.mkv')):
                continue
            output_key = f"{input_key.rsplit('.', 1)[0]}_transcoded.mp4"
            
            try:
                response = transcoder.create_job(
                    PipelineId=TRANSCODER_PIPELINE_ID,
                    Input={'Key': input_key},
                    Outputs=[{'Key': output_key, 'PresetId': TRANSCODER_PRESET_ID, 'Bucket': OUTPUT_BUCKET}]
                )
                print(f"已创建任务：{response['Job']['Id']} 对应文件：{input_key}")
                # 加个小延迟，避免触发API速率限制
                time.sleep(0.5)
            except Exception as e:
                print(f"处理文件 {input_key} 失败：{str(e)}")

if __name__ == '__main__':
    batch_create_transcode_jobs()