AWS Lambdaのトリガーとイベント：S3・DynamoDB StreamsでLambda関数を動かす方法

AWS Lambdaのトリガーとは、Lambda関数を実行するきっかけとなる仕組みです。本記事では、イベント駆動アーキテクチャを支える「トリガー」と「イベント」の仕組みを解説し、S3とDynamoDB Streamsの2つのトリガーを組み合わせた実践例を紹介します（「Mastering AWS Lambda with Bao」シリーズ）。

はじめに

前回のエピソードでは、Amazon EventBridgeによってトリガーされるAWS Lambda関数を作成し、DynamoDBからデータを取得・処理してSQSキューへ送信する方法を解説しました。

今回は、AWS Lambdaのイベント駆動アーキテクチャの核となる「トリガーとイベント」をさらに深く掘り下げます。トリガーを使うと、LambdaはさまざまなAWSサービスで発生したアクションやイベントに応答できるようになり、完全に自動化されたスケーラブルなワークフローを構築できます。

このエピソードで学べること：

• Lambdaのトリガーとイベントの仕組み
• よく使われるAWS Lambdaトリガーの一覧
• 2つのトリガーを使った実例でLambdaの動作を確認する方法

前提条件

始める前に、次のものを準備してください。

• AWSアカウント：AWSリソースを作成・管理できる権限があること。
• Node.jsの基礎知識：JavaScriptとNode.jsに慣れていると、Lambda関数のコードを理解しやすくなります。

AWS Lambdaのトリガーとイベントを理解する

AWS Lambdaのトリガーとは？

AWS Lambdaにおけるトリガーとは、関数のコードを実行するきっかけとなるものです。トリガーは、さまざまなAWSサービスや外部アプリケーションから発生する「起動メカニズム」として機能します。トリガーが発火するとJSON形式のイベントがLambda関数に送られ、関数はそれを入力データとして処理します。

たとえば、S3バケットへのファイルのアップロード、Amazon EventBridgeで指定したスケジュール、API Gatewayが受け取るHTTPリクエストなどがトリガーになります。

• S3バケットへのファイルアップロードをきっかけに、Lambda関数がファイルを処理する。
• DynamoDBテーブルの変更をきっかけに、Lambdaが追加の計算や通知の送信を行う。

イベントの仕組み

トリガーが発火すると、何が起きたのかを表す構造化されたJSONドキュメント（イベント）が生成されます。Lambdaはこのイベントを入力として受け取り、関数を実行します。

S3トリガーから届くイベントの例（主要なフィールドのみ抜粋）：

1{
2  "Records": [
3    {
4      "eventVersion": "2.1",
5      "eventSource": "aws:s3",
6      "awsRegion": "ap-southeast-1",
7      "eventTime": "2024-12-16T08:56:51.798Z",
8      "eventName": "ObjectCreated:Put",
9      "s3": {
10        "bucket": {
11          "name": "upload-csv-lambda-st",
12          "arn": "arn:aws:s3:::upload-csv-lambda-st"
13        },
14        "object": {
15          "key": "random.csv",
16          "size": 38
17        }
18      }
19    }
20  ]
21}
Copy

イベントはRecords配列で届き、各レコードのeventSource（aws:s3など）を見ると発生元を判別できます。今回のコードでも、この値で処理を振り分けます。

DynamoDB Streamsから届くイベントは次のような形式です。追加されたアイテムはdynamodb.NewImageに、{ "S": "..." }のような型付きの形式で入っています。

1{
2  "Records": [
3    {
4      "eventID": "8f10a30f30bad0859bcddf48dda76bef",
5      "eventName": "INSERT",
6      "eventSource": "aws:dynamodb",
7      "awsRegion": "ap-southeast-1",
8      "dynamodb": {
9        "Keys": {
10          "FileName": { "S": "random.csv" },
11          "UploadTimestamp": { "S": "2024-12-16T08:56:53.286Z" }
12        },
13        "NewImage": {
14          "FileName": { "S": "random.csv" },
15          "UploadTimestamp": { "S": "2024-12-16T08:56:53.286Z" },
16          "Status": { "S": "Processed" }
17        },
18        "StreamViewType": "NEW_AND_OLD_IMAGES"
19      }
20    }
21  ]
22}
Copy

