将 Amazon Lambda 与 Amazon DynamoDB 结合使用 - Amazon Lambda
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 中国的 Amazon Web Services 服务入门 (PDF)

将 Amazon Lambda 与 Amazon DynamoDB 结合使用

注意

如果想要将数据发送到 Lambda 函数以外的目标,或要在发送数据之前丰富数据,请参阅 Amazon EventBridge Pipes(Amazon EventBridge 管道)。

您可以使用 Amazon Lambda 函数来处理 Amazon DynamoDB Streams中的记录。使用 DynamoDB Streams,每次更新 DynamoDB 表时,您都可以触发 Lambda 函数以执行其他工作。

Lambda 从流中读取记录并同步调用您的函数,带有一个包含流记录的事件。Lambda 分批读取记录并调用您的函数来处理批处理中的记录。

示例事件

{ "Records": [ { "eventID": "1", "eventVersion": "1.0", "dynamodb": { "Keys": { "Id": { "N": "101" } }, "NewImage": { "Message": { "S": "New item!" }, "Id": { "N": "101" } }, "StreamViewType": "NEW_AND_OLD_IMAGES", "SequenceNumber": "111", "SizeBytes": 26 }, "awsRegion": "us-west-2", "eventName": "INSERT", "eventSourceARN": "arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2024-06-10T19:26:16.525", "eventSource": "aws:dynamodb" }, { "eventID": "2", "eventVersion": "1.0", "dynamodb": { "OldImage": { "Message": { "S": "New item!" }, "Id": { "N": "101" } }, "SequenceNumber": "222", "Keys": { "Id": { "N": "101" } }, "SizeBytes": 59, "NewImage": { "Message": { "S": "This item has changed" }, "Id": { "N": "101" } }, "StreamViewType": "NEW_AND_OLD_IMAGES" }, "awsRegion": "us-west-2", "eventName": "MODIFY", "eventSourceARN": "arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2024-06-10T19:26:16.525", "eventSource": "aws:dynamodb" } ]}

轮询和批处理流

Lambda 以每秒 4 次的基本频率轮询 DynamoDB 流中的分区来获取记录。如果记录可用,Lambda 会调用函数并等待结果。如果处理成功,Lambda 会恢复轮询,直到其收到更多记录。

默认情况下,Lambda 会在记录可用时尽快调用您的函数。如果 Lambda 从事件源中读取的批处理只有一条记录,则 Lambda 将会只向该函数发送一条记录。为避免在记录数量较少的情况下调用该函数,您可以配置 batching window(批处理时段),让事件源缓冲最多五分钟的记录。调用函数前,Lambda 会持续从事件源中读取记录,直到收集完整批处理、批处理时段到期或批处理达到 6MB 的有效负载时为止。有关更多信息,请参阅 批处理行为

警告

Lambda 事件源映射至少处理每个事件一次,有可能出现重复处理记录的情况。为避免与重复事件相关的潜在问题,我们强烈建议您将函数代码设为幂等性。要了解更多信息,请参阅 Amazon 知识中心的如何使我的 Lambda 函数具有幂等性

配置 ParallelizationFactor 设置以同时使用多个 Lambda 调用处理 DynamoDB 流的一个分片。您可以指定 Lambda 通过从 1(默认值)到 10 的并行化因子从分区中轮询的并发批次数。增加每个分片的并发批次数后,Lambda 仍然可以确保项目(分区和排序键)级别的顺序处理。

轮询和流的起始位置

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

  • 在事件源映射创建期间,可能需要几分钟才能开始轮询来自流的事件。

  • 在事件源映射更新期间,可能需要几分钟才能停止和重新开始轮询来自流的事件。

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

DynamoDB Streams 中的分片同时读取器

对于作为非全局表的单区域表,您可以设计最多两个 Lambda 函数来同时从同一个 DynamoDB Streams 分片读取数据。超过此限制会导致请求被拒。对于全局表,我们建议您将并行函数的数量限制为一个,以避免请求节流。

执行角色权限

