教程：将 Amazon Lambda 与 Amazon Kinesis 结合使用

在本教程中，您将创建 Lambda 函数来处理 Kinesis 流中的事件。

先决条件

本教程假设您对 Lambda 基本操作和 Lambda 控制台有一定了解。如果您还没有了解，请按照 使用控制台创建 Lambda 函数 中的说明创建您的第一个 Lambda 函数。

要完成以下步骤，您需要命令行终端或 Shell，以便运行命令。在单独的数据块中列出了命令和预期输出：

aws --version

您应看到以下输出：

aws-cli/2.0.57 Python/3.7.4 Darwin/19.6.0 exe/x86_64

对于长命令，使用转义字符 (\) 将命令拆分为多行。

在 Linux 和 macOS 中，可使用您首选的外壳程序和程序包管理器。在 Windows 10 中，您可以 安装 Windows Subsystem for Linux，获取 Ubuntu 和 Bash 与 Windows 集成的版本。

创建执行角色

创建执行角色，向您的函数授予访问 Amazon 资源的权限。

AWSLambdaKinesisExecutionRole 策略具有该函数从 Kinesis 中读取项目并将日志写入 CloudWatch Logs 所需的权限。

创建函数

以下示例代码接收 Kinesis 事件输入并对其所包含的消息进行处理。为了展示这个过程，代码会将一些传入的事件数据写入 CloudWatch Logs。

console.log('Loading function');

exports.handler = function(event, context) {
    //console.log(JSON.stringify(event, null, 2));
    event.Records.forEach(function(record) {
        // Kinesis data is base64 encoded so decode here
        var payload = Buffer.from(record.kinesis.data, 'base64').toString('ascii');
        console.log('Decoded payload:', payload);
    });
};

测试 Lambda 函数

使用 invoke Amazon Lambda CLI 命令和示例 Kinesis 事件手动调用 Lambda 函数。

创建 Kinesis 流

使用 create-stream 命令创建流。

aws kinesis create-stream --stream-name lambda-stream --shard-count 1

运行下面的 describe-stream 命令以获取流 ARN。

aws kinesis describe-stream --stream-name lambda-stream

您应看到以下输出：

{
    "StreamDescription": {
        "Shards": [
            {
                "ShardId": "shardId-000000000000",
                "HashKeyRange": {
                    "StartingHashKey": "0",
                    "EndingHashKey": "340282366920746074317682119384634633455"
                },
                "SequenceNumberRange": {
                    "StartingSequenceNumber": "49591073947768692513481539594623130411957558361251844610"
                }
            }
        ],
        "StreamARN": "arn:aws:kinesis:us-west-2:123456789012:stream/lambda-stream",
        "StreamName": "lambda-stream",
        "StreamStatus": "ACTIVE",
        "RetentionPeriodHours": 24,
        "EnhancedMonitoring": [
            {
                "ShardLevelMetrics": []
            }
        ],
        "EncryptionType": "NONE",
        "KeyId": null,
        "StreamCreationTimestamp": 1544828156.0
    }
}

您将使用下一步中的流 ARN 来将该流关联到您的 Lambda 函数。

在 Amazon Lambda 中添加事件源

运行以下 Amazon CLI add-event-source 命令。

aws lambda create-event-source-mapping --function-name ProcessKinesisRecords \
--event-source  arn:aws:kinesis:us-west-2:123456789012:stream/lambda-stream \
--batch-size 100 --starting-position LATEST

记下映射 ID 以供将来使用。您可以通过运行 list-event-source-mappings 命令获取事件源映射的列表。

aws lambda list-event-source-mappings --function-name ProcessKinesisRecords \
--event-source arn:aws:kinesis:us-west-2:123456789012:stream/lambda-stream

在该响应中，您可以验证状态值是否为 enabled。可以禁用事件源映射，以临时暂停轮询而不丢失任何记录。

测试设置

要测试事件源映射，请将事件记录添加到 Kinesis 流中。--data 值是一个字符串，CLI 先将其编码为 base64 字符串，然后才发送到 Kinesis。您可以多次运行同一命令来向流中添加多条记录。

aws kinesis put-record --stream-name lambda-stream --partition-key 1 \
--data "Hello, this is a test."

Lambda 使用执行角色来读取来自流的记录。然后它将调用 Lambda 函数，批量传递记录。该函数解码每条记录中的数据并记录它，将输出发送到 CloudWatch Logs 中。在 CloudWatch 控制台中查看这些日志。

清除资源

除非您想要保留为本教程创建的资源，否则可立即将其删除。通过删除不再使用的Amazon资源，可防止您的Amazon账户产生不必要的费用。