AWS Lambdaでよく使われるイベント駆動トリガー

代表的なトリガーは次のとおりです。

• Amazon S3：ファイルアップロードの処理。例：画像のリサイズ、メタデータの抽出、バケット間のファイル移動。
• Amazon DynamoDB Streams：テーブルのデータ変更への対応。例：更新の伝播、新規エントリの分析。
• Amazon API Gateway：REST APIやWebSocket APIの構築。例：ユーザー入力の処理、動的データの返却。
• Amazon EventBridge：アプリケーションやAWSサービスのイベントへの対応。例：スケジュールジョブ、カスタムイベント。
• Amazon SQS：メッセージの非同期処理。例：メッセージキューによるマイクロサービスの疎結合化。
• Amazon Kinesis：リアルタイムストリーミングデータの処理。例：ログやクリックストリームの分析。
• AWS IoT Core：IoTデバイスからのメッセージ処理。例：センサーデータの分析、デバイスの制御。

トリガーとイベントを活用すれば、複雑なワークフローもスムーズに自動化できます。対応しているサービスの一覧はAWS Lambda開発者ガイドを参照してください。

IAMロールの設定（任意）

Lambdaトリガーを設定する前に、必要な権限を持つIAMロールを作成します。関数の作成時に自動で作られるロールを使うこともできますが、その場合は後からS3・DynamoDB・SNSの権限を追加する必要があります。

IAMロールの作成

1. IAMコンソールでRolesを開き、Create roleをクリックします。

2. AWS service → Lambdaを選択し、Nextをクリックします。

3. 次のマネージドポリシーをアタッチします。

• AmazonS3ReadOnlyAccess：S3のファイル読み取り
• AmazonDynamoDBFullAccess：DynamoDBへの書き込みとDynamoDB Streamsの読み取り
• AmazonSNSFullAccess：SNSへの通知発行
• CloudWatchLogsFullAccess：実行ログの出力

4. Nextをクリックし、ロール名（例：LambdaTriggerRole）を入力します。

5. Create roleをクリックします。

ワークフローの設定

今回は、2つのトリガーを使ったシンプルなワークフローを作成します。

1. S3トリガー：アップロードされたファイルを処理し、メタデータをDynamoDBに保存する。
2. DynamoDB Streamsトリガー：新しいメタデータが追加されたら、SNSで通知を送る。

作成するリソースは次の4つです。すべて同じリージョンに作成してください（この記事ではap-southeast-1を使用）。画面のラベルは英語表示のコンソールに合わせています。

• S3バケット：upload-csv-lambda-st
• DynamoDBテーブル：DemoFileMetadata
• SNSトピック：DemoFileProcessingNotifications
• Lambda関数：DemoFileProcessing

ステップ1：S3バケットの作成

1. S3コンソールを開きます。

2. Create bucketをクリックし、次のように設定します。

• Bucket name：一意の名前（例：upload-csv-lambda-st）
• Region：任意のリージョン（この記事ではap-southeast-1）

3. Create bucketをクリックします。

ステップ2：DynamoDBテーブルの作成

1. DynamoDBコンソールを開きます。

2. Create tableをクリックし、次のように設定します。

• Table name：DemoFileMetadata
• Partition key：FileName（String）
• Sort key：UploadTimestamp（String）

3. Create tableをクリックします。

4. 作成したテーブルのExports and streamsタブを開き、DynamoDB stream detailsのTurn onをクリックします。

5. View typeでNew and old imagesを選び、Turn on streamをクリックします。