AWSLambdaDynamoDBExecutionRole Amazon 托管策略包括 Lambda 从您的 DynamoDB 流中读取所需的权限。将此托管策略添加到您的函数的执行角色。

要将失败批处理的记录发送到某个标准 SQS 队列或 SNS 主题,您的函数需要额外的权限。每项目标服务均需要不同的权限,如下所示:

添加权限并创建事件源映射

创建事件源映射以指示 Lambda 将流中的记录发送到 Lambda 函数。您可以创建多个事件源映射,以使用多个 Lambda 函数处理相同的数据,或使用单个函数处理来自多个流的项目。

要将函数配置为从 DynamoDB Streams 中读取,请将 AWSLambdaDynamoDBExecutionRole Amazon 托管策略附加到执行角色,然后创建 DynamoDB 触发器。

要添加权限并创建触发器
  1. 打开 Lamba 控制台的函数页面

  2. 选择一个函数的名称。

  3. 选择 Configuration(配置)选项卡,然后选择 Permissions(权限)。

  4. 角色名称下,选择至执行角色的链接。此角色将在 IAM 控制台中打开角色。

    至执行角色的链接
  5. 选择添加权限,然后选择附加策略

    在 IAM 控制台中附加策略
  6. 在搜索字段中输入 AWSLambdaDynamoDBExecutionRole。向执行角色添加此策略。这是一项 Amazon 托管策略,其中包含您的函数从 DynamoDB 流中读取所需的权限。有关此策略的更多信息,请参阅《Amazon Managed Policy Reference》中的 AWSLambdaDynamoDBExecutionRole

  7. 在 Lambda 控制台中返回您的函数。在 Function overview(函数概览)下,选择 Add trigger(添加触发器)。

    Lambda 控制台的函数概述部分
  8. 选择触发器类型。

  9. 配置必填选项,然后选择 Add(添加)。

Lambda 支持 DynamoDB 事件源的以下选项:

事件源选项
  • DynamoDB table(DynamoDB 表)– 要从中读取记录的 DynamoDB 表。

  • Batch size(批处理大小)– 每个批次中要发送到函数的记录数,最高为 10000。Lambda 通过单个调用将批处理中的所有记录传递给函数,前提是事件的总大小未超出同步调用的有效负载限制 (6 MB)。

  • Batch window(批处理时段)– 指定在调用函数之前收集记录的最长时间(以秒为单位)。

  • Starting position(开始位置)– 仅处理新记录或所有现有记录。

    • Latest(最新)– 处理已添加到流中的新记录。

    • Trim horizon(时间范围)– 处理流中的所有记录。

    在处理任何现有记录后,函数将继续处理新记录。

  • 失败时的目标 – 无法处理的记录的标准 SQS 队列或标准 SNS 主题。当 Lambda 因为某批记录太旧或已用尽所有重试而将其丢弃时,Lambda 会将有关该批处理的详细信息发送到该队列或主题。

  • Retry attempts(重试次数)– 函数返回错误时 Lambda 重试的最大次数。这不适用于批处理未到达函数的服务错误或限制。

  • Maximum age of record(记录的最长时限)– Lambda 发送到您的函数的记录的最长期限。

  • Split batch on error(出现错误时拆分批)– 当函数返回错误时,在重试之前将批次拆分为两批。原始批量大小设置会保持不变。

  • Concurrent batches per shard(每个分片的并发批处理数)– 同时处理来自同一个分片的多个批处理。

  • Enabled(已启用)– 设置为 true 可启用事件源映射。设置为 false 可停止处理记录。Lambda 会跟踪已处理的最后一条记录,并在重新启用映射后从停止位置重新开始处理。

注意

对于 Lambda 作为 DynamoDB 触发器的一部分而调用的 GetRecords API 调用,您无需付费。

之后,要管理事件源配置,请在设计器中选择触发器。

错误处理

