AWS LambdaとAmazon SQSを使ったイベントソースマッピングの解説。

はじめに

現代のクラウドコンピューティングでは、大量のデータ処理を効率的に行うために、様々な技術が活用されています。その中でも、AWS Lambda と Amazon SQS を組み合わせたイベントソースマッピングは、非同期処理の強力なソリューションです。本記事では、SQS キューにメッセージが一定件数溜まるか、一定秒数が経過すると Lambda が処理を実行するバッファリングの流れについて解説します。

概要

AWS Lambda は、イベント駆動型のサーバーレスコンピューティングサービスであり、事前に設定されたイベントに基づいてコードを自動的に実行します。一方、Amazon SQS（Simple Queue Service）は、メッセージの送受信を管理する分散型キューイングサービスです。これらを組み合わせることで、特定の条件でバッチ処理を実行することができます。このバッファリングの仕組みは、大量のデータ処理を効率的に行うための強力なツールとなります。

アーキテクチャ

このソリューションの基本的なアーキテクチャは以下の通りです

データ生成: 様々なデータソースからメッセージが SQS キューに送信される。
バッファリング: SQS キューがメッセージを受け取り、指定された条件（メッセージが一定件数溜まるか、一定秒数が経過）を満たすまでメッセージをキューに保持する。
処理のトリガー: 一定の条件が満たされると、イベントソースマッピングにより Lambda 関数がトリガーされる。
データ処理: Lambda 関数がキュー内のメッセージをバッチで処理し、結果を S3 などに保存または次のステップに渡す。

メリット

スケーラビリティ: メッセージの量に応じて自動的にスケールし、大量のデータ処理に対応可能である。
コスト効率: 必要なときにのみ Lambda 関数が実行されるため、無駄なリソース消費を抑制することができる。
非同期処理: メッセージがキューに溜まる間、他のタスクを並行して実行できる。

デメリット

遅延の可能性: メッセージが指定の条件に達するまで待機するため、リアルタイム処理には不向きである。
複雑性の増加: システムの構成が複雑になるため、設定やデバッグに手間がかかる場合がある。

コードのサンプル

1. Terraform でのイベントソースマッピング設定

以下の例では、Terraform を使用して SQS キューと Lambda 関数を設定し、イベントソースマッピングを構成する例となっている。

SQS キューの設定

resource "aws_sqs_queue" "my_queue" {
  name                        = "my_queue"
  visibility_timeout_seconds  = 30
  message_retention_seconds   = 86400
}

コードの説明

name: SQS キューの名前を指定する。
visibility_timeout_seconds: メッセージが取得されてから他の消費者がそのメッセージを再取得できるまでの時間（秒単位）を指定する。
message_retention_seconds: メッセージが SQS キューに保持される最大時間（秒単位）を指定する。

Lambda 関数の設定

resource "aws_lambda_function" "my_lambda" {
  filename         = "lambda_function.zip"
  function_name    = "my_lambda"
  runtime          = "python3.11"
  handler          = "lambda_function.lambda_handler"
  role             = aws_iam_role.lambda_execution_role.arn
}

コードの説明

filename: Lambda 関数のデプロイパッケージのパスを指定します。デプロイパッケージは ZIP ファイル形式である必要があります。
function_name: Lambda 関数の名前を指定する。
runtime: Lambda 関数の実行環境を指定します。ここでは、Python 3.11 を指定する。
handler: Lambda 関数のエントリーポイントを指定します。形式は「ファイル名.関数名」となる。
role: Lambda 関数に割り当てる IAM ロールの ARN を指定する。このロールは Lambda 関数が AWS リソースにアクセスするための権限を持つ。
イベントソースマッピングの設定

resource "aws_lambda_event_source_mapping" "sqs_lambda" {
  event_source_arn                          = aws_sqs_queue.my_queue.arn
  function_name                             = aws_lambda_function.my_lambda.arn
  batch_size                                = 50
  maximum_batching_window_in_seconds        = 10
}

コードの説明

event_source_arn: Lambda 関数がイベントを受け取るソース（ここでは SQS キュー）の ARN を指定する。
function_name: イベントを処理する Lambda 関数の ARN を指定する。
batch_size: Lambda 関数に渡されるメッセージの最大数を指定し、指定されたバッチサイズ分のメッセージがまとめて Lambda 関数に渡される。
maximum_batching_window_in_seconds: Lambda 関数がイベントソースからメッセージをバッチ処理する際の最大待機時間を指定し。この時間内に指定されたバッチサイズに達しない場合でも、バッチ処理が実行される。

2. Lambda 関数の Python コード（S3 に保存するケース）

以下の例では、Lambda 関数がトリガーされ、SQS メッセージの内容を S3 バケットに保存する。

import json
import boto3
import os
from datetime import datetime

s3 = boto3.client('s3')
bucket_name = os.environ['S3_BUCKET_NAME']

def lambda_handler(event, context):
    for record in event['Records']:
        message = json.loads(record['body'])
        file_name = f"data_{datetime.now().strftime('%Y%m%d%H%M%S')}.json"
        s3.put_object(Bucket=bucket_name, Key=file_name, Body=json.dumps(message))
        print(f"Saved message to {file_name} in {bucket_name}")

    return {
        'statusCode': 200,
        'body': json.dumps('Processing complete')
    }

3. Lambda 関数の Python コード（データベースに保存するケース）

以下の例では、Lambda 関数がトリガーされ、SQS メッセージの内容を DynamoDB テーブルに保存する。

import json
import boto3
import os

dynamodb = boto3.resource('dynamodb')
table_name = os.environ['DYNAMODB_TABLE_NAME']
table = dynamodb.Table(table_name)

def lambda_handler(event, context):
    for record in event['Records']:
        message = json.loads(record['body'])
        table.put_item(Item=message)
        print(f"Saved message to DynamoDB: {message}")

    return {
        'statusCode': 200,
        'body': json.dumps('Processing complete')
    }

ユースケース

大量のスクレイピング処理の自動化

ニュースサイトや e コマースサイトなどの情報を定期的に収集する場合、大量のスクレイピング処理を自動化することが求められる。各ページのデータを SQS にメッセージとして送信する。メッセージが 50 件溜まるか、10 秒が経過すると、Lambda 関数がトリガーされ、スクレイピング結果をバッチ処理する。この方法により、人手を介さずに大量のデータを効率的に収集し、処理することが可能である。

データ分析パイプラインの前処理

データ分析パイプラインにおいて、前処理が必要となるケースがある。例えば、センサーデータやログデータを収集し、一定数のデータが集まったら前処理を行う。SQS キューにデータを送り、Lambda 関数でデータのクレンジングやフィルタリングを行い、その結果を S3 バケットに保存する。

通知システムの実装

ユーザーからのフィードバックやエラー通知を効率的に管理するために、SQS キューを利用する。一定数の通知が集まったら Lambda 関数でまとめて処理し、管理者にメールや、データベースに記録する。

まとめ

AWS Lambda と Amazon SQS を組み合わせたイベントソースマッピングによるバッファリングは、大量のデータ処理を効率的に行うための優れた方法である。特定の条件を満たすと自動的に処理が開始されるため、スケーラビリティやコスト効率が高まる。リアルタイム性が求められないシナリオにおいて、この技術を活用することで、手作業を減らし、システム全体の効率を向上させることができる。

参考

備考

Hakky ではエンジニアを募集中です！まずは話してみたいなどでも構いませんので、ぜひお気軽に採用ページからお問い合わせくださいませ。