LambdaからS3に大量ファイルを高速アップロードする方法

5月に入り、朝から気持ちの良い青空が広がっています。

今回は、サーバーレスの代表選手「AWS Lambda」の処理のお話です。

Lambdaの処理のタイムアウトは15分のため、バッチ処理等でLambdaを利用する場合は15分以内の処理時間に収める必要があります。

また、「Amazon API Gateway」のバックエンドとしてLambdaを利用する場合は、API Gatewayのタイムアウトは29秒のため、29秒以内の処理時間に収める必要があります。

可能な限り、Lambdaの処理を高速化したいですね。
高速化の方法として

1. Lambdaのスペック（メモリ＋コア数）を上げる

が基本としてありますが、それとあわせて

2. Lambdaの中で並行、並列処理（マルチスレッド化、マルチプロセス化）を行う

が効果的です。

Lambdaの言語としてPythonを利用する場合、
PythonはGIL（プログラムが1つのCPUコア上で1つだけ実行されることを保証する仕組み）があるので、CPUバウンド（CPUに負荷がかかる数値計算等）な処理をマルチスレッドで高速化することは困難となります。この場合は、マルチプロセス化が有効です。

一方、I/Oバウンド（入出力に負担がかかるファイル読み書き、DB接続、ネットワーク通信等）な処理は、マルチスレッド化が有効となります。

このようなイメージとなり、入出力待ちのCPUの空き時間に、他のスレッドが割り当てられるため、処理時間が短くなります。

I/Oバウンドの処理の例として、「Amazon EFS」にある大量のファイルを「Amazon S3」にアップロードする場合があります。
もし、シングルスレッドで、この処理を実行すると

このように、直列の処理になるので、時間がかかってしまいますね。

マルチスレッド化した場合のコードを見てみましょう。

1. ライブラリをインポート

import boto3 #AWSのサービスへのアクセス用
import threading #マルチスレッド実装用
import os #ファイル取得用

2. 変数の定義

s3 = boto3.resource('s3')
bucket_name = 's3-png' #S3のバケット名
num_threads = 10 #スレッドの数
directory = '/mnt/png' #EFSのパス

3. S3へのアップロード処理

def upload_files(file_names):
    for file_name in file_names:
        s3.Bucket(bucket_name).upload_file(file_name, file_name)

4. メイン処理

def lambda_handler(event, context):

    file_names = []
    for root, dirs, files in os.walk(directory):
        for file in files:
            if file.endswith('.png'): #.pngの拡張子を持つファイルを取得
                file_names.append(os.path.join(root, file))

    threads = []
    for i in range(num_threads):
        thread_files = file_names[i::num_threads] #ファイルを分割して各スレッドに割り当てる
        thread = threading.Thread(target=upload_files, args=(thread_files,)) #各スレッドはその割り当てられたファイルでupload_files関数を実行する
        threads.append(thread)
        thread.start() #各スレッドを開始
    
    for thread in threads:
        thread.join() #全てのスレッドが終了するのを待つ

    #成功レスポンス
    return {
        'statusCode': 200,
        'body': 'All files uploaded successfully'
    }