🐡

MediaConvertによる動画変換基盤とStep Functionsを使った環境別デプロイの構築

shogawa

2025/12/26に公開

 はじめにこんにちは、Gakken LEAP のエンジニアの shogawa です。

今回は、動画配信システムの運用を引き継いだ際に直面した課題と、AWS MediaConvert による動画変換基盤の構築、さらに Step Functions を使った環境別デプロイの仕組みについて紹介します。

 前提AWSの環境が整っていること
AWS Lambda、Step Functions、MediaConvert、S3 の基本的な知識があること
Python（または他の言語）でLambda関数を作成できること

 背景：既存構成を把握するところからのスタートプロジェクトでは、他社が開発した動画配信システムを移行し運用することとなりましたが、情報の整理が必要な状態から手探りで運用を開始することになりました。
動作している環境を調査したところ、以下のような構成が判明しました：

入力バケット（S3）: 動画の元となるmp4ファイルが保存されている

配信バケット（S3）: 実際に配信に使用されるストリーミング変換後のファイル（.m3u8や.tsファイル）が保存されている
しかし、以下のような課題がありました：

検証環境も本番環境も区別がない - 全て同じ配信バケットを参照している

変換プロセスが不明 - どうやって入力バケットのmp4ファイルが配信バケットのm3u8ファイルに変換されているのか不明

リリースフロー未整備 - 変換後すぐに本番公開されてしまう状態

 既存構成（推測）
 解決策：環境分離とワークフロー整備
 新構成の設計方針以下の方針で新しい構成を検討しました：

環境を明確に分離する - 検証環境（stg）と本番環境（prd）を用意

変換後すぐに公開しない - 変換後のファイルを一時保管し、承認後にリリース

ワークフローを可視化する - Step Functions で処理フローを管理

 新構成の全体像この構成により、以下を実現できました：
✅ 変換後のファイルは変換済みバケットに一時保存
✅ 検証リリース向けのStep Functions（Step Functions-stg）と本番リリース向けのStep Functions（Step Functions-prd）を用意し、それぞれのタイミングで手動実行
✅ Step Functions-stgは検証環境（配信バケット-stg）へコピー、Step Functions-prdは本番環境（配信バケット-prd）へコピー
✅ 各環境のシステムは、それぞれ対応する配信バケットを参照

 実装：S3アップロード検知とMediaConvert起動
 Lambda関数（動画変換起動）入力バケットにmp4ファイルがアップロードされると、S3イベント通知でLambda関数が起動し、MediaConvertジョブを実行します。
import json
import boto3
import os
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()

s3_client = boto3.client('s3')
mediaconvert_client = boto3.client('mediaconvert', endpoint_url=os.environ['MEDIACONVERT_ENDPOINT'])

def create_mediaconvert_job(input_bucket, input_key, output_bucket):
    """MediaConvertジョブを作成してHLS形式に変換"""
    # 出力先パスの生成
    file_name_without_ext = os.path.splitext(os.path.basename(input_key))[0]
    directory_path = os.path.dirname(input_key)
    output_key_prefix = os.path.join(directory_path, file_name_without_ext).strip('/') + '/'
    
    logger.info(f"Output: s3://{output_bucket}/{output_key_prefix}")
    
    # S3に出力先ディレクトリを事前作成（エラー防止）
    s3_client.put_object(Bucket=output_bucket, Key=output_key_prefix)
    
    # MediaConvert ジョブ設定（主要な部分のみ抜粋）
    job_settings = {
        "Role": os.environ['MEDIACONVERT_ROLE'],
        "Settings": {
            "OutputGroups": [{
                "Name": "Apple HLS",
                "Outputs": [{
                    "NameModifier": "_hls",
                    "ContainerSettings": {"Container": "M3U8"},
                    "VideoDescription": {
                        "CodecSettings": {
                            "Codec": "H_264",
                            "H264Settings": {
                                "Bitrate": 1228800,  # 約1.2Mbps
                                "RateControlMode": "CBR",
                                "GopSize": 90
                            }
                        }
                    },
                    "AudioDescriptions": [{
                        "CodecSettings": {
                            "Codec": "AAC",
                            "AacSettings": {"Bitrate": 96000, "SampleRate": 48000}
                        }
                    }]
                }],
                "OutputGroupSettings": {
                    "Type": "HLS_GROUP_SETTINGS",
                    "HlsGroupSettings": {
                        "SegmentLength": 10,
                        "Destination": f"s3://{output_bucket}/{output_key_prefix}"
                    }
                }
            }],
            "Inputs": [{
                "FileInput": f"s3://{input_bucket}/{input_key}",
                "AudioSelectors": {"Audio Selector 1": {"DefaultSelection": "DEFAULT"}},
                "VideoSelector": {"ColorSpace": "FOLLOW"}
            }]
        }
    }
    
    response = mediaconvert_client.create_job(**job_settings)
    logger.info(f"Job created: {response['Job']['Id']}")
    return response['Job']['Id']