ステップ3：SNSトピックの作成

1. SNSコンソールのTopicsでCreate topicをクリックします。

2. 次のように設定します。

• Type：Standard
• Name：DemoFileProcessingNotifications

3. Create topicをクリックします。

4. 作成したトピックの画面でCreate subscriptionをクリックします。

5. Protocolで通知方法（この記事ではEmail）を選び、Endpointに受信先のメールアドレスを入力してCreate subscriptionをクリックします。

6. 届いた確認メールのリンクをクリックしてサブスクリプションを承認します（StatusがConfirmedになれば完了）。

7. トピックのARNを控えておきます。後でLambdaの環境変数に設定します。

ステップ4：Lambda関数の作成

1. LambdaコンソールでCreate functionをクリックします。

2. Author from scratchを選び、次のように設定します。

• Function name：DemoFileProcessing
• Runtime：Node.js 24.x（スクリーンショットは執筆時点のNode.js 20.x）
• Execution role：Use an existing roleを選び、先ほど作成したLambdaTriggerRoleを指定

3. Create functionをクリックします。

4. Configuration → Environment variablesで、次の環境変数を追加します。

• SNS_TOPIC_ARN：ステップ3で控えたトピックのARN
• TABLE_NAME：DemoFileMetadata

ステップ5：トリガーの設定

S3トリガーの追加

1. Function overviewでAdd triggerをクリックします。

2. S3を選択し、次のように設定します。

• Bucket：upload-csv-lambda-st
• Event types：All object create events
• Suffix：.csv（CSVファイルだけをトリガーの対象にする）
• Recursive invocation：注意事項を読み、チェックを入れる

3. Addをクリックします。

DynamoDB Streamsトリガーの追加

1. もう一度Add triggerをクリックします。

2. DynamoDBを選択し、DynamoDB tableでDemoFileMetadataを指定します。Activate triggerにチェックが入っていることを確認します。

3. Batch size（既定値100）とStarting position(Latest)を確認し、Addをクリックします。

Lambda関数のコード

S3とDynamoDB Streamsの両方のイベントを1つの関数で処理するコードです。関数のCodeタブでindex.mjsの内容を次のコードに置き換え、Deployをクリックします。

1import { DynamoDBClient, PutItemCommand } from "@aws-sdk/client-dynamodb";
2import { SNSClient, PublishCommand } from "@aws-sdk/client-sns";
3
4// クライアントはハンドラーの外で初期化し、ウォームスタート時に再利用する
5const dynamodb = new DynamoDBClient({});
6const sns = new SNSClient({});
7
8// 設定値は Lambda の環境変数から読み込む
9const TABLE_NAME = process.env.TABLE_NAME ?? "DemoFileMetadata";
10const SNS_TOPIC_ARN = process.env.SNS_TOPIC_ARN;
11
12export const handler = async (event) => {
13  console.log("Event received:", JSON.stringify(event, null, 2));
14
15  try {
16    // 1回の呼び出しに複数のレコードが含まれるため、1件ずつ発生元を判定する
17    for (const record of event.Records ?? []) {
18      switch (record.eventSource) {
19        case "aws:s3":
20          await handleS3Record(record);
21          break;
22        case "aws:dynamodb":
23          await handleDynamoDBRecord(record);
24          break;
25        default:
26          console.warn("Unsupported event source:", record.eventSource);
27      }
28    }
29
30    return {
31      statusCode: 200,
32      body: "Event processed successfully!",
33    };
34  } catch (error) {
35    console.error("Error processing event:", error);
36    throw error; // 再スローして失敗を Lambda に伝える
37  }
38};
39
40// S3 トリガー:アップロードされたファイルのメタデータを DynamoDB に保存する
41async function handleS3Record(record) {
42  const bucketName = record.s3.bucket.name;
43  // オブジェクトキーは URL エンコードされている(スペースは "+" になる)
44  const objectKey = decodeURIComponent(record.s3.object.key.replace(/\+/g, " "));
45  const uploadTimestamp = new Date().toISOString();
46
47  await dynamodb.send(
48    new PutItemCommand({
49      TableName: TABLE_NAME,
50      Item: {
51        FileName: { S: objectKey },
52        UploadTimestamp: { S: uploadTimestamp },
53        Status: { S: "Processed" },
54      },
55    })
56  );
57
58  console.log(`Metadata saved: s3://${bucketName}/${objectKey}`);
59}
60
61// DynamoDB Streams トリガー:新しく追加されたアイテムを SNS で通知する
62async function handleDynamoDBRecord(record) {
63  if (record.eventName !== "INSERT") return; // 更新・削除は対象外
64
65  const { FileName, UploadTimestamp } = record.dynamodb.NewImage;
66  const message = `File ${FileName.S} uploaded at ${UploadTimestamp.S} has been processed.`;
67
68  await sns.send(
69    new PublishCommand({
70      TopicArn: SNS_TOPIC_ARN,
71      Message: message,
72    })
73  );
74
75  console.log("Notification sent:", message);
76}
77
Copy

