ffmpeg-modal-containers

快速参考

容器类型	镜像设置	GPU	使用场景
CPU（debian_slim）	.apt_install("ffmpeg")	否	批处理、I/O 密集型任务
GPU（debian_slim）	.apt_install("ffmpeg").pip_install("torch")	是	ML 推理，不支持 NVENC
GPU（CUDA 镜像）	from_registry("nvidia/cuda:...")	是	完整 CUDA 工具包，可能支持 NVENC

GPU 类型	每小时价格	NVENC	最适用于
T4	~$0.59	是（Turing）	推理 + 编码
A10G	~$1.10	是（Ampere）	4K 编码、ML
L40S	~$1.95	是（Ada）	重度 ML + 视频
H100	~$4.25	是（Hopper）	训练；用于视频处理性能过剩

何时使用此技能

适用于无服务器视频处理：

• 需要扩展到数百个容器的批量转码
• 使用 Modal 的 map/starmap 进行并行视频处理
• GPU 加速编码（但 NVENC 有限制）
• 经济高效的突发处理（仅按执行时间付费）
• 与 ML 模型集成（Whisper、视频分析）

关键决策：Modal 擅长并行 CPU 工作负载和 GPU 上的 ML 推理。若要纯粹使用硬件 NVENC 编码，请先确认 GPU 功能。

---

Modal.com 上的 FFmpeg（2025）

关于在 Modal 的无服务器 Python 平台上使用 CPU 和 GPU 容器运行 FFmpeg 的完整指南。

概述

Modal 是一个无服务器平台，可在云端运行 Python 代码，具备以下特性：

• 冷启动不到一秒 - 容器可在几毫秒内启动
• 弹性 GPU 容量 - 可使用 T4、A10G、L40S、H100 GPU
• 并行处理 - 可即时扩展至数千个容器
• 按使用量付费 - 按 CPU 周期计费，而非空闲时间

Modal 与传统云服务对比

特性	Modal	传统 VM
冷启动	<1 秒	数分钟
扩展	自动扩展至数千个	手动设置
计费	按执行次数	按小时
GPU 访问	gpu="any" 装饰器	配置复杂
设置	Python 装饰器	基础设施即代码

FFmpeg 基本设置

CPU 容器（最简单）

import modal
import subprocess
from pathlib import Path

app = modal.App("ffmpeg-processor")

# Create image with FFmpeg installed
ffmpeg_image = modal.Image.debian_slim(python_version="3.12").apt_install("ffmpeg")

@app.function(image=ffmpeg_image)
def transcode_video(input_bytes: bytes, output_format: str = "mp4") -> bytes:
    """Transcode video to specified format."""
    import tempfile

    with tempfile.TemporaryDirectory() as tmpdir:
        input_path = Path(tmpdir) / "input"
        output_path = Path(tmpdir) / f"output.{output_format}"

        # Write input file
        input_path.write_bytes(input_bytes)

        # Run FFmpeg
        result = subprocess.run([
            "ffmpeg", "-y",
            "-i", str(input_path),
            "-c:v", "libx264",
            "-preset", "veryfast",
            "-crf", "23",
            "-c:a", "aac",
            "-b:a", "128k",
            "-movflags", "+faststart",
            str(output_path)
        ], capture_output=True, text=True)

        if result.returncode != 0:
            raise RuntimeError(f"FFmpeg error: {result.stderr}")

        return output_path.read_bytes()

@app.local_entrypoint()
def main():
    # Read local file
    video_bytes = Path("input.mp4").read_bytes()

    # Process remotely on Modal
    output_bytes = transcode_video.remote(video_bytes)

    # Save result locally
    Path("output.mp4").write_bytes(output_bytes)
    print("Transcoding complete!")

运行你的第一个 Modal 应用

# Install Modal
pip install modal

# Authenticate (one-time)
modal setup

# Run the app
modal run your_script.py

GPU 容器

ML + FFmpeg 的基本 GPU 设置

import modal

app = modal.App("ffmpeg-gpu")

# GPU image with FFmpeg and PyTorch
gpu_image = (
    modal.Image.debian_slim(python_version="3.12")
    .apt_install("ffmpeg")
    .pip_install("torch", "torchaudio", "transformers")
)