def lambda_handler(event, context):
    """S3イベントをトリガーとして動画変換ジョブを実行"""
    logger.info("Event: " + json.dumps(event))
    
    input_bucket = os.environ['INPUT_BUCKET']
    output_bucket = os.environ['OUTPUT_BUCKET']
    
    try:
        for record in event['Records']:
            input_key = record['s3']['object']['key']
            logger.info(f"Processing: {input_key}")
            create_mediaconvert_job(input_bucket, input_key, output_bucket)
        
        return {'statusCode': 200, 'body': json.dumps('Jobs created successfully.')}
    except Exception as e:
        logger.error(f"Error: {e}")
        return {'statusCode': 500, 'body': json.dumps(f"Error: {str(e)}")}

 ポイント
S3イベント通知: 入力バケットに .mp4 ファイルがアップロードされたときにLambdaを起動

出力先ディレクトリの事前作成: MediaConvertが出力する前にS3にディレクトリを作成（エラー防止）

HLS形式への変換: H.264（動画）+ AAC（音声）でエンコード、10秒間隔のセグメントファイル（.ts）とプレイリスト（.m3u8）を生成

ファイルパス管理: 入力ファイルのディレクトリ構造を保持したまま出力

環境変数: MEDIACONVERT_ENDPOINT, MEDIACONVERT_ROLE, INPUT_BUCKET, OUTPUT_BUCKET を設定

エラーハンドリング: 各処理でエラーをログ出力し、適切に例外を処理

 注意事項：大量ファイルの同時アップロード数百ファイル単位で同時にアップロードする場合、以下のような問題が発生する可能性があります：

S3アップロードエラー: 大量の同時アップロードによりS3側でタイムアウトやレート制限が発生

→今後の課題

Lambda同時実行数の上限: デフォルトの同時実行数制限に達し、一部のファイルでLambdaが起動されない

MediaConvertジョブが実行されない: エラーによりジョブ自体が作成されないケースがある

 S3イベント通知の設定入力バケットのプロパティから、以下のようにイベント通知を設定します。

 設定例

設定項目
設定値


名前
VideoUploadTrigger

イベントタイプ
すべてのオブジェクト作成イベント (s3:ObjectCreated:*)

フィルター（サフィックス）
.mp4

送信先タイプ
Lambda 関数

送信先

TriggerMediaConvertFunction (上記のLambda関数)


 設定手順S3コンソールで入力バケットを開く

プロパティ タブを選択

イベント通知 セクションで イベント通知を作成 をクリック
上記の設定値を入力

変更を保存 をクリック
これにより、.mp4 ファイルが入力バケットにアップロードされるたびに、自動的にLambda関数が起動されます。

 動作確認：MediaConvertジョブの確認Lambda関数が正常に動作し、MediaConvertのジョブが実行されているかは、以下の方法で確認できます。

 1. MediaConvertコンソールでジョブを確認AWSコンソールの MediaConvert → ジョブ から、実行されたジョブの履歴を確認できます。
確認項目：


項目
説明
例


ジョブID
各ジョブに付与される一意のID
1234567890123-abcde

最初の入力ファイル名
変換元のファイル名
sample_video.mp4

送信時間
ジョブが開始された時刻
2025-11-07 18:32:30

終了時刻
ジョブが完了した時刻
2025-11-07 18:32:55

