使用 Lambda 处理 Amazon DocumentDB 事件

您可以通过将 Amazon DocumentDB 集群配置为事件源，从而使用 Lambda 函数来处理 Amazon DocumentDB（与 MongoDB 兼容）更改流中的事件。然后，您可以在 Amazon DocumentDB 集群每次发生数据更改时调用 Lambda 函数，从而实现事件驱动型工作负载的自动化。

Lambda 按照事件到达的顺序依次处理 Amazon DocumentDB 更改流中的事件。因此，您的函数一次只能处理一条来自 Amazon DocumentDB 的并发调用。要监控函数，您可以跟踪其并发指标。

Amazon DocumentDB 事件示例

{
    "eventSourceArn": "arn:aws:rds:us-east-1:123456789012:cluster:canaryclusterb2a659a2-qo5tcmqkcl03",
    "events": [
        {
            "event": {
                "_id": {
                    "_data": "0163eeb6e7000000090100000009000041e1"
                },
                "clusterTime": {
                    "$timestamp": {
                        "t": 1676588775,
                        "i": 9
                    }
                },
                "documentKey": {
                    "_id": {
                        "$oid": "63eeb6e7d418cd98afb1c1d7"
                    }
                },
                "fullDocument": {
                    "_id": {
                        "$oid": "63eeb6e7d418cd98afb1c1d7"
                    },
                    "anyField": "sampleValue"
                },
                "ns": {
                    "db": "test_database",
                    "coll": "test_collection"
                },
                "operationType": "insert"
            }
        }
    ],
    "eventSource": "aws:docdb"
}

有关此示例中的事件及其形状的更多信息，请参阅 MongoDB 文档网站上的更改事件。

先决条件和权限

将 Amazon DocumentDB 用作 Lambda 函数的事件源之前，请注意以下先决条件。您必须：

配置网络安全

要通过事件源映射向 Lambda 提供对 Amazon DocumentDB 的完全访问权限，集群必须使用公有端点（公有 IP 地址），或者您必须提供对您在其中创建了集群的 Amazon VPC 的访问权限。

将 Amazon DocumentDB 与 Lambda 结合使用时，创建 Amazon PrivateLink VPC 端点，以便您的函数可以访问 Amazon VPC 中的资源。

创建端点以提供对以下资源的访问权限：

也可以在 Amazon VPC 中的每个公有子网上配置 NAT 网关。有关更多信息，请参阅 为连接到 VPC 的 Lambda 函数启用互联网访问权限。

为 Amazon DocumentDB 创建事件源映射时，Lambda 会检查为 Amazon VPC 配置的子网和安全组是否已经存在弹性网络接口（ENI）。如果 Lambda 发现现有 ENI，则会尝试重用这些 ENI。否则，Lambda 会创建新的 ENI 来连接到事件源并调用函数。

为包含集群的 Amazon VPC 配置安全组。默认情况下，Amazon DocumentDB 使用以下端口：27017。

如果集群使用身份验证，您还可以限制 Secrets Manager 端点的端点策略。要调用 Secrets Manager API，Lambda 会使用函数角色而非 Lambda 服务主体。

{
      "Statement": [
          {
              "Action": "secretsmanager:GetSecretValue",
              "Effect": "Allow",
              "Principal": {
                  "AWS": [
                      "arn:aws::iam::123456789012:role/my-role"
                  ]
              },
              "Resource": "arn:aws::secretsmanager:us-west-2:123456789012:secret:my-secret"
          }
      ]
  }

当您使用 Amazon VPC 端点时，Amazon 会使用端点的弹性网络接口（ENI）路由 API 调用来调用函数。Lambda 服务主体需要针对使用这些 ENI 的任何角色和函数调用 lambda:InvokeFunction。