@app.function(image=gpu_image, gpu="T4")
def transcribe_and_process(audio_bytes: bytes) -> dict:
    """Transcribe audio with Whisper, then process with FFmpeg."""
    import tempfile
    import torch
    from transformers import pipeline

    with tempfile.TemporaryDirectory() as tmpdir:
        input_path = Path(tmpdir) / "input.mp3"
        input_path.write_bytes(audio_bytes)

        # GPU-accelerated transcription
        transcriber = pipeline(
            model="openai/whisper-base",
            device="cuda"
        )
        result = transcriber(str(input_path))

        # FFmpeg audio normalization (CPU-based in this setup)
        normalized_path = Path(tmpdir) / "normalized.mp3"
        subprocess.run([
            "ffmpeg", "-y",
            "-i", str(input_path),
            "-af", "loudnorm=I=-16:TP=-1.5:LRA=11",
            str(normalized_path)
        ], check=True)

        return {
            "transcription": result["text"],
            "normalized_audio": normalized_path.read_bytes()
        }

高级 GPU 功能所需的完整 CUDA 工具包

如需使用 NVENC 或完整的 CUDA 工具包：

import modal

cuda_version = "12.4.0"
flavor = "devel"  # Full toolkit
os_version = "ubuntu22.04"
tag = f"{cuda_version}-{flavor}-{os_version}"

# Full CUDA image with FFmpeg
cuda_ffmpeg_image = (
    modal.Image.from_registry(f"nvidia/cuda:{tag}", add_python="3.12")
    .entrypoint([])  # Remove base image entrypoint
    .apt_install(
        "ffmpeg",
        "git",
        "libglib2.0-0",
        "libsm6",
        "libxrender1",
        "libxext6",
        "libgl1",
    )
    .pip_install("numpy", "Pillow")
)

app = modal.App("ffmpeg-cuda")

@app.function(image=cuda_ffmpeg_image, gpu="A10G")
def gpu_transcode(input_bytes: bytes) -> bytes:
    """Transcode video with GPU acceleration if available."""
    import subprocess
    import tempfile
    from pathlib import Path

    with tempfile.TemporaryDirectory() as tmpdir:
        input_path = Path(tmpdir) / "input.mp4"
        output_path = Path(tmpdir) / "output.mp4"

        input_path.write_bytes(input_bytes)

        # Check for NVENC support
        check_result = subprocess.run(
            ["ffmpeg", "-encoders"],
            capture_output=True,
            text=True
        )

        has_nvenc = "h264_nvenc" in check_result.stdout

        if has_nvenc:
            # GPU encoding with NVENC
            cmd = [
                "ffmpeg", "-y",
                "-hwaccel", "cuda",
                "-hwaccel_output_format", "cuda",
                "-i", str(input_path),
                "-c:v", "h264_nvenc",
                "-preset", "p4",
                "-cq", "23",
                "-c:a", "aac",
                "-b:a", "128k",
                str(output_path)
            ]
        else:
            # Fallback to CPU encoding
            cmd = [
                "ffmpeg", "-y",
                "-i", str(input_path),
                "-c:v", "libx264",
                "-preset", "veryfast",
                "-crf", "23",
                "-c:a", "aac",
                "-b:a", "128k",
                str(output_path)
            ]

        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            raise RuntimeError(f"FFmpeg error: {result.stderr}")

        return output_path.read_bytes()

关于 NVENC 支持的重要说明

Modal 的 GPU 容器主要使用 NVIDIA GPU 进行 ML 推理。NVENC 视频编码支持取决于：

1. FFmpeg 构建版本 - 必须包含 --enable-nvenc
2. NVIDIA 驱动程序 - 必须提供视频编码功能
3. 容器设置 - 可能需要 NVIDIA_DRIVER_CAPABILITIES=compute,video,utility

如需确保支持 NVENC，请使用自定义 Docker 镜像，或通过以下方式进行验证：

@app.function(image=cuda_ffmpeg_image, gpu="T4")
def check_nvenc():
    """Check NVENC availability."""
    import subprocess

    # Check GPU
    gpu_result = subprocess.run(["nvidia-smi"], capture_output=True, text=True)
    print("GPU Info:", gpu_result.stdout)

    # Check FFmpeg encoders
    enc_result = subprocess.run(
        ["ffmpeg", "-encoders"],
        capture_output=True,
        text=True
    )

    nvenc_encoders = [line for line in enc_result.stdout.split('\n') if 'nvenc' in line]
    print("NVENC Encoders:", nvenc_encoders)

    return {
        "has_nvenc": len(nvenc_encoders) > 0,
        "encoders": nvenc_encoders
    }

并行视频处理

Modal 在视频处理方面的杀手级特性，是可跨多个容器并行执行。

使用 map() 进行批处理

import modal
from pathlib import Path

app = modal.App("batch-transcode")

ffmpeg_image = modal.Image.debian_slim().apt_install("ffmpeg")