ステータス
ジョブの実行状態

COMPLETE（成功）、ERROR（失敗）

ステータスが COMPLETE であれば、動画変換が正常に完了しています。

 2. 出力先バケット（変換済みバケット）を確認MediaConvertジョブが成功すると、変換済みバケットに以下のファイルが生成されます：
s3://converted-bucket/
└── videos/
    └── sample_video/
        ├── sample_video_hls.m3u8      # マスタープレイリスト
        ├── sample_video_hls00000.ts   # 動画セグメント1
        ├── sample_video_hls00001.ts   # 動画セグメント2
        ├── sample_video_hls00002.ts   # 動画セグメント3
        └── ...
確認手順：
S3コンソールで変換済みバケットを開く
入力ファイル名に対応するディレクトリが作成されていることを確認

.m3u8 ファイル（プレイリスト）と .ts ファイル（セグメント）が存在することを確認
これらのファイルが正常に生成されていれば、動画変換基盤が正しく動作しています。

 Step Functionsによる環境別デプロイ変換完了後、検証リリース向けと本番リリース向けの2つのStep Functionsを用意し、それぞれのリリースタイミングで手動実行します。

Step Functions-stg（検証リリース向け）: 変換済みバケットから配信バケット-stg（検証環境）へファイルをコピー

Step Functions-prd（本番リリース向け）: 変換済みバケットから配信バケット-prd（本番環境）へファイルをコピー
Step Functions を使うことで、ワークフローの可視化と柔軟な制御が可能になります。
実際の運用では、複数のディレクトリを並列処理することで、大量ファイルのコピーを効率化しています。

 Step Functions定義（JSON）以下は、検証環境へのコピーを行うStep Functions-stgの定義例です。本番環境向けのStep Functions-prdも同様の定義で、コピー先を本番バケットに変更して作成します。
{
  "Comment": "動画ファイルの並列コピーワークフロー（検証環境向け）",
  "StartAt": "ParallelCopy",
  "States": {
    "ParallelCopy": {
      "Type": "Parallel",
      "Comment": "複数ディレクトリを並列コピー",
      "Branches": [
        {
          "StartAt": "CopyBatch1",
          "States": {
            "CopyBatch1": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:CopyVideoFilesFunction",
              "Comment": "バッチ1のコピー",
              "Parameters": {
                "sourceBucket": "converted-bucket",
                "destinationBucket": "distribution-stg-bucket",
                "prefixes": [
                  "videos/batch1/"
                ],
                "destinationPrefix": "hls/"
              },
              "End": true
            }
          }
        },
        {
          "StartAt": "CopyBatch2",
          "States": {
            "CopyBatch2": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:CopyVideoFilesFunction",
              "Comment": "バッチ2のコピー",
              "Parameters": {
                "sourceBucket": "converted-bucket",
                "destinationBucket": "distribution-stg-bucket",
                "prefixes": [
                  "videos/batch2/"
                ],
                "destinationPrefix": "hls/"
              },
              "End": true
            }
          }
        }
      ],
      "End": true
    }
  }
}

 ワークフローの特徴
環境別のStep Functions: 検証用（Step Functions-stg）と本番用（Step Functions-prd）を分けて用意し、リリースタイミングで手動実行

Parallelステート: 複数のディレクトリのコピー処理を並列実行することで、処理時間を短縮

ブランチ構成: 上記の例では、バッチ1とバッチ2のディレクトリを並列処理

パラメータ:

sourceBucket: コピー元（変換済みバケット）

destinationBucket: コピー先（検証用は配信バケット-stg、本番用は配信バケット-prd）

prefixes: コピー対象のディレクトリリスト（複数指定可能）

destinationPrefix: コピー先のプレフィックス


 実運用での活用実際の運用では、以下のように運用しています：
Step Functions-stg（検証リリース向け）:
検証リリース時に手動実行
複数ディレクトリを並列コピーして配信バケット-stgへデプロイ
検証環境で動作確認
Step Functions-prd（本番リリース向け）:
本番リリース時に手動実行
複数ディレクトリを並列コピーして配信バケット-prdへデプロイ
本番環境へ反映
実装例：
動画カテゴリごとにディレクトリを分割（video/1/, video/2/, ...）
各ディレクトリを並列でコピー
処理時間を大幅に短縮

 Lambda関数（ファイルコピー）実際のファイルコピー用Lambda関数は以下のような処理を行います：
