チュートリアル: DynamoDB テーブルの新しい項目の処理
トピック
このチュートリアルでは、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 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 実行ロールを作成する
このステップでは、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) を実行する。このチュートリアルの後半で、この関数を作成します。
-
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 メールアドレスに確認メッセージを送信します。そのメッセージの [Confirm subscription] リンクをクリックして、受信登録プロセスを完了します。
ステップ 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 の 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 nodejs4.3 -
ここで、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 コンソールを開きます。
-
ナビゲーションペインで [Logs] を選択します。
-
次のロググループを選択:
/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 コンソールを開きます。
-
ナビゲーションペインで [Logs] を選択します。
-
次のロググループを選択:
/aws/lambda/publishNewBark -
最新のログストリーミングを選択して、関数からの出力 (およびエラー) を表示します。
