AWS Lambda 與 Amazon DynamoDB 使用 - AWS Lambda

本文為英文版的機器翻譯版本,如內容有任何歧義或不一致之處,概以英文版為準。

AWS Lambda 與 Amazon DynamoDB 使用

注意

如果您想要將資料傳送到 Lambda 函數以外的目標,或在傳送資料之前豐富資料,請參閱 Amazon EventBridge 管道

您可以使用 AWS Lambda 函數來處理 Amazon DynamoDB 串流中的記錄。您可以透過 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/2023-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/2023-06-10T19:26:16.525", "eventSource": "aws:dynamodb" } ]}

輪詢和批次處理串流

Lambda 會輪詢您 DynamoDB 串流中的碎片,其記錄的基本速率為每秒 4 次。當記錄可用時,Lambda 會調用您的函數,並等待結果。如果處理成功,Lambda 會恢復輪詢,直到收到多筆記錄。

Lambda 預設會在記錄可用時立即調用函數。如果 Lambda 從事件來源中讀取的批次只有一筆記錄,Lambda 只會傳送一筆記錄至函數。為避免調用具有少量記錄的函數,您可設定批次間隔,請求事件來源緩衝記錄最長達五分鐘。調用函數之前,Lambda 會繼續從事件來源中讀取記錄,直到收集到完整批次、批次間隔到期或者批次達到 6 MB 的承載限制。如需詳細資訊,請參閱 批次處理行為

警告

Lambda 事件來源對應至少處理每個事件一次,並且可能會重複處理記錄。為了避免與重複事件相關的潛在問題,我們強烈建議您將函數代碼設為冪等。若要深入了解,請參閱 AWS 知識中心如何讓 Lambda 函數具有冪等性

如果函數傳回錯誤,Lambda 會不斷重試批次直到處理成功或資料過期。若要避免停滯碎片,您可以設定事件來源映射,使用較小批次大小重試、限制重試次數,或捨棄過舊的記錄。若要保留已捨棄的事件,您可以設定事件來源映射,將失敗批次的相關詳細資訊傳送至標準 SQS 佇列或標準 SNS 主題。

若要增加並行,您還可以平行處理來自每個碎片的多個批次。Lambda 可同時處理每個碎片中最多 10 個批次。如果您增加每個碎片的並行批次數,Lambda 仍可確保在項目 (分區和排序索引鍵) 層級進行順序處理。

將此ParallelizationFactor設定設定設定為同時處理具有多個 Lambda 叫用的 Kinesis 或 DynamoDB 資料串流碎片。您可以透過從 1 (預設) 到 10 的並行化因子指定 Lambda 從碎片輪詢的並行批次數。例如,當 ParallelizationFactor 設定為 2 時,您最多可以有 200 個並行 Lambda 調用,來處理 100 個 Kinesis 資料碎片 (不過在實務中,ConcurrentExecutions 指標可能有不同值)。當資料量急劇波動並且 IteratorAge 高時,這有助於縱向擴展處理輸送量。

您也可以使用ParallelizationFactor與 Kinesis 彙總。事件來源對應的行為取決於您是否使用增強型散發:

  • 沒有增強型散發:聚合事件中的所有事件都必須具有相同的分區索引鍵。分割索引鍵也必須與彙總事件的索引鍵相符。如果彙總事件內的事件具有不同的分區索引鍵,Lambda 無法保證按照分區索引鍵按順序處理事件。

  • 透過增強型散發:首先,Lambda 會將彙總的事件解碼為其個別事件。聚集的事件可以具有與其包含的事件不同的分區索引鍵。不過,與磁碟分割索引鍵不對應的事件會遭到捨棄並遺失。Lambda 不會處理這些事件,也不會將它們傳送到設定的故障目的地。

輪詢和串流開始位置

請注意,建立和更新事件來源映射期間的串流輪詢最終會一致。

  • 在建立事件來源映射期間,從串流開始輪詢事件可能需要幾分鐘時間。

  • 在更新事件來源映射期間,從串流停止並重新開始輪詢事件可能需要幾分鐘時間。

這種行為表示如果您指定 LATEST 當作串流的開始位置,事件來源映射可能會在建立或更新期間遺漏事件。若要確保沒有遺漏任何事件,請將串流開始位置指定為 TRIM_HORIZON

DynamoDB Streams 中的碎片同時讀取

對於不是全域資料表的單一區域資料表,您最多可以設計兩個 Lambda 函數,以便同時讀取同一個 DynamoDB Streams 碎片。超過此限制會導致請求調節。對於全域資料表,建議您將同時函數的數量限制為一個,以避免請求限流。