import boto3
import os

s3 = boto3.client('s3')

def lambda_handler(event, context):
    """
    Step Functionsから呼び出され、複数ディレクトリのファイルをコピー
    
    Parameters:
        sourceBucket: コピー元バケット（変換済みバケット）
        destinationBucket: コピー先バケット（配信バケット-stgまたは配信バケット-prd）
        prefixes: コピー対象のディレクトリリスト（例: ["videos/batch1/"]）
        destinationPrefix: コピー先のプレフィックス（例: "hls/"）
    """
    source_bucket = event['sourceBucket']
    destination_bucket = event['destinationBucket']
    prefixes = event['prefixes']
    destination_prefix = event['destinationPrefix']
    
    # 各プレフィックス（ディレクトリ）を順次処理
    for prefix in prefixes:
        copy_directory(source_bucket, destination_bucket, prefix, destination_prefix)
    
    return {'status': 'completed'}

def copy_directory(source_bucket, destination_bucket, source_prefix, destination_prefix):
    """
    指定されたプレフィックスのファイルを再帰的にコピー
    """
    # S3のページネーターを使用して大量ファイルに対応
    paginator = s3.get_paginator('list_objects_v2')
    
    for page in paginator.paginate(Bucket=source_bucket, Prefix=source_prefix):
        for obj in page.get('Contents', []):
            # コピー元の情報
            copy_source = {'Bucket': source_bucket, 'Key': obj['Key']}
            
            # コピー先のキーを生成
            # 例: source_prefix が "videos/batch1/" の場合
            #     obj['Key'] が "videos/batch1/sample/file.ts" なら
            #     relative_path は "sample/file.ts"
            #     stripped_prefix は "batch1/"
            #     dest_key は "hls/batch1/sample/file.ts" となる
            relative_path = obj['Key'][len(source_prefix):]
            stripped_prefix = source_prefix.split('/', 1)[-1]  # 最初の階層を除去
            dest_key = os.path.join(destination_prefix, stripped_prefix, relative_path).replace("\\", "/")
            
            # ファイルをコピー
            s3.copy_object(CopySource=copy_source, Bucket=destination_bucket, Key=dest_key)

 コピー処理のポイント
シンプルな実装: イベント駆動で必要な処理のみを実装

複数ディレクトリの処理: prefixes リストを順次処理することで、一度の実行で複数ディレクトリをコピー

ページネーション: paginator を使用して、大量のファイルを確実に取得

パス変換: ソースのディレクトリ構造を保ちつつ、デスティネーションプレフィックスを適切に追加

クロスプラットフォーム対応: replace("\\", "/") でWindows環境でも正しいS3パスを生成

 実行例Step Functionsから以下のパラメータで呼び出された場合：
{
  "sourceBucket": "converted-bucket",
  "destinationBucket": "distribution-stg-bucket",
  "prefixes": ["videos/batch1/"],
  "destinationPrefix": "hls/"
}
コピー結果：

s3://converted-bucket/videos/batch1/sample/video.m3u8

→ s3://distribution-stg-bucket/hls/batch1/sample/video.m3u8

s3://converted-bucket/videos/batch1/sample/video00000.ts

→ s3://distribution-stg-bucket/hls/batch1/sample/video00000.ts

 メリットと効果この構成により、以下のような効果が得られました：

 1. 環境分離の実現検証環境（stg）と本番環境（prd）を明確に分離
検証後に本番リリースという正常なフローを確立

 2. ワークフローの可視化Step Functions により、デプロイフローが一目で分かる
各ステップの実行状況をコンソールから確認可能

 3. 柔軟な制御検証用と本番用のStep Functionsを分離することで、各環境へのリリースタイミングを個別に制御
検証環境での確認後、本番環境へのリリースを別タイミングで実行可能
必要に応じて承認ステップを追加可能

 4. 運用の安定化変換後すぐに本番公開されない仕組み
