lambdaの使い方例 - S3に置いてある動画を処理する
この記事でできる事
lambdaの関数を作成/テストできる
lambda関数に適切なロールを設定できる
pythonからs3の基本的な操作を行う事ができる
前提条件
S3のバケットにmp4の動画なんかがバラっとあるとする。aws cloudshellなんかで見る事ができる。
aws s3 ls <bucket>とかやってこれを何らかの形で処理したいとした場合にlambdaを使ってみるトレーニング
lambda関数の作成
https://ap-northeast-1.console.aws.amazon.com/lambda/home?region=ap-northeast-1#/functions
アーキテクチャの設定は後々重要になるのでarm64でもx86_64でもいいけど、選択した方はよく覚えておくこと。ここではarm64を選択。またランタイムはPython 3.13を選択している。
lambdaの権限としては今回は当該のバケットの読み書きを可能にする必要がある。
いずれにせよs3-video-process-role-syrfp8ln というロールが自動作成されるという事が理解できるだろう、後でこれを変更する。
ロールにS3への権限を与える
当該ロールをみつける
https://us-east-1.console.aws.amazon.com/iam/home?region=ap-northeast-1#/roles から
先程作成された名前で検索すると当該のロールが出てくるので、これに対してインラインポリシーを作成する
以下のようなjsonをセット。S3に関する操作体系の権限だ。S3のバケット名は適切に置き換える事。
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "Statement1",
"Effect": "Allow",
"Action": [
"s3:ListBucket",
"s3:GetObject",
"s3:PutObject",
"s3:DeleteObject"
],
"Resource": [
"arn:aws:s3:::<バケット名>",
"arn:aws:s3:::<バケット名>/*"
]
}
]
}名前は <バケット名>-s3-access-policy など、適当に命名する。
lambda関数から権限を確認していく
うまいことロールの設定ができたかどうかのチェックをするため、lambda関数で、s3の中身にアクセスし、書いたり消せたりするか試してから実際の処理に入る。
というわけでlambda関数のコードを見ていこう
このHello from lambda的なコードを以下のように置き換える。
import boto3
s3 = boto3.client('s3')
def lambda_handler(event, context):
bucket_name = "bucket-name"
# 1. バケットのリスト
try:
response = s3.list_objects_v2(Bucket=bucket_name)
print("S3 List Response: ", response.get('Contents', []))
except Exception as e:
print("Error listing bucket:", e)
# 2. ダミーファイルのアップロード
try:
s3.put_object(
Bucket=bucket_name,
Key="dummy-test-file.txt",
Body="This is a test file."
)
print("Dummy file created successfully.")
except Exception as e:
print("Error creating dummy file:", e)
# 3. ダミーファイルの削除
try:
s3.delete_object(
Bucket=bucket_name,
Key="dummy-test-file.txt"
)
print("Dummy file deleted successfully.")
except Exception as e:
print("Error deleting dummy file:", e)
return {"status": "done"}
デプロイとテスト
上記コードをDeployする。ここで言うDeployとは、コードをAWS上に保存してAWS上のlambdaから実行する準備をする事を差す
テスト(実行)を行う。これは上記画像から行ってもいいが、下のタブから行ってもいい。今回はタブに遷移している
以降の画面から新しいイベントを作成し「テスト」を押す。イベントとはある程度lambda関数の引数を柔軟に変えるようなセットの事を示すがここでは単純にコードを実行できればよいので特に考える事はない。
すると以下のように実行される
statusがdoneになっていればよい
下の方にログが出てるはずだ(s3に何か入ってればだけど、まあ今回は動画が詰まっているのを想定している)
ffmpegを実行できるようにする
動画を処理するにあたってはffmpegコマンドをpythonから実行する手法を取る事にする。ただし、lambdaの実行環境にはffmpegコマンドは用意されていないから、取得してアップロードする必要がある
ffmpegバイナリの取得
上記サイトからから使っているアーキテクチャに応じたバイナリをゲットする
ここでは実行環境でここでarm64を選んでいたのでそのtar.xvを取得。展開
amd64とかと間違えないように
wget https://johnvansickle.com/ffmpeg/builds/ffmpeg-git-arm64-static.tar.xz
tar xvf ffmpeg-git-arm64-static.tar.xzffmpegをzipしてアップロードする
tarで展開するとゴロっとファイルが表われるが、必要なのはffmpegコマンドで使うバイナリだけなので、それだけをzipする
cd ffmpeg-git-20240629-arm64-static
zip ffmpeg.zip ffmpeg # ffmpegコマンドだけをzip
zipinfoするとこんな感じになるだろう
zipinfo ffmpeg.zip
Archive: ffmpeg.zip
Zip file size: 25669451 bytes, number of entries: 1
-rwxr-xr-x 3.0 unx 51138224 bx defN 24-Jun-30 13:07 ffmpeg
1 file, 51138224 bytes uncompressed, 25669289 bytes compressed: 49.8%50Mを越えてなければ大丈夫。今回は25Mほどになった
レイヤーの追加
lambda環境に新たなファイルなどをセットする場合「レイヤー」という概念で行われる。言葉は難儀な感じもあるが、要するにzipから当該環境の/optにマウントするという事だ。今回はzipの中にffmpegコマンド1つが入っているためレイヤーがうまく追加された場合は /opt/ffmpeg となるはずである。
でレイヤーの追加だが、これは左ペインから追加していく
レイヤーの作成を押すと、アップロードできるようになる。アーキテクチャーに関してはここでは宣言だけなので実際にここを間違えてもそんな問題ないかもしれないが間違えないようにする
ここで作成されたARNを控えておくこと。まあ控えるの忘れたらまたコピーしにくる手間が発生するだけだが。
関数にレイヤーを割り当てる
当該関数に戻ってきて
これを押すとレイヤーの追加ができる
ここで
先ほどコピーしたARNを貼り付けるわけだ。ここでffmpegが正しく機能しているかどうか検証コードに置き換えてテストしてみよう
レイヤー(ffmpeg)検証コード
import subprocess
import os
def lambda_handler(event, context):
try:
# FFmpegのバージョンを確認
result = subprocess.run(["/opt/ffmpeg", "-version"], capture_output=True, text=True, check=True)
print("FFmpeg version:")
print(result.stdout)
return {
"status": "success",
"ffmpeg_version": result.stdout
}
except Exception as e:
print(f"Error: {e}")
return {
"status": "error",
"message": str(e)
}
例によってどっちからテストしてもいいんだけど、無事レイヤーが追加されていれば以下のようになるだろう
動画を1つピックアップしてffmpegを実行
ここではmp4が入ってると仮定する。movだとちょっとこの処理だとうまくいかないかもしれないね。とりあえず動画を1つピックアップするコード
import boto3
s3 = boto3.client('s3')
def lambda_handler(event, context):
bucket_name = "バケット名"
try:
# S3バケット内のオブジェクトをリスト
response = s3.list_objects_v2(Bucket=bucket_name)
contents = response.get('Contents', [])
# mp4を検索
video_file = None
for obj in contents:
key = obj['Key']
if key.lower().endswith('mp4'):
video_file = key
break
if video_file:
print(f"Selected video file: {video_file}")
else:
print("No MP4 or MOV files found in the bucket.")
return {"selected_video": video_file}
except Exception as e:
print(f"Error: {e}")
return {"error": str(e)}ここでは音声を抜き出してS3にアップロードしてみる
import boto3
import os
import subprocess
s3 = boto3.client('s3')
def lambda_handler(event, context):
bucket_name = "ritsumei"
try:
# S3バケット内のオブジェクトをリスト
response = s3.list_objects_v2(Bucket=bucket_name)
contents = response.get('Contents', [])
# mp4ファイルを検索
video_file = None
for obj in contents:
key = obj['Key']
if key.lower().endswith('.mp4'):
video_file = key
break
if not video_file:
print("No MP4 files found in the bucket.")
return {"status": "error", "message": "No MP4 files found"}
print(f"Selected video file: {video_file}")
# 動画を一時ディレクトリにダウンロード
video_path = f"/tmp/{os.path.basename(video_file)}"
audio_path = video_path.rsplit('.', 1)[0] + ".mp3"
s3.download_file(bucket_name, video_file, video_path)
print(f"Downloaded video file to {video_path}")
# FFmpegで音声を抽出
try:
ffmpeg_cmd = [
"/opt/ffmpeg", "-i", video_path, "-q:a", "0", "-map", "a", audio_path
]
subprocess.run(ffmpeg_cmd, check=True)
print(f"Audio extracted to {audio_path}")
except subprocess.CalledProcessError as e:
print(f"Error during audio extraction: {e}")
return {"status": "error", "message": "Audio extraction failed"}
# 抽出した音声をS3にアップロード
try:
audio_file_name = os.path.basename(video_file).rsplit('.', 1)[0] + ".mp3"
s3.upload_file(audio_path, bucket_name, audio_file_name)
print(f"Uploaded audio to S3 as {audio_file_name}")
except Exception as e:
print(f"Error uploading audio to S3: {e}")
return {"status": "error", "message": "Audio upload failed"}
return {"status": "success", "audio_file": audio_file_name}
except Exception as e:
print(f"Error: {e}")
return {"status": "error", "message": str(e)}
このコードはうまく動作するが、lambdaのスペックを変更する場合は以下の設定項目をちょいちょいいじっていくとよい。特にタイムアウトが3秒だとほとんど失敗するはずなので増やしておこう(最大15分)
#aws #aws初心者 #aws学習 #lambda #python #boto3