執行角色許可

AWSLambdaDynamoDBExecutionRole AWS 管政策包括 Lambda 需要從您的 DynamoDB 串流讀取的許可。將此受管原則新增至函數的執行角色。

若要將失敗批次的記錄傳送到標準 SQS 佇列或標準 SNS 主題,您的函數需要額外許可。每個目的地服務都需要不同的許可,如下所示:

新增權限並建立事件來源對應

建立事件來源映射,指示 Lambda 從您的串流傳送記錄至 Lambda 函數。您可以建立多個事件來源映射,來使用多個 Lambda 函數處理相同資料,或使用單一函數處理來自多個串流的項目。

若要將您的函數設定為從 DynamoDB Streams 讀取,請將AWSLambdaDynamoDBExecutionRole AWS 受管理的原則附加至您的執行角色,然後建立 DynamoDB 觸發器。

若要新增權限並建立觸發器
  1. 開啟 Lambda 主控台中的 函數頁面

  2. 選擇函數的名稱。

  3. 依序選擇 Configuration (組態) 索引標籤和 Permissions (許可)。

  4. 在 [角色名稱] 下,選擇指向您的執行角色的連結。此連結會在 IAM 主控台中開啟角色。

    連結至執行角色
  5. 選擇新增許可,然後選擇連接政策

    在 IAM 主控台中附加政策
  6. 在搜尋欄位中輸入 AWSLambdaDynamoDBExecutionRole。將此原則新增至您的執行角色。這是一項 AWS 受管政策,其中包含您的函數需要從 DynamoDB 串流讀取的權限。如需有關此原則的詳細資訊,請參閱AWSLambdaDynamoDBExecutionRoleAWS 管理的原則參考中的。

  7. 返回 Lambda 主控台中的函數。在 函式概觀 下,選擇 新增觸發條件

    Lambda 主控台的函數概觀區段
  8. 選擇觸發條件類型。

  9. 設定需要的選項,然後選擇 新增

Lambda 針對事件來源支援下列選項:

事件來源選項
  • DynamoDB 資料表 - 從中讀取記錄的 DynamoDB 資料表。

  • 批次大小 – 每個批次中要傳送至函數的記錄數量,最高為 10,000。Lambda 會將批次中所有記錄以單一呼叫傳送至函數,前提是事件的總大小不超過同步調用的酬載限制 (6 MB)。

  • 批次間隔 - 指定調用函數前收集記錄的最長時間 (秒)。

  • 開始位置 - 只處理新記錄,或所有現有的記錄。

    • 最新 - 處理已新增到串流的記錄。

    • 水平修剪 - 處理所有在串流中的記錄。

    處理任何現有的記錄後,該函式已跟上進度並持續處理新的記錄。

  • 故障目的地 - 用於無法處理之記錄的標準 SQS 佇列或標準 SNS 主題。當 Lambda 捨棄太舊或已耗盡所有重試的一批記錄時,Lambda 會將該批次的詳細資料傳送至此佇列或主題。

  • 重試嘗試 - 當函數傳回錯誤時,Lambda 重試的次數上限。這不適用於服務錯誤或調節,其中批次並沒有到達函數。

  • 記錄最大存留期 - Lambda 傳送至函數之記錄的最大存留期。

  • 在錯誤時分割批次 - 當函數傳回錯誤時,先將批次分割為兩個,再進行重試。您原始的批次大小設定仍會維持不變。

  • 每個碎片的並行批次 - 同時處理來自同一個碎片的多個批次。

  • 已啟用 - 設定為 true 可啟用事件來源映射。設定為 false 以停止處理記錄。Lambda 會追蹤上次處理的進度,並在重新啟用映射時從該時間點恢復處理。

注意

作為 DynamoDB 觸發程序的一部分,Lambda 呼叫的 GetRecords API 呼叫不需支付費用。

若要稍後管理事件來源的組態,請選擇設計工具中的觸發。

事件來源映射 API

若要使用 AWS Command Line Interface (AWS CLI)AWS SDK 來管理事件來源,可以使用下列 API 操作:

下列範例會使用 AWS CLI 名為的函數對應my-function至其 Amazon 資源名稱 (ARN) 指定的 DynamoDB 串流,批次大小為 500。

aws lambda create-event-source-mapping --function-name my-function --batch-size 500 --maximum-batching-window-in-seconds 5 --starting-position LATEST \ --event-source-arn arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2023-06-10T19:26:16.525

您應該會看到下列輸出:

