AWS Lambda 搭配 Amazon DynamoDB 使用 - AWS Lambda
輪詢和批次處理串流輪詢和串流開始位置同時讀取範例事件

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

AWS Lambda 搭配 Amazon DynamoDB 使用

注意

如要將資料傳送到 Lambda 函數以外的目標，或在傳送資料之前讓資料更豐富，請參閱 Amazon EventBridge Pipes。

您可以使用 AWS Lambda 函數來處理 Amazon DynamoDB 串流中的記錄。您可以透過 DynamoDB Streams，以在每次更新 DynamoDB 資料表時，觸發 Lambda 函數來執行額外的工作。

處理 DynamoDB 串流時，您需要實作部分批次回應邏輯，以防止在批次中的某些記錄失敗時重試成功處理的記錄。適用於 的 Powertools 的 Batch Processor 公用程式 AWS Lambda 可在 Python、TypeScript、.NET 和 Java 中使用，並透過自動處理部分批次回應邏輯、縮短開發時間並改善可靠性來簡化此實作。

主題

輪詢和批次處理串流

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

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

警告

Lambda 事件來源映射至少會處理每個事件一次，而且可能會重複處理記錄。為避免與重複事件相關的潛在問題，強烈建議您讓函數程式碼具有等冪性。若要進一步了解，請參閱 AWS 知識中心中的如何使 Lambda 函數具有等冪性。

Lambda 在傳送下一批進行處理前不會等待任何已設定的擴充完成。換句話說，當 Lambda 處理下一批記錄時，您的擴充功能可能會繼續執行。如果您違反任何帳戶的 並行 設定或限制，便可能會產生限流的問題。若要偵測此是否為潛在問題，請監控您的函數，並確認您看到的 並行指標 是否高於事件來源映射的預期值。由於兩次調用之間的時間很短，Lambda 可能會短暫報告比碎片數目更高的並行用量。即使對於沒有延伸項目的 Lambda 函數也可能如此。

設定 ParallelizationFactor 設定來同時處理 DynamoDB 資料串流的一個碎片與多個 Lambda 調用。您可以透過從 1 (預設) 到 10 的並行化因子指定 Lambda 從碎片輪詢的並行批次數。例如，當 ParallelizationFactor 設定為 2 時，您最多可以有 200 個並行 Lambda 調用，來處理 100 個 DynamoDB 串流碎片 (不過在實務中，ConcurrentExecutions 指標可能有不同值)。當資料量急劇波動並且 IteratorAge 較高時，這有助於向上擴展處理輸送量。如果增加每個碎片的並行批次數量，Lambda 仍會確保在項目 (分割區和排序索引鍵) 層級進行依序處理。

輪詢和串流開始位置

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

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

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

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

DynamoDB Streams 中的碎片同時讀取

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

範例事件

{
  "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"
    }
  ]}