默认情况下，Amazon VPC 端点具有开放的 IAM 策略，允许对资源进行广泛访问。最佳实践是，将这些策略限制为使用该端点执行所需的操作。为确保事件源映射能够调用 Lambda 函数，VPC 端点策略必须允许 Lambda 服务主体调用 sts:AssumeRole 和 lambda:InvokeFunction。将 VPC 端点策略限制为仅允许来自组织内部的 API 调用，会导致事件源映射无法正常运行，因此这些策略中需要 "Resource": "*"。

以下 VPC 端点策略示例展示了如何向 Amazon STS 的 Lambda 服务主体和 Lambda 端点授予所需的访问权限。

{
      "Statement": [
          {
              "Action": "sts:AssumeRole",
              "Effect": "Allow",
              "Principal": {
                  "Service": [
                      "lambda.amazonaws.com"
                  ]
              },
              "Resource": "*"
          }
      ]
    }

{
      "Statement": [
          {
              "Action": "lambda:InvokeFunction",
              "Effect": "Allow",
              "Principal": {
                  "Service": [
                      "lambda.amazonaws.com"
                  ]
              },
              "Resource": "*"
          }
      ]
  }

创建 Amazon DocumentDB 事件源映射（控制台）

要配置 Lambda 函数以从 Amazon DocumentDB 集群更改流中读取数据，请创建一个 事件源映射。这一部分介绍了如何从 Lambda 控制台执行此操作。有关 Amazon SDK 和 Amazon CLI 说明，请参阅 创建 Amazon DocumentDB 事件源映射（SDK 或 CLI）。

Lambda 支持 Amazon DocumentDB 事件源的以下选项：

创建 Amazon DocumentDB 事件源映射（SDK 或 CLI）

要使用 Amazon SDK 来管理 Amazon DocumentDB 事件源映射，您可以使用以下 API 操作：

要使用 Amazon CLI 创建事件源映射，请使用 create-event-source-mapping 命令。以下示例使用此命令将名为 my-function 的函数映射到一个 Amazon DocumentDB 更改流。事件源由 Amazon 资源名称（ARN）指定，批处理大小为 500，从以 Unix 时间表示的时间戳开始。该命令还指定了 Lambda 用来连接到 Amazon DocumentDB 的 Secrets Manager 密钥。此外，它还包含用于指定要从中读取数据的数据库和集合的 document-db-event-source-config 参数。

aws lambda create-event-source-mapping --function-name my-function \
    --event-source-arn arn:aws:rds:us-west-2:123456789012:cluster:privatecluster7de2-epzcyvu4pjoy
    --batch-size 500 \
    --starting-position AT_TIMESTAMP \
    --starting-position-timestamp 1541139109 \
    --source-access-configurations '[{"Type":"BASIC_AUTH","URI":"arn:aws:secretsmanager:us-east-1:123456789012:secret:DocDBSecret-BAtjxi"}]' \
    --document-db-event-source-config '{"DatabaseName":"test_database", "CollectionName": "test_collection"}' \

应看到类似如下内容的输出：

{
    "UUID": "2b733gdc-8ac3-cdf5-af3a-1827b3b11284",
    "BatchSize": 500,
    "DocumentDBEventSourceConfig": {
        "CollectionName": "test_collection",
        "DatabaseName": "test_database",
        "FullDocument": "Default"
    },
    "MaximumBatchingWindowInSeconds": 0,
    "EventSourceArn": "arn:aws:rds:us-west-2:123456789012:cluster:privatecluster7de2-epzcyvu4pjoy",
    "FunctionArn": "arn:aws:lambda:us-west-2:123456789012:function:my-function",
    "LastModified": 1541348195.412,
    "LastProcessingResult": "No records processed",
    "State": "Creating",
    "StateTransitionReason": "User action"
}

创建后，您可以使用 update-event-source-mapping 命令更新 Amazon DocumentDB 事件源的设置。以下示例会将批处理大小更新为 1000，并将批处理时长更新为 10 秒。对于此命令，您需要有事件源映射的 UUID，您可以通过 list-event-source-mapping 命令或 Lambda 控制台检索该值。