{ "UUID": "14e0db71-5d35-4eb5-b481-8945cf9d10c2", "BatchSize": 500, "MaximumBatchingWindowInSeconds": 5, "ParallelizationFactor": 1, "EventSourceArn": "arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2019-06-10T19:26:16.525", "FunctionArn": "arn:aws:lambda:us-east-2:123456789012:function:my-function", "LastModified": 1560209851.963, "LastProcessingResult": "No records processed", "State": "Creating", "StateTransitionReason": "User action", "DestinationConfig": {}, "MaximumRecordAgeInSeconds": 604800, "BisectBatchOnFunctionError": false, "MaximumRetryAttempts": 10000 }

設定額外選項,以自訂批次的處理方式,以及指定何時捨棄無法處理的記錄。下列範例會更新事件來源映射,以便在兩次重試嘗試後,或在記錄已超過一小時的情況下,將失敗記錄傳送至標準 SQS 佇列。

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:sqs:us-east-2:123456789012:dlq"}}'

您應該會看到以下輸出:

{ "UUID": "f89f8514-cdd9-4602-9e1f-01a5b77d449b", "BatchSize": 100, "MaximumBatchingWindowInSeconds": 0, "ParallelizationFactor": 1, "EventSourceArn": "arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2023-06-10T19:26:16.525", "FunctionArn": "arn:aws:lambda:us-east-2:123456789012:function:my-function", "LastModified": 1573243620.0, "LastProcessingResult": "PROBLEM: Function call failed", "State": "Updating", "StateTransitionReason": "User action", "DestinationConfig": {}, "MaximumRecordAgeInSeconds": 604800, "BisectBatchOnFunctionError": false, "MaximumRetryAttempts": 10000 }

系統會以非同步的方式套用更新的設定,在處理完成之前不會反映在輸出中。使用 get-event-source-mapping 命令檢視目前狀態。

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

您應該會看到以下輸出:

{ "UUID": "f89f8514-cdd9-4602-9e1f-01a5b77d449b", "BatchSize": 100, "MaximumBatchingWindowInSeconds": 0, "ParallelizationFactor": 1, "EventSourceArn": "arn:aws:dynamodb:us-east-2:123456789012:table/my-table/stream/2023-06-10T19:26:16.525", "FunctionArn": "arn:aws:lambda:us-east-2:123456789012:function:my-function", "LastModified": 1573244760.0, "LastProcessingResult": "PROBLEM: Function call failed", "State": "Enabled", "StateTransitionReason": "User action", "DestinationConfig": { "OnFailure": { "Destination": "arn:aws:sqs:us-east-2:123456789012:dlq" } }, "MaximumRecordAgeInSeconds": 3600, "BisectBatchOnFunctionError": false, "MaximumRetryAttempts": 2 }

若要同時處理多個批次,請使用 --parallelization-factor 選項。

aws lambda update-event-source-mapping --uuid 2b733gdc-8ac3-cdf5-af3a-1827b3b11284 \ --parallelization-factor 5

錯誤處理

從 DynamoDB 串流讀取記錄的事件來源映射會同步調用函數,並在錯誤時重試。如果 Lambda 調節函數或傳回錯誤,而未調用函數,則 Lambda 會重試,直至記錄過期或超過您在事件來源映射上設定的最大保留期。

如果函數收到記錄但傳回錯誤,Lambda 會重試,直至批次中的記錄過期、超過最大保留期或達到設定的重試配額。對於函數錯誤,您也可以設定事件來源映射,將失敗的批次分割為兩個批次。使用較小的批次重試可隔離不良的記錄,並解決逾時問題。分割批次並不會計入重試配額。

如果錯誤處理措施失敗,Lambda 會捨棄相應記錄,並繼續處理串流中的批次。使用預設設定時,這表示不良的記錄可能會封鎖受影響碎片上的處理長達一天。若要避免此情況,在設定函數的事件來源映射時,請使用合理的重試次數和符合您使用案例的記錄最大保留期。

若要保留已捨棄批次的記錄,請設定失敗事件的目的地。Lambda 會將包含批次相關詳細資訊的文件傳送至目的地佇列或主題。

設定失敗事件記錄的目的地
  1. 開啟 Lambda 主控台中的 函數頁面

  2. 選擇一個函數。

  3. 函數概觀 下,選擇 新增目的地

  4. 對於 來源,選擇 串流調用

  5. 對於 串流 ,選擇對應至該函數的串流。

  6. 對於 目的地類型,請選擇接收調用記錄的資源類型。

  7. 對於 目的地,請選擇一個資源。

  8. 選擇 儲存

下列範例顯示 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 調用是無狀態的,您無法在沒有外部資料庫的情況下,將其用於處理多個持續調用的資料。然而,使用輪轉時段,您可以在不同的調用間維護狀態。此狀態包含之前為目前時段處理之訊息的彙總結果。狀態可以是每個分區最多 1 MB。如果超過該大小,則 Lambda 會提前終止時段。

