チュートリアル: DynamoDB ストリーム および Lambdaを使用して新しい項目を処理する
トピック
このチュートリアルでは、AWS Lambda トリガーを作成して、DynamoDB テーブルからのストリームを処理します。
このチュートリアルのシナリオは、シンプルなソーシャルネットワークである Woofer です。Woofer ユーザーは、他の Woofer ユーザーに送信される bark (短いテキストメッセージ) を使用して通信します。次の図は、このアプリケーションのコンポーネントとワークフローを示しています。
-
ユーザーは DynamoDB テーブル (
BarkTable) に項目を書き込みます。テーブルの各項目は bark を表します。 -
新しいストリームレコードが書き込まれ、新しい項目が
BarkTableに追加されたことを反映します。 -
新しいストリームレコードは AWS Lambda 関数 (
publishNewBark) をトリガーします。 -
ストリームレコードに、新しい項目が
BarkTableに追加されたことが示された場合、Lambda 関数はストリームレコードからデータを読み取り、Amazon Simple Notification Service (Amazon SNS) のトピックにメッセージを発行します。 -
メッセージは Amazon SNS トピックの受信者によって受信されます(このチュートリアルでは、唯一の受信者は E メールアドレスです)。
開始する前に
このチュートリアルでは AWS Command Line Interface AWS CLI を使用します。まだ行っていない場合は、「AWS Command Line Interface ユーザーガイド」の手順に従って、AWS CLI をインストールおよび設定します。
ステップ 1: ストリームが有効になった DynamoDB テーブルを作成する
このステップでは、Woofer ユーザーからのすべての bark を保存する DynamoDB テーブル (BarkTable) を作成します。プライマリキーは、Username (パーティションキー) と Timestamp (ソートキー) で構成されます。これらの属性は両方とも文字列型になります。
BarkTable ではストリームが有効になっています。このチュートリアルの後半では、AWS Lambda 関数をストリームと関連付けてトリガーを作成します。
-
次のコマンドを入力して、テーブルを作成します。
aws dynamodb create-table \ --table-name BarkTable \ --attribute-definitions AttributeName=Username,AttributeType=S AttributeName=Timestamp,AttributeType=S \ --key-schema AttributeName=Username,KeyType=HASH AttributeName=Timestamp,KeyType=RANGE \ --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 \ --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES -
出力で、
LatestStreamArnを探します。... "LatestStreamArn": "arn:aws:dynamodb:region:accountID:table/BarkTable/stream/timestamp ...regionaccountID
ステップ 2: Lambda 実行ロールを作成する
このステップでは、AWS Identity and Access Management (IAM) ロール (WooferLambdaRole) を作成し、それにアクセス権限を割り当てます。このロールは、「ステップ 4: Lambda 関数を作成してテストする」で作成する Lambda 関数で使用されます。
また、ロールのポリシーを作成します。このポリシーには、Lambda 関数が実行時に必要とするすべてのアクセス許可が含まれます。
-
次の内容で、
trust-relationship.jsonというファイルを作成します。{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "lambda.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } -
WooferLambdaRoleを作成するため、以下のコマンドを入力します。aws iam create-role --role-name WooferLambdaRole \ --path "/service-role/" \ --assume-role-policy-document file://trust-relationship.json -
次の内容で、
role-policy.jsonというファイルを作成します。(regionaccountID{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "lambda:InvokeFunction", "Resource": "arn:aws:lambda:region:accountID:function:publishNewBark*" }, { "Effect": "Allow", "Action": [ "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents" ], "Resource": "arn:aws:logs:region:accountID:*" }, { "Effect": "Allow", "Action": [ "dynamodb:DescribeStream", "dynamodb:GetRecords", "dynamodb:GetShardIterator", "dynamodb:ListStreams" ], "Resource": "arn:aws:dynamodb:region:accountID:table/BarkTable/stream/*" }, { "Effect": "Allow", "Action": [ "sns:Publish" ], "Resource": [ "*" ] } ] }ポリシーには 4 つのステートメントがあり、これにより
WooferLambdaRoleは以下を実行できます。-
Lambda 関数 (
publishNewBark) を実行します。このチュートリアルの後半で、この関数を作成します。 -
Amazon CloudWatch Logs にアクセスします。Lambda 関数はランタイム時に診断を CloudWatch Logs に書き込みます。
-
BarkTableの DynamoDB ストリームからデータを読み取ります。 -
Amazon SNS にメッセージを発行する。
-
-
次のコマンドを入力して、
WooferLambdaRoleにポリシーをアタッチします。aws iam put-role-policy --role-name WooferLambdaRole \ --policy-name WooferLambdaRolePolicy \ --policy-document file://role-policy.json
ステップ 3: Amazon SNS トピックを作成する
このステップでは、Amazon SNS トピック (wooferTopic) を作成し、そのトピックに E メールアドレスをサブスクライブします。Lambda 関数はこのトピックを使用して、Woofer ユーザーからの新しい bark を公開します。
-
次のコマンドを入力して、新しい Amazon SNS トピックを作成します。
aws sns create-topic --name wooferTopic -
次のコマンドを入力して、
wooferTopicに E メールアドレスをサブスクライブします(regionaccountIDexample@example.comaws sns subscribe \ --topic-arn arn:aws:sns:region:accountID:wooferTopic \ --protocol email \ --notification-endpointexample@example.com -
Amazon SNS は E メールアドレスに確認メッセージを送信します。そのメッセージの [サブスクリプションを確認] リンクを選択して、サブスクリプションプロセスを完了します。
ステップ 4: Lambda 関数を作成してテストする
このステップでは、AWS Lambda 関数 (publishNewBark) を作成して BarkTable からのストリームレコードを処理します。
publishNewBark 関数は、BarkTable の新しい項目に対応するストリームイベントのみを処理します。この関数は、そのようなイベントからデータを読み取ってから、Amazon SNS を呼び出してデータを公開します。
-
次の内容で、
publishNewBark.jsというファイルを作成します。regionaccountID'use strict'; var AWS = require("aws-sdk"); var sns = new AWS.SNS(); exports.handler = (event, context, callback) => { event.Records.forEach((record) => { console.log('Stream record: ', JSON.stringify(record, null, 2)); if (record.eventName == 'INSERT') { var who = JSON.stringify(record.dynamodb.NewImage.Username.S); var when = JSON.stringify(record.dynamodb.NewImage.Timestamp.S); var what = JSON.stringify(record.dynamodb.NewImage.Message.S); var params = { Subject: 'A new bark from ' + who, Message: 'Woofer user ' + who + ' barked the following at ' + when + ':\n\n ' + what, TopicArn: 'arn:aws:sns:region:accountID:wooferTopic' }; sns.publish(params, function(err, data) { if (err) { console.error("Unable to send message. Error JSON:", JSON.stringify(err, null, 2)); } else { console.log("Results from sending message: ", JSON.stringify(data, null, 2)); } }); } }); callback(null, `Successfully processed ${event.Records.length} records.`); }; -
publishNewBark.jsを含める zip ファイルを作成します。zip コマンドラインユーティリティがある場合は、次のコマンドを入力してこれを行うことができます。zip publishNewBark.zip publishNewBark.js -
Lambda 関数を作成する場合は、「ステップ 2: Lambda 実行ロールを作成する」で作成した
WooferLambdaRoleの Amazon リソースネーム (ARN) を指定します。この ARN を取得するには、次のコマンドを入力します。aws iam get-role --role-name WooferLambdaRole出力で、
WooferLambdaRoleの ARN を探します。... "Arn": "arn:aws:iam::region:role/service-role/WooferLambdaRole" ...次のコマンドを入力して、Lambda 関数を作成します。
roleARNをWooferLambdaRoleの ARN に置き換えます。aws lambda create-function \ --region us-east-1 \ --function-name publishNewBark \ --zip-file fileb://publishNewBark.zip \ --roleroleARN\ --handler publishNewBark.handler \ --timeout 5 \ --runtime nodejs10.x -
ここで、
publishNewBarkをテストして、これが動作することを確認します。これを行うには、DynamoDB ストリーム の実際のレコードに似た情報を入力します。次の内容で、
payload.jsonというファイルを作成します。{ "Records": [ { "eventID": "7de3041dd709b024af6f29e4fa13d34c", "eventName": "INSERT", "eventVersion": "1.1", "eventSource": "aws:dynamodb", "awsRegion": "us-west-2", "dynamodb": { "ApproximateCreationDateTime": 1479499740, "Keys": { "Timestamp": { "S": "2016-11-18:12:09:36" }, "Username": { "S": "John Doe" } }, "NewImage": { "Timestamp": { "S": "2016-11-18:12:09:36" }, "Message": { "S": "This is a bark from the Woofer social network" }, "Username": { "S": "John Doe" } }, "SequenceNumber": "13021600000000001596893679", "SizeBytes": 112, "StreamViewType": "NEW_IMAGE" }, "eventSourceARN": "arn:aws:dynamodb:us-east-1:123456789012:table/BarkTable/stream/2016-11-16T20:42:48.104" } ] }次のコマンドを入力して、
publishNewBark関数をテストします。aws lambda invoke --function-name publishNewBark --payload file://payload.json output.txtテストが成功すると、次の出力が表示されます。
{ "StatusCode": 200 }さらに、
output.txtファイルには次のテキストが含まれます。"Successfully processed 1 records."また、数分以内に新しい E メールメッセージが届きます。
注記 AWS Lambda は診断情報を Amazon CloudWatch Logs に書き込みます。Lambda 関数でエラーが発生した場合、この診断情報をトラブルシューティングに使用できます。
-
https://console.aws.amazon.com/cloudwatch/
にある CloudWatch コンソールを開きます。 -
ナビゲーションペインで [ログ] を選択します。
-
次のロググループを選択:
/aws/lambda/publishNewBark -
最新のログストリーミングを選択して、関数からの出力 (およびエラー) を表示します。
-
ステップ 5: トリガーを作成してテストする
「ステップ 4: Lambda 関数を作成してテストする」で、Lambda 関数をテストして、正しく実行されたことを確認しました。このステップでは、Lambda 関数 (publishNewBark) をイベントソース (BarkTable ストリーム) に関連付けて、トリガーを作成します。
-
トリガーを作成する場合、
BarkTableストリーム用の ARN を指定する必要があります。この ARN を取得するには、次のコマンドを入力します。aws dynamodb describe-table --table-name BarkTable出力で、
LatestStreamArnを探します。... "LatestStreamArn": "arn:aws:dynamodb:region:accountID:table/BarkTable/stream/timestamp ... -
次のコマンドを入力してトリガーを作成します
streamARNaws lambda create-event-source-mapping \ --region us-east-1 \ --function-name publishNewBark \ --event-sourcestreamARN\ --batch-size 1 \ --starting-position TRIM_HORIZON -
トリガーをテストします。次のコマンドを入力して、
BarkTableに項目を追加します。aws dynamodb put-item \ --table-name BarkTable \ --item Username={S="Jane Doe"},Timestamp={S="2016-11-18:14:32:17"},Message={S="Testing...1...2...3"}数分以内に新しい E メールメッセージが届きます。
-
DynamoDB コンソールを開き、さらにいくつかの項目を
BarkTableに追加します。UsernameおよびTimestamp属性の値を指定する必要があります(必須ではないものの、Messageの値を指定する必要があります)。BarkTableに追加した各項目について、新しい E メールメッセージが届きます。Lambda 関数は、
BarkTableに追加した新しい項目のみを処理します。テーブル内の項目を更新または削除すると、この関数は何も行いません。
AWS Lambda は診断情報を Amazon CloudWatch Logs に書き込みます。Lambda 関数でエラーが発生した場合、この診断情報をトラブルシューティングに使用できます。
-
https://console.aws.amazon.com/cloudwatch/
にある CloudWatch コンソールを開きます。 -
ナビゲーションペインで [ログ] を選択します。
-
次のロググループを選択:
/aws/lambda/publishNewBark -
最新のログストリーミングを選択して、関数からの出力 (およびエラー) を表示します。