DynamoDB 事件源映射的错误处理取决于错误是在调用函数之前还是在函数调用期间发生的:

  • 调用前:如果 Lambda 事件源映射由于节流或其他问题而无法调用该函数,则它会一直重试,直到记录过期或超过事件源映射上配置的最大期限(MaximumRecordAgeInSeconds)。

  • 调用期间:如果调用函数但返回错误,Lambda 会重试,直到记录过期、超过最大期限(MaximumRecordAgeInSeconds)或达到配置的重试配额(MaximumRetryAttempts)。对于函数错误,您还可以配置 BisectBatchOnFunctionError,将失败的批次拆分为两个较小的批次,从而隔离错误记录并避免超时。拆分批次不会消耗重试配额。

如果错误处理措施失败,Lambda 将丢弃记录并继续处理数据流中的批次。使用默认设置时,这意味着错误的记录可能会阻止受影响的分区上的处理,时间长达一天。为了避免这种情况,请配置函数的事件源映射,使用合理的重试次数和适合您的使用案例的最长记录期限。

配置失败调用的目标

要保留失败的事件源映射调用的记录,请在函数的事件源映射中添加一个目标。发送到目标的每条记录都是一个 JSON 文档,其中包含有关失败调用的元数据。您可以将任何 Amazon SNS 主题或 Amazon SQS 队列配置为目标。您的执行角色必须具有目标的权限:

要使用控制台配置失败时的目标,请执行以下步骤:

  1. 打开 Lamba 控制台的 Functions(函数)页面。

  2. 选择函数。

  3. Function overview (函数概览) 下,选择 Add destination (添加目标)

  4. 对于,请选择事件源映射调用

  5. 对于事件源映射,请选择为此函数配置的事件源。

  6. 条件中,选择失败时。对于事件源映射调用,这是唯一可接受的条件。

  7. 对于目标类型,请选择 Lambda 要发送调用记录的目标类型。

  8. 对于 Destination (目标),请选择一个资源。

  9. 选择保存

您还可以使用 Amazon Command Line Interface(Amazon CLI)配置失败时的目标。例如,以 create-event-source-mapping 命令将带有 SQS 失败时目标的事件源映射添加到 MyFunction

aws lambda create-event-source-mapping \ --function-name "MyFunction" \ --event-source-arn arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2024-06-10T19:26:16.525 \ --destination-config '{"OnFailure": {"Destination": "arn:aws:sqs:us-east-1:123456789012:dest-queue"}}'

以下 update-event-source-mapping 命令更新事件源映射,以在两次重试之后或记录超过一小时后将失败的调用记录发送到 SNS 目标。

aws lambda update-event-source-mapping \ --uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \ --maximum-retry-attempts 2 \ --maximum-record-age-in-seconds 3600 \ --destination-config '{"OnFailure": {"Destination": "arn:aws:sns:us-east-1:123456789012:dest-topic"}}'

更新的设置是异步应用的,并且直到该过程完成才反映在输出中。使用 get-event-source-mapping 命令查看当前状态。

要移除目标,请提供一个空字符串作为 destination-config 参数的实际参数:

aws lambda update-event-source-mapping \ --uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \ --destination-config '{"OnFailure": {"Destination": ""}}'

以下示例显示了 DynamoDB 流的调用记录。

例 调用记录
{ "requestContext": { "requestId": "316aa6d0-8154-xmpl-9af7-85d5f4a6bc81", "functionArn": "arn:aws:lambda:us-east-2:123456789012:function:myfunction", "condition": "RetryAttemptsExhausted", "approximateInvokeCount": 1 }, "responseContext": { "statusCode": 200, "executedVersion": "$LATEST", "functionError": "Unhandled" }, "version": "1.0", "timestamp": "2019-11-14T00:13:49.717Z", "DDBStreamBatchInfo": { "shardId": "shardId-00000001573689847184-864758bb", "startSequenceNumber": "800000000003126276362", "endSequenceNumber": "800000000003126276362", "approximateArrivalOfFirstRecord": "2019-11-14T00:13:19Z", "approximateArrivalOfLastRecord": "2019-11-14T00:13:19Z", "batchSize": 1, "streamArn": "arn:aws:dynamodb:us-east-2:123456789012:table/mytable/stream/2019-11-14T00:04:06.388" } }

您可以使用此信息从流中检索受影响的记录以进行故障排除。实际记录不包括在内,因此您必须处理此记录并在记录过期和丢失之前从流中检索它们。