参考：元記事のソースコード（GitHub、SDK v2版）

コードの解説

1. AWS SDK v3のクライアントを初期化する

1import { DynamoDBClient, PutItemCommand } from "@aws-sdk/client-dynamodb";
2import { SNSClient, PublishCommand } from "@aws-sdk/client-sns";
3
4// クライアントはハンドラーの外で初期化し、ウォームスタート時に再利用する
5const dynamodb = new DynamoDBClient({});
6const sns = new SNSClient({});
Copy

• Node.js 18以降のランタイムにはAWS SDK for JavaScript v3が組み込まれているため、追加のインストールは不要です。
• v3は必要なサービスのクライアント（ここではDynamoDBとSNS）だけを読み込むモジュール構成です。
• クライアントをハンドラーの外で初期化すると、ウォームスタート時に接続が再利用され、処理が速くなります。

2. 設定値を環境変数から読み込む

1// 設定値は Lambda の環境変数から読み込む
2const TABLE_NAME = process.env.TABLE_NAME ?? "DemoFileMetadata";
3const SNS_TOPIC_ARN = process.env.SNS_TOPIC_ARN;
Copy

• SNSトピックのARNをコードに直接書かず、環境変数から読み込みます。
• 開発・本番など環境ごとの切り替えも、コードを変更せずに行えます。

3. レコードごとにイベントの発生元を判定する

1for (const record of event.Records ?? []) {
2  switch (record.eventSource) {
3    case "aws:s3":
4      await handleS3Record(record);
5      break;
6    case "aws:dynamodb":
7      await handleDynamoDBRecord(record);
8      break;
9    default:
10      console.warn("Unsupported event source:", record.eventSource);
11  }
12}
Copy

• event.Recordsには複数のレコードが含まれることがあります（DynamoDB Streamsのバッチサイズは既定で100）。
• 元記事のコードはRecords[0]だけを見て処理を分けていましたが、レコードごとにeventSourceを判定するほうが安全です。
• 想定外の発生元のイベントは、警告ログを出してスキップします。

4. S3のイベントを処理し、メタデータを保存する

1// S3 トリガー:アップロードされたファイルのメタデータを DynamoDB に保存する
2async function handleS3Record(record) {
3  const bucketName = record.s3.bucket.name;
4  // オブジェクトキーは URL エンコードされている(スペースは "+" になる)
5  const objectKey = decodeURIComponent(record.s3.object.key.replace(/\+/g, " "));
6  const uploadTimestamp = new Date().toISOString();
7
8  await dynamodb.send(
9    new PutItemCommand({
10      TableName: TABLE_NAME,
11      Item: {
12        FileName: { S: objectKey },
13        UploadTimestamp: { S: uploadTimestamp },
14        Status: { S: "Processed" },
15      },
16    })
17  );
18
19  console.log(`Metadata saved: s3://${bucketName}/${objectKey}`);
20}
Copy

