教程 1：对 Amazon DynamoDB 使用筛选器处理所有事件，以及对 AWS Lambda 使用 AWS CLI 进行处理

输入以下命令以创建表。

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
...

记录 region 和 accountID，因为您在本教程接下来的步骤中需要这些信息。

使用以下内容创建名为 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 的文件。（将 region 和 accountID 替换为您的 AWS 区域和帐户 ID。）

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "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": [
                "*"
            ]
        }
    ]
}                

策略有四个语句，允许 WooferLambdaRole 执行以下操作：

输入以下命令以将策略附加到 WooferLambdaRole。

aws iam put-role-policy --role-name WooferLambdaRole \
    --policy-name WooferLambdaRolePolicy \
    --policy-document file://role-policy.json

输入以下命令以创建新 Amazon SNS 主题。

aws sns create-topic --name wooferTopic

输入以下命令以使用电子邮件地址订阅 wooferTopic。（使用您的 AWS 区域和账户 ID 替换 region 和 accountID，并使用有效的电子邮件地址替换 example@example.com。）

aws sns subscribe \
    --topic-arn arn:aws:sns:region:accountID:wooferTopic \
    --protocol email \
    --notification-endpoint example@example.com

Amazon SNS 将向您的电子邮件地址发送确认邮件。选择该邮件中的确认订阅链接以完成订阅过程。

使用以下内容创建名为 publishNewBark.js 的文件。将 region 和 accountID 替换为您的 AWS 区域和帐户 ID。

'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 函数时，为 WooferLambdaRole 指定您在 第 2 步：创建一个 Lambda 执行角色 中创建的 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 region \
    --function-name publishNewBark \
    --zip-file fileb://publishNewBark.zip \
    --role roleARN \
    --handler publishNewBark.handler \
    --timeout 5 \
    --runtime nodejs16.x

现在测试 publishNewBark，验证它可以正常使用。为此，您将提供类似于来自 DynamoDB Streams 的真实记录的输入。

使用以下内容创建名为 payload.json 的文件。将 region 和 accountID 替换为您的 AWS 区域和账户 ID。

{
    "Records": [
        {
            "eventID": "7de3041dd709b024af6f29e4fa13d34c",
            "eventName": "INSERT",
            "eventVersion": "1.1",
            "eventSource": "aws:dynamodb",
            "awsRegion": "region",
            "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:region:account ID:table/BarkTable/stream/2016-11-16T20:42:48.104"
        }
    ]
}

输入以下命令以测试 publishNewBark 函数。

aws lambda invoke --function-name publishNewBark --payload file://payload.json --cli-binary-format raw-in-base64-out output.txt

如果测试成功，您将看到以下输出。

{
    "StatusCode": 200,
    "ExecutedVersion": "$LATEST"
}

此外，output.txt 文件将包含以下文本。

"Successfully processed 1 records."

您还会在数分钟内收到一封新电子邮件。

在创建触发器时，您需要为 BarkTable 流指定 ARN。输入以下命令检索此 ARN。

aws dynamodb describe-table --table-name BarkTable

在输出中，查找 LatestStreamArn。

...
 "LatestStreamArn": "arn:aws:dynamodb:region:accountID:table/BarkTable/stream/timestamp
...

输入以下命令以创建触发器。使用实际流 ARN 替换 streamARN。

aws lambda create-event-source-mapping \
    --region region \
    --function-name publishNewBark \
    --event-source streamARN  \
    --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"}

您应在数分钟内收到一封新电子邮件。

打开 DynamoDB 控制台并再将几个项目添加到 BarkTable。您必须为 Username 和 Timestamp 属性指定值。（您还应为 Message 指定值，虽然该值并非必需。） 对于添加到 BarkTable 中的每个项目，您应收到一封新电子邮件。

Lambda 函数仅处理您添加到 BarkTable 的新项目。如果您在表中更新或删除项目，函数不执行任何操作。