Amazon CloudWatch 指标

在您的函数处理完一批记录后,Lambda 将发出 IteratorAge 指标。该指标指示处理完成时批处理中最后一条记录的时间。如果您的函数正在处理新事件,则可使用迭代器期限来估算新记录的添加时间与函数处理新记录的时间之间的延迟。

迭代器期限中的上升趋势可以指示您的函数问题。有关更多信息,请参阅 使用 Lambda 函数指标

时间窗口

Lambda 函数可以运行连续流处理应用程序。流表示通过您的应用程序持续流动的无边界数据。要分析这种不断更新的输入中的信息,可以使用按时间定义的窗口来限制包含的记录。

滚动窗口是定期打开和关闭的不同窗口。预设情况下,Lambda 调用是无状态的,在没有外部数据库的情况下,无法使用它们跨多次连续调用处理数据。但是,有了滚动窗口后,您可以在不同调用中保持状态。此状态包含之前为当前窗口处理的消息的汇总结果。您的状态最多可以是每个分片 1MB。如果超过该大小,Lambda 将提前终止窗口。

流中的每条记录都属于特定窗口。Lambda 将至少处理每条记录一次,但不保证每条记录只处理一次。在极少数情况下(例如错误处理),某些记录可能会被多次处理。第一次处理记录时始终按顺序处理。如果多次处理记录,则可能会不按顺序处理。

聚合和处理

系统将调用您的用户托管函数以便聚合和处理该聚合的最终结果。Lambda 汇总在该窗口中接收的所有记录。您可以分多个批次接收这些记录,每个批次都作为单独的调用。每次调用都会收到一个状态。因此,当使用滚动窗口时,Lambda 函数响应必须包含 state 属性。如果响应不包含 state 属性,Lambda 会将其视作失败的调用。为了满足该条件,您的函数可以返回一个具有以下 JSON 形状的 TimeWindowEventResponse 对象:

TimeWindowEventResponse
{ "state": { "1": 282, "2": 715 }, "batchItemFailures": [] }
注意

对于 Java 函数,我们建议使用 Map<String, String> 来表示状态。

在窗口末尾,标志 isFinalInvokeForWindow 被设置 true,以表示这是最终状态,并且已准备好进行处理。处理完成后,窗口完成,最终调用完成,然后状态将被删除。

在窗口结束时,Lambda 会对针对聚合结果的操作应用最终处理。您的最终处理将同步调用。成功调用后,函数会检查序列号并继续进行流处理。如果调用失败,则您的 Lambda 函数将暂停进一步处理,直到成功调用为止。

例 DynamodbTimeWindowEvent
{ "Records":[ { "eventID":"1", "eventName":"INSERT", "eventVersion":"1.0", "eventSource":"aws:dynamodb", "awsRegion":"us-east-1", "dynamodb":{ "Keys":{ "Id":{ "N":"101" } }, "NewImage":{ "Message":{ "S":"New item!" }, "Id":{ "N":"101" } }, "SequenceNumber":"111", "SizeBytes":26, "StreamViewType":"NEW_AND_OLD_IMAGES" }, "eventSourceARN":"stream-ARN" }, { "eventID":"2", "eventName":"MODIFY", "eventVersion":"1.0", "eventSource":"aws:dynamodb", "awsRegion":"us-east-1", "dynamodb":{ "Keys":{ "Id":{ "N":"101" } }, "NewImage":{ "Message":{ "S":"This item has changed" }, "Id":{ "N":"101" } }, "OldImage":{ "Message":{ "S":"New item!" }, "Id":{ "N":"101" } }, "SequenceNumber":"222", "SizeBytes":59, "StreamViewType":"NEW_AND_OLD_IMAGES" }, "eventSourceARN":"stream-ARN" }, { "eventID":"3", "eventName":"REMOVE", "eventVersion":"1.0", "eventSource":"aws:dynamodb", "awsRegion":"us-east-1", "dynamodb":{ "Keys":{ "Id":{ "N":"101" } }, "OldImage":{ "Message":{ "S":"This item has changed" }, "Id":{ "N":"101" } }, "SequenceNumber":"333", "SizeBytes":38, "StreamViewType":"NEW_AND_OLD_IMAGES" }, "eventSourceARN":"stream-ARN" } ], "window": { "start": "2020-07-30T17:00:00Z", "end": "2020-07-30T17:05:00Z" }, "state": { "1": "state1" }, "shardId": "shard123456789", "eventSourceARN": "stream-ARN", "isFinalInvokeForWindow": false, "isWindowTerminatedEarly": false }