• オブジェクトキーはURLエンコードされて届くため、decodeURIComponent()でデコードします（スペースは+になっているので先に置き換えます）。
• PutItemCommandでDemoFileMetadataテーブルにFileName・UploadTimestamp・Statusを書き込みます。
• 低レベルAPIのため、属性値は{ S: "..." }（文字列型）のように型を付けて指定します。

5. DynamoDB StreamsのイベントをSNSで通知する

1// DynamoDB Streams トリガー:新しく追加されたアイテムを SNS で通知する
2async function handleDynamoDBRecord(record) {
3  if (record.eventName !== "INSERT") return; // 更新・削除は対象外
4
5  const { FileName, UploadTimestamp } = record.dynamodb.NewImage;
6  const message = `File ${FileName.S} uploaded at ${UploadTimestamp.S} has been processed.`;
7
8  await sns.send(
9    new PublishCommand({
10      TopicArn: SNS_TOPIC_ARN,
11      Message: message,
12    })
13  );
14
15  console.log("Notification sent:", message);
16}
Copy

• INSERT（新規追加）のイベントだけを処理し、更新（MODIFY）や削除（REMOVE）は無視します。
• 追加されたアイテムの値はNewImageから取り出します。
• PublishCommandでSNSトピックにメッセージを発行すると、サブスクリプションに登録したメールアドレスに通知が届きます。

6. エラーハンドリング

1  } catch (error) {
2    console.error("Error processing event:", error);
3    throw error;
4  }
Copy

• エラーはCloudWatch Logsに記録したうえで再スローし、Lambdaに失敗を通知します。
• S3からの非同期呼び出しは自動で再試行され、DynamoDB Streamsでは同じバッチがレコードの有効期限まで再試行されます。
• 戻り値のstatusCodeとbodyは、この構成では使われません（API Gatewayなどから呼び出す場合に意味を持ちます）。

Lambda関数のテスト

1. コードを貼り付けてDeployをクリックします。

2. S3コンソールで、トリガーに設定したバケット（upload-csv-lambda-st）を開き、Uploadをクリックします。

3. 任意のCSVファイル（例：random.csv）を選択してアップロードします。

4. 結果を確認します。

DynamoDBテーブル：Explore itemsでDemoFileMetadataを開くと、StatusがProcessedのアイテムが追加されています。

SNS通知：登録したメールアドレスに、処理完了の通知が届きます。

CloudWatch Logs：ロググループ/aws/lambda/DemoFileProcessingに、S3とDynamoDB Streamsそれぞれから受け取ったイベントが記録されています。

これで、2つのトリガーで動作するLambda関数が完成しました。

まとめ

今回は、AWS Lambdaの基本となるトリガーとイベントについて解説しました。トリガーは、S3へのファイルアップロードやDynamoDBテーブルの変更といった特定のアクションにLambda関数を応答させる仕組みです。一方、イベントは、何がトリガーになったのかの詳細を含む、関数に渡される構造化データです。

実例では、1つのLambda関数で複数のトリガーを処理する方法を紹介しました。

• S3トリガーが、アップロードされたファイルのメタデータをDynamoDBに保存する。
• DynamoDB Streamsトリガーが、テーブルに追加されたメタデータをSNSで通知する。

この例から、Lambdaのイベント駆動アーキテクチャがいかに柔軟で、AWSサービスと連携してワークフローを自動化できるかがおわかりいただけたと思います。次回は、AWS Lambda関数のベストプラクティスとして、パフォーマンスの最適化、エラーハンドリング、セキュリティの強化を解説します。

出典：Triggers and Events: How AWS Lambda Connects with the World（SupremeTech、著者：Bao Dang）。日本語向けに再構成し、コードをAWS SDK v3に更新しています。