@app.function(image=ffmpeg_image, timeout=600)
def transcode_single(video_bytes: bytes, video_id: str) -> tuple[str, bytes]:
    """Transcode a single video."""
    import subprocess
    import tempfile

    with tempfile.TemporaryDirectory() as tmpdir:
        input_path = Path(tmpdir) / "input"
        output_path = Path(tmpdir) / "output.mp4"

        input_path.write_bytes(video_bytes)

        subprocess.run([
            "ffmpeg", "-y",
            "-i", str(input_path),
            "-c:v", "libx264",
            "-preset", "fast",
            "-crf", "23",
            "-c:a", "aac",
            str(output_path)
        ], check=True, capture_output=True)

        return video_id, output_path.read_bytes()

@app.local_entrypoint()
def main():
    # Prepare batch of videos
    video_files = list(Path("videos").glob("*.mp4"))
    inputs = [(f.read_bytes(), f.stem) for f in video_files]

    # Process all videos in parallel (up to 100 containers)
    results = list(transcode_single.starmap(inputs))

    # Save results
    for video_id, output_bytes in results:
        Path(f"output/{video_id}.mp4").write_bytes(output_bytes)
        print(f"Processed: {video_id}")

逐帧并行处理

如需最大程度的并行处理，可独立处理每一帧：

import modal
from pathlib import Path

app = modal.App("parallel-frames")

ffmpeg_image = modal.Image.debian_slim().apt_install("ffmpeg")

@app.function(image=ffmpeg_image)
def extract_frames(video_bytes: bytes, fps: int = 1) -> list[bytes]:
    """Extract frames from video."""
    import subprocess
    import tempfile

    with tempfile.TemporaryDirectory() as tmpdir:
        input_path = Path(tmpdir) / "input.mp4"
        input_path.write_bytes(video_bytes)

        # Extract frames
        subprocess.run([
            "ffmpeg", "-y",
            "-i", str(input_path),
            "-vf", f"fps={fps}",
            f"{tmpdir}/frame_%04d.png"
        ], check=True, capture_output=True)

        # Read all frames
        frames = []
        for frame_path in sorted(Path(tmpdir).glob("frame_*.png")):
            frames.append(frame_path.read_bytes())

        return frames

@app.function(image=ffmpeg_image)
def process_frame(frame_bytes: bytes, frame_id: int) -> bytes:
    """Process a single frame (add watermark, filter, etc.)."""
    import subprocess
    import tempfile

    with tempfile.TemporaryDirectory() as tmpdir:
        input_path = Path(tmpdir) / "frame.png"
        output_path = Path(tmpdir) / "processed.png"

        input_path.write_bytes(frame_bytes)

        # Apply processing (example: add text overlay)
        subprocess.run([
            "ffmpeg", "-y",
            "-i", str(input_path),
            "-vf", f"drawtext=text='Frame {frame_id}':fontsize=24:fontcolor=white:x=10:y=10",
            str(output_path)
        ], check=True, capture_output=True)

        return output_path.read_bytes()

@app.function(image=ffmpeg_image)
def combine_frames(frames: list[bytes], fps: int = 24) -> bytes:
    """Combine processed frames back into video."""
    import subprocess
    import tempfile

    with tempfile.TemporaryDirectory() as tmpdir:
        # Write frames
        for i, frame_bytes in enumerate(frames):
            frame_path = Path(tmpdir) / f"frame_{i:04d}.png"
            frame_path.write_bytes(frame_bytes)

        output_path = Path(tmpdir) / "output.mp4"

        subprocess.run([
            "ffmpeg", "-y",
            "-framerate", str(fps),
            "-i", f"{tmpdir}/frame_%04d.png",
            "-c:v", "libx264",
            "-pix_fmt", "yuv420p",
            str(output_path)
        ], check=True, capture_output=True)

        return output_path.read_bytes()

@app.local_entrypoint()
def main():
    video_bytes = Path("input.mp4").read_bytes()

    # Step 1: Extract frames (single container)
    frames = extract_frames.remote(video_bytes, fps=24)
    print(f"Extracted {len(frames)} frames")

    # Step 2: Process frames in parallel (many containers)
    args = [(frame, i) for i, frame in enumerate(frames)]
    processed_frames = list(process_frame.starmap(args))
    print(f"Processed {len(processed_frames)} frames")

    # Step 3: Combine frames (single container)
    output = combine_frames.remote(processed_frames, fps=24)

    Path("output.mp4").write_bytes(output)
    print("Video processing complete!")

用于大文件的 Modal Volumes

对于无法作为函数参数传递的过大视频文件，请使用 Modal Volumes：

设置和使用 Volume

import modal
from pathlib import Path

app = modal.App("video-volume")

# Create persistent volume for video storage
video_volume = modal.Volume.from_name("video-storage", create_if_missing=True)

ffmpeg_image = modal.Image.debian_slim().apt_install("ffmpeg")