配置

您可以在创建或更新事件源映射时配置滚动窗口。要配置翻转窗口,请以秒为单位进行指定(TumblingWindowInSeconds)。以下示例 Amazon Command Line Interface (Amazon CLI) 命令会创建一个滚动窗口为 120 秒的流式事件源映射。为聚合和处理定义的 Lambda 函数被命名为 tumbling-window-example-function

aws lambda create-event-source-mapping \ --event-source-arn arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2024-06-10T19:26:16.525 \ --function-name tumbling-window-example-function \ --starting-position TRIM_HORIZON \ --tumbling-window-in-seconds 120

Lambda 根据记录插入到流的时间来确定滚动窗口的边界。所有记录都有一个大致的时间戳,供 Lambda 在确定边界时使用。

滚动窗口聚合不支持重新分片。当分区结束时,Lambda 会认为窗口已关闭,子分区将以全新的状态启动自己的窗口。

滚动窗口完全支持现有的重试策略 maxRetryAttemptsmaxRecordAge

例 Handler.py – 聚合和处理

以下 Python 函数演示了如何聚合然后处理您的最终状态:

def lambda_handler(event, context): print('Incoming event: ', event) print('Incoming state: ', event['state']) #Check if this is the end of the window to either aggregate or process. if event['isFinalInvokeForWindow']: # logic to handle final state of the window print('Destination invoke') else: print('Aggregate invoke') #Check for early terminations if event['isWindowTerminatedEarly']: print('Window terminated early') #Aggregation logic state = event['state'] for record in event['Records']: state[record['dynamodb']['NewImage']['Id']] = state.get(record['dynamodb']['NewImage']['Id'], 0) + 1 print('Returning state: ', state) return {'state': state}

报告批处理项目失败

在使用和处理来自事件源的流式数据时,默认情况下,Lambda 仅在批处理完全成功时,才会在批次的最高序列号处设置检查点。Lambda 会将所有其他结果视为完全失败并重试批处理,直至达到重试次数上限。要允许在处理来自流的批次时部分成功,请开启 ReportBatchItemFailures。允许部分成功有助于减少对记录重试的次数,尽管这并不能完全阻止在成功记录中重试的可能性。

要开启 ReportBatchItemFailures,请在 FunctionResponseTypes 列表中包含枚举值 ReportBatchItemFailures。此列表指示为函数启用了哪些响应类型。您可以在创建更新事件源映射时配置此列表。

报告语法

配置批处理项目失败的报告时,将返回 StreamsEventResponse 类,其中包含批处理项目失败列表。您可以使用 StreamsEventResponse 对象返回批处理中第一个失败记录的序列号。您还可以使用正确的响应语法来创建自己的自定义类。以下 JSON 结构显示了所需的响应语法:

{ "batchItemFailures": [ { "itemIdentifier": "<SequenceNumber>" } ] }
注意

如果 batchItemFailures 数组包含多个项目,Lambda 会使用序列号最小的记录作为检查点。然后,Lambda 会重试从该检查点开始的所有记录。

成功和失败的条件

如果返回以下任意一项,则 Lambda 会将批处理视为完全成功:

  • 空的 batchItemFailure 列表

  • Null batchItemFailure 列表

  • 空的 EventResponse

  • Null EventResponse

如果返回以下任何一项,则 Lambda 会将批处理视为完全失败:

  • 空字符串 itemIdentifier

  • Null itemIdentifier

  • 包含错误密钥名的 itemIdentifier

Lambda 会根据您的重试策略在失败时重试。

将批次一分为二

