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）

CPU コンテナと GPU コンテナを使って、Modal のサーバーレス Python プラットフォームで FFmpeg を実行するための完全ガイドです。

概要

Modal は、クラウドで Python コードを実行するためのサーバーレスプラットフォームです。次の機能があります：

• 1 秒未満のコールドスタート - コンテナは数ミリ秒で起動します
• 柔軟な GPU 容量 - T4、A10G、L40S、H100 GPU にアクセスできます
• 並列処理 - 数千のコンテナに即座にスケールできます
• 従量課金 - アイドル時間ではなく CPU サイクルに基づいて課金されます

Modal と従来のクラウドの比較

機能	Modal	従来の VM
コールドスタート	1 秒未満	数分
スケーリング	数千まで自動	手動で設定
課金	実行ごと	時間ごと
GPU アクセス	gpu="any" デコレーター	複雑なプロビジョニング
設定	Python デコレーター	Infrastructure as Code

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 コンテナは、主に ML 推論のために NVIDIA GPU を使用します。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)

Volumes へのアップロード

# 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. 大きなファイルにはVolumeを使用する - 大容量のデータを引数として渡すのは避けます
4. 適切なタイムアウトを設定する - 動画処理には時間がかかることがあります
5. Volumeの変更をコミットする - 書き込み後は必ず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]ではありません）
  • GPUアクセラレーション対応の動画入出力（NVDEC/NVENC）向けのffmpegcv
  • マルチスレッドストリーミング向けのVidGear
  • MLのバッチ動画読み込み向けのDecord（OpenCVより2倍高速）
  • フレーム単位の精密な処理向けのPyAV
  • Modalのmap()を使用した並列フレーム処理のパターン

参考資料

• Modalのドキュメント
• Modalの例 - Blender Video
• Modal CUDAガイド
• Modal Volumes
• Modalの料金