串流中的每個記錄都屬於一個特定時段。Lambda 至少會處理一次每筆記錄,但不保證每筆記錄只會處理一次。在極少數情況下,例如錯誤處理,某些記錄可能會處理多次。第一次時一律會依序處理記錄。如果多次處理記錄,則可能不會按順序處理。

彙總與處理

調用您的使用者管理函數進行彙總,以及處理該彙總的最終結果。Lambda 會彙總時段中接收的所有記錄。您可以在多個批次中接收這些記錄,各自作為單獨的調用。每次調用會收到一個狀態。因此,當使用輪轉時段時,您的 Lambda 函數回應必須包含 state 屬性。如果回應不包含 state 屬性,Lambda 會將此視為失敗的調用。為了滿足此條件,您的函數可以返回一個 TimeWindowEventResponse 物件,它具有下列 JSON 形狀:

範例 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 }

組態

您可以在建立或更新事件來源對映時設定輪轉時段。若要設定輪轉時段,請以秒為單位指定時段。下列範例 AWS Command Line Interface (AWS CLI) 命令會建立具有 120 秒暫停視窗的串流事件來源對應。針對彙總與處理定義的 Lambda 函數命名為 tumbling-window-example-function

aws lambda create-event-source-mapping --event-source-arn arn:aws:dynamodb:us-east-1:123456789012:stream/lambda-stream --function-name "arn:aws:lambda:us-east-1:123456789018:function:tumbling-window-example-function" --region us-east-1 --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
AWS SDK for .NET
注意

還有更多關於 GitHub。尋找完整範例,並了解如何在無伺服器範例儲存庫中設定和執行。

使用 .NET 報告使用 Lambda 批次項目失敗。

// 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
SDK for Go V2
注意

還有更多關於 GitHub。尋找完整範例,並了解如何在無伺服器範例儲存庫中設定和執行。

使用 Go 使用 Lambda 報告批次項目失敗。

// 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
適用於 Java 2.x 的 SDK
注意

還有更多關於 GitHub。尋找完整範例,並了解如何在無伺服器範例儲存庫中設定和執行。

使用 Java 使用 Lambda 報告批次項目失敗。

// 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 (v3) 的開發套件
注意

還有更多關於 GitHub。尋找完整範例,並了解如何在無伺服器範例儲存庫中設定和執行。

使用 Lambda 使用報告批次項目失敗 JavaScript

// 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: [] }; };

使用 Lambda 使用報告批次項目失敗 TypeScript

// 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 }
Python
適用於 Python (Boto3) 的 SDK
注意

還有更多關於 GitHub。尋找完整範例,並了解如何在無伺服器範例儲存庫中設定和執行。

使用 Python 使用 Lambda 報告批次項目失敗。

# 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 的開發套件
注意

還有更多關於 GitHub。尋找完整範例,並了解如何在無伺服器範例儲存庫中設定和執行。

使用紅寶石使用 Lambda 報告批次項目失敗。

# 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 報告批次項目失敗。

// 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 事件來源類型都共用相同CreateEventSourceMappingUpdateEventSourceMappingAPI 作業。但是,只有一些參數適用於 DynamoDB Streams。

適用於 DynamoDB Streams 的事件來源參數
參數 必要 預設 備註

BatchSize

100

上限:10,000

BisectBatchOnFunctionError

N

false

DestinationConfig

N

捨棄記錄的標準 Amazon SQS 佇列或標準 Amazon SNS 主題目的地

已啟用

N

true

EventSourceArn

Y

資料串流或串流消費者的 ARN

FilterCriteria

N

Lambda 事件篩選

FunctionName

Y

FunctionResponseTypes

N

若要讓函數報告批次中的特定失敗,請將值 ReportBatchItemFailures 包含在 FunctionResponseTypes 中。如需詳細資訊,請參閱 報告批次項目失敗

MaximumBatchingWindowInSeconds

N

0

MaximumRecordAgeInSeconds

N

-1

-1 表示 infinite:失敗的記錄會重試,直到記錄過期為止。DynamoDB Streams 的資料保留限制為 24 小時。

下限:-1

上限:604,800

MaximumRetryAttempts

N

-1

-1 代表無限:系統會重試失敗的記錄,直到記錄過期為止

下限:0

上限:10,000

ParallelizationFactor

N

1

上限:10

StartingPosition

Y

TRIM_HORIZON 或 LATEST

TumblingWindowInSeconds

N

下限:0

上限:900