如果调用失败并且已开启 BisectBatchOnFunctionError,则无论您的 ReportBatchItemFailures 设置如何,批次都将一分为二。

当收到批处理部分成功响应且同时开启 BisectBatchOnFunctionErrorReportBatchItemFailures 时,批次将在返回的序列号处一分为二,并且 Lambda 将仅重试剩余记录。

以下函数代码示例将返回批处理中处理失败消息的 ID 列表:

.NET
Amazon SDK for .NET
注意

在 GitHub 上查看更多内容。在无服务器示例存储库中查找完整示例,并了解如何进行设置和运行。

报告使用 .NET 通过 Lambda 进行 DynamoDB 批处理项目失败。

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 using System.Text.Json; using System.Text; using Amazon.Lambda.Core; using Amazon.Lambda.DynamoDBEvents; // Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class. [assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))] namespace AWSLambda_DDB; public class Function { public StreamsEventResponse FunctionHandler(DynamoDBEvent dynamoEvent, ILambdaContext context) { context.Logger.LogInformation($"Beginning to process {dynamoEvent.Records.Count} records..."); List<StreamsEventResponse.BatchItemFailure> batchItemFailures = new List<StreamsEventResponse.BatchItemFailure>(); StreamsEventResponse streamsEventResponse = new StreamsEventResponse(); foreach (var record in dynamoEvent.Records) { try { var sequenceNumber = record.Dynamodb.SequenceNumber; context.Logger.LogInformation(sequenceNumber); } catch (Exception ex) { context.Logger.LogError(ex.Message); batchItemFailures.Add(new StreamsEventResponse.BatchItemFailure() { ItemIdentifier = record.Dynamodb.SequenceNumber }); } } if (batchItemFailures.Count > 0) { streamsEventResponse.BatchItemFailures = batchItemFailures; } context.Logger.LogInformation("Stream processing complete."); return streamsEventResponse; } }
Go
适用于 Go V2 的 SDK
注意

在 GitHub 上查看更多内容。在无服务器示例存储库中查找完整示例,并了解如何进行设置和运行。