aws lambda update-event-source-mapping --function-name my-function \
    --uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \
    --batch-size 1000 \
    --batch-window 10

您应看到如下输出：

{
    "UUID": "2b733gdc-8ac3-cdf5-af3a-1827b3b11284",
    "BatchSize": 500,
    "DocumentDBEventSourceConfig": {
        "CollectionName": "test_collection",
        "DatabaseName": "test_database",
        "FullDocument": "Default"
    },
    "MaximumBatchingWindowInSeconds": 0,
    "EventSourceArn": "arn:aws:rds:us-west-2:123456789012:cluster:privatecluster7de2-epzcyvu4pjoy",
    "FunctionArn": "arn:aws:lambda:us-west-2:123456789012:function:my-function",
    "LastModified": 1541359182.919,
    "LastProcessingResult": "OK",
    "State": "Updating",
    "StateTransitionReason": "User action"
}

Lambda 会异步更新设置，因此在更新完成之前，您可能无法在输出中看到这些更改。要查看事件源映射的当前设置，请使用 get-event-source-mapping 命令。

aws lambda get-event-source-mapping --uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b

您应看到如下输出：

{
    "UUID": "2b733gdc-8ac3-cdf5-af3a-1827b3b11284",
    "DocumentDBEventSourceConfig": {
        "CollectionName": "test_collection",
        "DatabaseName": "test_database",
        "FullDocument": "Default"
    },
    "BatchSize": 1000,
    "MaximumBatchingWindowInSeconds": 10,
    "EventSourceArn": "arn:aws:rds:us-west-2:123456789012:cluster:privatecluster7de2-epzcyvu4pjoy",
    "FunctionArn": "arn:aws:lambda:us-west-2:123456789012:function:my-function",
    "LastModified": 1541359182.919,
    "LastProcessingResult": "OK",
    "State": "Enabled",
    "StateTransitionReason": "User action"
}

要删除 Amazon DocumentDB 事件源映射，请使用 delete-event-source-mapping 命令：

aws lambda delete-event-source-mapping \
    --uuid 2b733gdc-8ac3-cdf5-af3a-1827b3b11284

轮询和流的起始位置

请注意，事件源映射创建和更新期间的流轮询最终是一致的。

此行为意味着，如果你指定 LATEST 作为流的起始位置，事件源映射可能会在创建或更新期间错过事件。为确保不会错过任何事件，请将流的起始位置指定为 TRIM_HORIZON 或 AT_TIMESTAMP。

监控 Amazon DocumentDB 事件源

为便于您监控 Amazon DocumentDB 事件源，在函数处理完一批记录后，Lambda 将发送 IteratorAge 指标。迭代器期限是最近事件的时间戳和当前时间戳之间的差值。基本上，IteratorAge 指标表示自处理该批处理中最后一条记录后已经过的时间。如果函数当前正在处理新事件，则可使用迭代器期限来估算新记录的添加时间与函数处理该记录的时间之间的延迟。如果 IteratorAge 呈上升趋势，则可能说明您的函数存在问题。有关更多信息，请参阅 将 CloudWatch 指标与 Lambda 结合使用。

Amazon DocumentDB 更改流未经过优化，无法处理事件之间的较大时间间隔。如果 Amazon DocumentDB 事件源在很长一段时间内都没有收到任何事件，那么 Lambda 可能会禁用事件源映射。此时段的长度可能从几周到几个月不等，具体取决于集群规模和其他工作负载。

Lambda 支持最大 6MB 的负载。但是，Amazon DocumentDB 更改流事件的大小可达 16MB。如果更改流尝试向 Lambda 发送大于 6MB 的更改流事件，Lambda 会丢弃该消息并发送 OversizedRecordCount 指标。Lambda 将尽力发送所有指标。