@app.function(
    image=ffmpeg_image,
    volumes={"/data": video_volume},
    timeout=1800  # 30 minutes for large files
)
def transcode_from_volume(input_filename: str, output_filename: str):
    """Transcode video from volume to volume."""
    import subprocess

    input_path = Path("/data") / input_filename
    output_path = Path("/data") / output_filename

    if not input_path.exists():
        raise FileNotFoundError(f"Input file not found: {input_path}")

    subprocess.run([
        "ffmpeg", "-y",
        "-i", str(input_path),
        "-c:v", "libx264",
        "-preset", "medium",
        "-crf", "22",
        "-c:a", "aac",
        "-b:a", "192k",
        str(output_path)
    ], check=True, capture_output=True)

    # Commit changes to volume (important!)
    video_volume.commit()

    return f"Transcoded: {output_filename}"

@app.function(volumes={"/data": video_volume})
def list_videos():
    """List all videos in the volume."""
    videos = list(Path("/data").glob("*.mp4"))
    return [v.name for v in videos]

@app.local_entrypoint()
def main():
    # Upload a file to the volume first
    # modal volume put video-storage local_video.mp4 video.mp4

    # Then transcode
    result = transcode_from_volume.remote("video.mp4", "video_transcoded.mp4")
    print(result)

    # List files
    files = list_videos.remote()
    print("Files in volume:", files)

上传到 Volume

# Upload file to volume
modal volume put video-storage local_video.mp4 video.mp4

# Download file from volume
modal volume get video-storage video_transcoded.mp4 local_output.mp4

# List volume contents
modal volume ls video-storage

Volume 最佳实践

@app.function(
    volumes={"/data": video_volume},
    ephemeral_disk=50 * 1024  # 50 GB ephemeral disk for temp files
)
def process_large_video(input_filename: str):
    """Process large video with ephemeral disk for temp storage."""
    import subprocess
    import shutil

    # Copy from volume to ephemeral disk for faster I/O
    input_volume_path = Path("/data") / input_filename
    temp_input = Path("/tmp") / input_filename
    shutil.copy(input_volume_path, temp_input)

    temp_output = Path("/tmp") / "output.mp4"

    # Process on fast ephemeral disk
    subprocess.run([
        "ffmpeg", "-y",
        "-i", str(temp_input),
        "-c:v", "libx264",
        "-preset", "slow",  # Higher quality, more processing
        "-crf", "18",
        str(temp_output)
    ], check=True, capture_output=True)

    # Copy result back to volume
    output_volume_path = Path("/data") / f"processed_{input_filename}"
    shutil.copy(temp_output, output_volume_path)

    # Commit to persist
    video_volume.commit()

    return str(output_volume_path)

成本优化、生产环境模式、Whisper 音频与故障排查

成本控制技巧（竞价/可抢占容器、自动扩缩容保护机制、CPU 与 GPU 选择）、生产级编排（排队、重试、进度报告、监控）、基于 Whisper 的转录流水线以及故障排查目录，详见 references/production-audio-troubleshooting.md。在强化 Modal FFmpeg 部署、添加语音转文本功能或诊断作业失败时，请加载该参考资料。

最佳实践

1. 使用 CPU 进行转码 - 对大多数编码任务来说，GPU 性能过剩。
2. 使用 map/starmap 并行处理 - 同时处理多个文件。
3. 使用卷处理大文件 - 避免将大量数据作为参数传递。
4. 设置合适的超时时间 - 视频处理可能耗时较长。
5. 提交卷的更改 - 写入后务必调用 commit()。
6. 使用临时磁盘 - 存放处理期间的临时文件。
7. 监控成本 - 跟踪执行时间和资源使用情况。
8. 妥善处理错误 - FFmpeg 可能无法处理损坏的输入文件。
9. 测试时使用快速预设 - 在生产环境中切换到较慢的预设。
10. 确认 GPU 功能 - NVENC 可能不可用。

相关技能

• ffmpeg-opencv-integration - 用于 FFmpeg + OpenCV 组合管线，包括：
  • BGR/RGB 颜色格式转换（OpenCV=BGR，FFmpeg=RGB）
  • 帧坐标注意事项（img[y,x] 而非 img[x,y]）
  • 使用 ffmpegcv 进行 GPU 加速的视频 I/O（NVDEC/NVENC）
  • 使用 VidGear 进行多线程流式处理
  • 使用 Decord 批量加载机器学习视频（比 OpenCV 快 2 倍）
  • 使用 PyAV 实现帧级精度
  • 使用 Modal map() 进行并行帧处理的模式

参考资料

• Modal 文档
• Modal 示例 - Blender Video
• Modal CUDA 指南
• Modal 卷
• Modal 定价