AWS Lambdaで並列処理を行うことで処理を高速化する(Python編)

はじめに

みなさんAWS Lambda(以降はLambdaと表記)使ってますか？
Lambdaには対応している言語が複数あるので、Go言語で記述されている方、Node.jsで記述されている方、Pythonで記述されているかたなど、
様々いらっしゃるかと思います。
APIのバックエンドとしてLambdaを設置する場合、Lambdaの処理時間は出来る限り短くしたい、と思われる方がほとんどだと思います。
理由としては、下記が挙げられるかと思います。

• API Gatewayの時間制限が29秒のため、29秒を超える処理が出来ない
• APIの呼び出しに時間がかかると、ユーザービリティを著しく損なう

では、Labmbdaの処理時間を短くしたいのですが、方法としては下記があります。

• Lambdaの性能を上げる
• Lambdaの中で並列処理(マルチスレッド化、マルチプロセス化)を行う

今回は、Lambdaでの並列処理について解説したいと思います。

実践

それぞれの検証は、下記共通の条件で実施しています。

メモリ：3008 MB
タイムアウト：1分

並列処理なしの場合

検証用に簡単なコードを書いてみます。
処理内容としては、Lamba関数の一覧を取得し、その後、ダミー処理を実行するソースコードになります。

import json
import os
import boto3
from aws_xray_sdk.core import patch

patch(['boto3'])

client = boto3.client('lambda')

def child_process_func():
    # Lambda関数の一覧を取得
    client.list_functions()
    
    # ダミー処理のため、特に意味はありません
    for i in range(1000000):
        a = i*i
    return os.getpid()

def lambda_handler(event, context):
    for i in range(3):
        print(child_process_func())
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

実際の処理時間を可視化してみると下記画像のようになります。
ListFunctionsと書かれている部分が、Lambda関数の一覧を取得する処理になります。
Lambda関数の一覧取得後、ダミー処理を入れているため、空白部分はダミー処理をしている時間となります。
Lambda関数の一覧取得+ダミー処理のセットを合計3回行っており、同期処理となるので、下記画像のようになります。
並列処理なしの場合は、合計処理時間が約442ミリ秒という結果でした。

並列処理(マルチスレッド化)

続いて、マルチスレッド処理に変更して、同じ内容のコードを作成してみましょう。
今回はThreadPoolExecutorクラスを使ってマルチスレッドを実現します。

import json
import os
from concurrent.futures import ThreadPoolExecutor
import boto3
from aws_xray_sdk.core import patch

patch(['boto3'])

client = boto3.client('lambda')

def child_process_func():
    # Lambda関数の一覧を取得
    client.list_functions()
    
    # ダミー処理のため、特に意味はありません
    for i in range(1000000):
        a = i*i
    return os.getpid()

def lambda_handler(event, context):
    process_list = []
    with ThreadPoolExecutor() as executor:
        for i in range(3):
            process_list.append(executor.submit(child_process_func))
        for process in process_list:
            print(process.result())

    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

処理時間を可視化した結果は下記です。
注目して頂きたいのは、ListFunctionsの処理部分が重なっているということです。(下記画像の赤枠部分)
マルチスレッド化したことにより、API呼び出しの待ち時間を別のAPI呼び出しに使うことが出来るため、計算リソースを効率よく利用できている事がわかります。
並列処理(マルチスレッド化)ありの場合は、合計処理時間が約282ミリ秒という結果でした。

ここで1つ、マルチスレッド化したことにより、ListFunctionsの処理時間が増加していることに疑問を持たれるかもしれません。
ここでポイントとなるのは、マルチスレッド化しても実行されるプロセスは1つのみ、という点です。
例えば、１つ目のListFunctionsがAPI呼び出しを完了したとしても、別の処理が既に実行されている場合は、1つめのListFunctionsは、
待たされてしまう、ということが発生します。
なので、マルチスレッド化する場合、API呼び出しやI/O呼び出しの待ち時間が長い場合は有効ですが、プログラムがずっと動き続けているような場合は、
マルチスレッド化することで逆に処理時間が伸びる、という結果になる可能性があります。

並列処理(マルチプロセス化)

Pythonは基本的には実行できるスレッドは1つのみ、という制限があります。(global interpreter lock)
ただし、この制限を回避するため、処理を別プロセスに切り出してマルチプロセスで実行することで、より速度を早めることが可能です。
マルチプロセス化を実現するために、multiprocessingパッケージを使います。

import json
import os
import multiprocessing
import boto3
from aws_xray_sdk.core import patch

patch(['boto3'])

client = boto3.client('lambda')

def child_process_func(conn):
    # Lambda関数の一覧を取得
    client.list_functions()
    
    # ダミー処理のため、特に意味はありません
    for i in range(1000000):
        a = i*i

    # マルチプロセスのため、プロセス間通信でデータを送信する
    conn.send(os.getpid())
    conn.close()


def lambda_handler(event, context):
    process_list = []
    parent_connection_list = []
    for i in range(3):
        parent_conn, child_conn = multiprocessing.Pipe()
        parent_connection_list.append(parent_conn)
        process_list.append(multiprocessing.Process(target=child_process_func, args=(child_conn,)))
        
    for process in process_list:
        process.start()
        
    for parent_connection in parent_connection_list:
        print(parent_connection.recv())

    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

処理時間を可視化した結果は下記です。
並列処理(マルチプロセス化)ありの場合は、合計処理時間が約223ミリ秒という結果でした。
マルチプロセスにすることで、ダミー処理の部分(FOR文でひたすら足し算している部分)が並列で実行されるため、その部分の処理が短縮されたと考えられます。

まとめ

AWS Lambdaでも、並列処理を行うことで効率よく処理を行うことが出来る、ということがご理解頂けたかと思います。
マルチスレッドとマルチプロセスは、同じように見えて実は全然利用用途が異なります。
API呼び出しなどが多い場合はマルチスレッドのほうがより効果的ですし、計算処理が多い場合はマルチプロセスの方が効果的です。

また、Lambdaでマルチプロセスを扱う場合、かなりLambdaの性能を上げないと効果が出ない可能性があります。
今回の検証も、Lambdaの性能を最大にして行っているのは、そのためです。

非同期処理は、コードの複雑性を上げる可能性もあるのですが、用法・容量を正しく守って、効率的なコードが書ければと思います。
では(^o^)