报告使用 Go 通过 Lambda 进行 DynamoDB 批处理项目失败。

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 package main import ( "context" "github.com/aws/aws-lambda-go/events" "github.com/aws/aws-lambda-go/lambda" ) type BatchItemFailure struct { ItemIdentifier string `json:"ItemIdentifier"` } type BatchResult struct { BatchItemFailures []BatchItemFailure `json:"BatchItemFailures"` } func HandleRequest(ctx context.Context, event events.DynamoDBEvent) (*BatchResult, error) { var batchItemFailures []BatchItemFailure curRecordSequenceNumber := "" for _, record := range event.Records { // Process your record curRecordSequenceNumber = record.Change.SequenceNumber } if curRecordSequenceNumber != "" { batchItemFailures = append(batchItemFailures, BatchItemFailure{ItemIdentifier: curRecordSequenceNumber}) } batchResult := BatchResult{ BatchItemFailures: batchItemFailures, } return &batchResult, nil } func main() { lambda.Start(HandleRequest) }
Java
SDK for Java 2.x
注意

在 GitHub 上查看更多内容。在无服务器示例存储库中查找完整示例,并了解如何进行设置和运行。

报告使用 Java 通过 Lambda 进行 DynamoDB 批处理项目失败。

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 import com.amazonaws.services.lambda.runtime.Context; import com.amazonaws.services.lambda.runtime.RequestHandler; import com.amazonaws.services.lambda.runtime.events.DynamodbEvent; import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse; import com.amazonaws.services.lambda.runtime.events.models.dynamodb.StreamRecord; import java.io.Serializable; import java.util.ArrayList; import java.util.List; public class ProcessDynamodbRecords implements RequestHandler<DynamodbEvent, Serializable> { @Override public StreamsEventResponse handleRequest(DynamodbEvent input, Context context) { List<StreamsEventResponse.BatchItemFailure> batchItemFailures = new ArrayList<>(); String curRecordSequenceNumber = ""; for (DynamodbEvent.DynamodbStreamRecord dynamodbStreamRecord : input.getRecords()) { try { //Process your record StreamRecord dynamodbRecord = dynamodbStreamRecord.getDynamodb(); curRecordSequenceNumber = dynamodbRecord.getSequenceNumber(); } catch (Exception e) { /* Since we are working with streams, we can return the failed item immediately. Lambda will immediately begin to retry processing from this failed item onwards. */ batchItemFailures.add(new StreamsEventResponse.BatchItemFailure(curRecordSequenceNumber)); return new StreamsEventResponse(batchItemFailures); } } return new StreamsEventResponse(); } }
JavaScript
适用于 JavaScript 的 SDK (v3)
注意

在 GitHub 上查看更多内容。在无服务器示例存储库中查找完整示例,并了解如何进行设置和运行。

使用 JavaScript 报告 Lambda 的 DynamoDB 批处理项目失败。

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 export const handler = async (event) => { const records = event.Records; let curRecordSequenceNumber = ""; for (const record of records) { try { // Process your record curRecordSequenceNumber = record.dynamodb.SequenceNumber; } catch (e) { // Return failed record's sequence number return { batchItemFailures: [{ itemIdentifier: curRecordSequenceNumber }] }; } } return { batchItemFailures: [] }; };

使用 Go 报告 Lambda 的 DynamoDB 批处理项目失败。

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 import { DynamoDBBatchItemFailure, DynamoDBStreamEvent } from "aws-lambda"; export const handler = async (event: DynamoDBStreamEvent): Promise<DynamoDBBatchItemFailure[]> => { const batchItemsFailures: DynamoDBBatchItemFailure[] = [] let curRecordSequenceNumber for(const record of event.Records) { curRecordSequenceNumber = record.dynamodb?.SequenceNumber if(curRecordSequenceNumber) { batchItemsFailures.push({ itemIdentifier: curRecordSequenceNumber }) } } return batchItemsFailures }
PHP
适用于 PHP 的 SDK
注意

在 GitHub 上查看更多内容。在无服务器示例存储库中查找完整示例,并了解如何进行设置和运行。

报告使用 PHP 通过 Lambda 进行 DynamoDB 批处理项目失败。

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 <?php # using bref/bref and bref/logger for simplicity use Bref\Context\Context; use Bref\Event\DynamoDb\DynamoDbEvent; use Bref\Event\Handler as StdHandler; use Bref\Logger\StderrLogger; require __DIR__ . '/vendor/autoload.php'; class Handler implements StdHandler { private StderrLogger $logger; public function __construct(StderrLogger $logger) { $this->logger = $logger; } /** * @throws JsonException * @throws \Bref\Event\InvalidLambdaEvent */ public function handle(mixed $event, Context $context): array { $dynamoDbEvent = new DynamoDbEvent($event); $this->logger->info("Processing records"); $records = $dynamoDbEvent->getRecords(); $failedRecords = []; foreach ($records as $record) { try { $data = $record->getData(); $this->logger->info(json_encode($data)); // TODO: Do interesting work based on the new data } catch (Exception $e) { $this->logger->error($e->getMessage()); // failed processing the record $failedRecords[] = $record->getSequenceNumber(); } } $totalRecords = count($records); $this->logger->info("Successfully processed $totalRecords records"); // change format for the response $failures = array_map( fn(string $sequenceNumber) => ['itemIdentifier' => $sequenceNumber], $failedRecords ); return [ 'batchItemFailures' => $failures ]; } } $logger = new StderrLogger(); return new Handler($logger);
Python
SDK for Python(Boto3)
注意

在 GitHub 上查看更多内容。在无服务器示例存储库中查找完整示例,并了解如何进行设置和运行。

报告使用 Python 通过 Lambda 进行 DynamoDB 批处理项目失败。

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 def handler(event, context): records = event.get("Records") curRecordSequenceNumber = "" for record in records: try: # Process your record curRecordSequenceNumber = record["dynamodb"]["SequenceNumber"] except Exception as e: # Return failed record's sequence number return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]} return {"batchItemFailures":[]}
Ruby
适用于 Ruby 的 SDK
注意

在 GitHub 上查看更多内容。在无服务器示例存储库中查找完整示例,并了解如何进行设置和运行。

报告使用 Ruby 通过 Lambda 进行 DynamoDB 批处理项目失败。

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 def lambda_handler(event:, context:) records = event["Records"] cur_record_sequence_number = "" records.each do |record| begin # Process your record cur_record_sequence_number = record["dynamodb"]["SequenceNumber"] rescue StandardError => e # Return failed record's sequence number return {"batchItemFailures" => [{"itemIdentifier" => cur_record_sequence_number}]} end end {"batchItemFailures" => []} end
Rust
适用于 Rust 的 SDK
注意

在 GitHub 上查看更多内容。在无服务器示例存储库中查找完整示例,并了解如何进行设置和运行。

报告使用 Rust 通过 Lambda 进行 DynamoDB 批处理项目失败。

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 use aws_lambda_events::{ event::dynamodb::{Event, EventRecord, StreamRecord}, streams::{DynamoDbBatchItemFailure, DynamoDbEventResponse}, }; use lambda_runtime::{run, service_fn, Error, LambdaEvent}; /// Process the stream record fn process_record(record: &EventRecord) -> Result<(), Error> { let stream_record: &StreamRecord = &record.change; // process your stream record here... tracing::info!("Data: {:?}", stream_record); Ok(()) } /// Main Lambda handler here... async fn function_handler(event: LambdaEvent<Event>) -> Result<DynamoDbEventResponse, Error> { let mut response = DynamoDbEventResponse { batch_item_failures: vec![], }; let records = &event.payload.records; if records.is_empty() { tracing::info!("No records found. Exiting."); return Ok(response); } for record in records { tracing::info!("EventId: {}", record.event_id); // Couldn't find a sequence number if record.change.sequence_number.is_none() { response.batch_item_failures.push(DynamoDbBatchItemFailure { item_identifier: Some("".to_string()), }); return Ok(response); } // Process your record here... if process_record(record).is_err() { response.batch_item_failures.push(DynamoDbBatchItemFailure { item_identifier: record.change.sequence_number.clone(), }); /* Since we are working with streams, we can return the failed item immediately. Lambda will immediately begin to retry processing from this failed item onwards. */ return Ok(response); } } tracing::info!("Successfully processed {} record(s)", records.len()); Ok(response) } #[tokio::main] async fn main() -> Result<(), Error> { tracing_subscriber::fmt() .with_max_level(tracing::Level::INFO) // disable printing the name of the module in every log line. .with_target(false) // disabling time is handy because CloudWatch will add the ingestion time. .without_time() .init(); run(service_fn(function_handler)).await }

Amazon DynamoDB Streams 配置参数

所有 Lambda 事件源类型共享相同的 CreateEventSourceMappingUpdateEventSourceMapping API 操作。但是,只有部分参数适用于 DynamoDB Streams。

适用于 DynamoDB Streams 的事件源参数
参数 必需 默认值 备注

BatchSize

100

最大值:10000

BisectBatchOnFunctionError

false

DestinationConfig

丢弃的记录的标准 Amazon SQS 队列或标准 Amazon SNS 主题目标。

启用

真实

EventSourceArn

Y

数据流或流使用者的 ARN

FilterCriteria

Lambda 事件筛选

FunctionName

FunctionResponseTypes

要使您的函数报告某个批处理中的特定失败,请在 FunctionResponseTypes 中包含值 ReportBatchItemFailures。有关更多信息,请参阅 报告批处理项目失败

MaximumBatchingWindowInSeconds

0

MaximumRecordAgeInSeconds

–1

-1 表示无限:会一直重试失败的记录,直到记录过期。DynamoDB Streams 的数据留存期限为 24 小时。

最小值:-1

最大值:604800

MaximumRetryAttempts

–1

-1 表示无限:会一直重试失败的记录,直到记录过期。

最小值:0

最大值:10000

ParallelizationFactor

1

最大值:10

StartingPosition

TRIM_HORIZON 或 LATEST(最新)

TumblingWindowInSeconds

最小值:0

最大值:900