エラー時の通知とリトライ機能

 今後の課題と改善点現在の構成で基本的な運用は可能になりましたが、以下のような改善余地があります：

 1. 手動操作の削減現在は検証用（Step Functions-stg）と本番用（Step Functions-prd）のStep Functionsをそれぞれのリリースタイミングで手動実行していますが、以下のような自動化を検討中：
MediaConvert 完了イベントを検知して自動的にStep Functions-stgを起動（検証環境への自動デプロイ）
承認ワークフローの組み込み（AWS Step Functions の人間の承認ステップ）を追加し、承認後に自動的にStep Functions-prdを起動

 2. 拡張性の向上
進捗通知: 変換進捗をリアルタイムに通知

大量ファイル処理の改善: SQSキューを導入して、数百ファイルの同時アップロードにも安定して対応

 3. コスト最適化MediaConvert の料金監視とアラート設定
不要な変換ファイルの自動削除（ライフサイクルポリシー）

 4. 監視・ログの強化CloudWatch Logs Insights でエラー分析

 まとめ既存構成を把握するところから、AWS MediaConvert による動画変換基盤の構築と、Step Functions を使った環境別デプロイの仕組みを整備した事例を紹介しました。
ポイント：
S3 + Lambda + MediaConvert でサーバーレスな動画変換を実現
Step Functions でワークフローを可視化し、環境別デプロイを制御
検証環境と本番環境を分離し、安全な運用フローを確立
まだ改善の余地はありますが、手動操作と不確実性が多かった運用から、ある程度自動化された運用への移行ができました。
今後は承認フローの自動化や、エラーへの対応を検討していきます。
同じような課題を抱えている方の参考になれば幸いです。

 参考資料AWS MediaConvert ドキュメント
AWS Step Functions ドキュメント
S3 イベント通知の設定

 エンジニア募集中Gakken LEAPは日々、教育のアップデートに取り組んでいます。

興味をお持ちいただいた方は、採用サイトをチェックしてみてください。

設定項目	設定値
名前	`VideoUploadTrigger`
イベントタイプ	すべてのオブジェクト作成イベント (`s3:ObjectCreated:*`)
フィルター（サフィックス）	`.mp4`
送信先タイプ	Lambda 関数
送信先	`TriggerMediaConvertFunction` (上記のLambda関数)

項目	説明	例
ジョブID	各ジョブに付与される一意のID	`1234567890123-abcde`
最初の入力ファイル名	変換元のファイル名	`sample_video.mp4`
送信時間	ジョブが開始された時刻	`2025-11-07 18:32:30`
終了時刻	ジョブが完了した時刻	`2025-11-07 18:32:55`
ステータス	ジョブの実行状態	`COMPLETE`（成功）、`ERROR`（失敗）

Gakken LEAP DevelopersPublication

Ed-Tech事業の開発及びコンサルティング業務 / 学研グループ各社およびグループ外企業のIT事業・サービスに関する企画、開発、導入支援、運用並びにそれらに関するコンサルティング / ベンチャー企業・ベンチャーキャピタルに対する投資と養成、を行なっている会社です。 gakken-leap.co.jp/

はじめに

前提

背景：既存構成を把握するところからのスタート

既存構成（推測）

解決策：環境分離とワークフロー整備

新構成の設計方針

新構成の全体像

実装：S3アップロード検知とMediaConvert起動

Lambda関数（動画変換起動）

ポイント

注意事項：大量ファイルの同時アップロード

S3イベント通知の設定

設定例

設定手順

動作確認：MediaConvertジョブの確認

1. MediaConvertコンソールでジョブを確認

2. 出力先バケット（変換済みバケット）を確認

Step Functionsによる環境別デプロイ

Step Functions定義（JSON）

ワークフローの特徴

実運用での活用

Lambda関数（ファイルコピー）

コピー処理のポイント

実行例

メリットと効果

1. 環境分離の実現

2. ワークフローの可視化

3. 柔軟な制御

4. 運用の安定化

今後の課題と改善点

1. 手動操作の削減

2. 拡張性の向上

3. コスト最適化

4. 監視・ログの強化

まとめ

参考資料

エンジニア募集中

Discussion