DynamoDB Triggers with Lambda: Setup, Examples and Retries

A DynamoDB trigger is a Lambda event source mapping that reads a table's DynamoDB Stream and invokes a function with batches of item changes. It runs after the database write succeeds, so it cannot reject or roll back that write. Lambda can deliver a record more than once; reliable triggers combine idempotent processing, bounded retries and a recovery destination.

What runs when an item changes?

The stream captures INSERT, MODIFY and REMOVE events. Lambda polls the stream, batches records, invokes your function and advances its checkpoint after successful processing. The writer does not wait for the function. A write that leaves the item unchanged produces no stream record.

Use triggers for notifications, search indexing, audit exports and derived data. Validate a write in the application or with a DynamoDB condition expression when it must be rejected before storage. See how DynamoDB Streams work for shard behavior, record retention and the full event format.

Enable the stream and prepare permissions

Choose a stream view that contains the data your handler needs:

An insert never has an old image; a delete never has a new one. A MODIFY record contains both only when you selected both. The examples use a People table with a string partition key named PersonId.

For a table without an active stream:

aws dynamodb update-table \
  --table-name People \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES

STREAM_ARN=$(aws dynamodb describe-table --table-name People \
  --query 'Table.LatestStreamArn' --output text)
aws dynamodb update-table \
  --table-name People \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES

STREAM_ARN=$(aws dynamodb describe-table --table-name People \
  --query 'Table.LatestStreamArn' --output text)

Use the table's Region for every command. Changing a view requires replacing the stream, which changes its ARN; reconnect mappings and stream policies afterward.

The function's execution role needs dynamodb:DescribeStream, dynamodb:GetShardIterator, dynamodb:GetRecords and dynamodb:ListStreams, plus CloudWatch Logs access. The AWS managed policy AWSLambdaDynamoDBExecutionRole supplies baseline permissions; a custom policy can restrict stream actions to the intended resource. These are runtime permissions, separate from the deployment identity's permission to create mappings and pass the execution role.

If your handler calls another service, grant that operation separately. A failure queue requires sqs:SendMessage on its ARN. An S3 failure destination requires bucket/object permissions and, with a customer managed key, the applicable KMS permission.

Create the Lambda trigger

Create a function named people-stream-handler, using nodejs24.x for the JavaScript example below or python3.13 for Python. Package the handler and its dependencies, and configure the execution role before adding the trigger.

In the Lambda console, open the function, choose Add trigger, select DynamoDB and the table, and configure the starting position and batch settings. Review the advanced retry and failure settings rather than accepting unlimited retries automatically.

The equivalent AWS CLI v2 command uses an existing standard SQS queue for failure metadata. Replace its example account ID and Region:

aws lambda create-event-source-mapping \
  --function-name people-stream-handler \
  --event-source-arn "$STREAM_ARN" \
  --starting-position TRIM_HORIZON \
  --batch-size 100 \
  --maximum-batching-window-in-seconds 5 \
  --maximum-retry-attempts 5 \
  --maximum-record-age-in-seconds 3600 \
  --bisect-batch-on-function-error \
  --function-response-types ReportBatchItemFailures \
  --destination-config '{"OnFailure":{"Destination":"arn:aws:sqs:us-east-1:123456789012:people-stream-failures"}}'
aws lambda create-event-source-mapping \
  --function-name people-stream-handler \
  --event-source-arn "$STREAM_ARN" \
  --starting-position TRIM_HORIZON \
  --batch-size 100 \
  --maximum-batching-window-in-seconds 5 \
  --maximum-retry-attempts 5 \
  --maximum-record-age-in-seconds 3600 \
  --bisect-batch-on-function-error \
  --function-response-types ReportBatchItemFailures \
  --destination-config '{"OnFailure":{"Destination":"arn:aws:sqs:us-east-1:123456789012:people-stream-failures"}}'

These retry and age values are example operating choices, not universal defaults. Keep the mapping UUID from the response. Use get-event-source-mapping --uuid to inspect status and update-event-source-mapping --uuid to change it.

CloudFormation setup

This deployable template creates a table, log-only Python function, role, failure queue and mapping. Replace the inline handler with your packaged business logic for a real service:

AWSTemplateFormatVersion: "2010-09-09"
Resources:
  PeopleTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: PersonId
          AttributeType: S
      KeySchema:
        - AttributeName: PersonId
          KeyType: HASH
      StreamSpecification:
        StreamViewType: NEW_AND_OLD_IMAGES
  Failures:
    Type: AWS::SQS::Queue
  HandlerRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - !Sub arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaDynamoDBExecutionRole
      Policies:
        - PolicyName: SendFailureMetadata
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action: sqs:SendMessage
                Resource: !GetAtt Failures.Arn
  Handler:
    Type: AWS::Lambda::Function
    Properties:
      Runtime: python3.13
      Handler: index.handler
      Role: !GetAtt HandlerRole.Arn
      Code:
        ZipFile: |
          def handler(event, context):
              for record in event["Records"]:
                  print(record["eventID"], record["eventName"])
              return {"batchItemFailures": []}
  Trigger:
    Type: AWS::Lambda::EventSourceMapping
    Properties:
      EventSourceArn: !GetAtt PeopleTable.StreamArn
      FunctionName: !Ref Handler
      StartingPosition: TRIM_HORIZON
      BatchSize: 100
      MaximumBatchingWindowInSeconds: 5
      MaximumRetryAttempts: 5
      MaximumRecordAgeInSeconds: 3600
      BisectBatchOnFunctionError: true
      FunctionResponseTypes:
        - ReportBatchItemFailures
      DestinationConfig:
        OnFailure:
          Destination: !GetAtt Failures.Arn
AWSTemplateFormatVersion: "2010-09-09"
Resources:
  PeopleTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: PersonId
          AttributeType: S
      KeySchema:
        - AttributeName: PersonId
          KeyType: HASH
      StreamSpecification:
        StreamViewType: NEW_AND_OLD_IMAGES
  Failures:
    Type: AWS::SQS::Queue
  HandlerRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - !Sub arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaDynamoDBExecutionRole
      Policies:
        - PolicyName: SendFailureMetadata
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action: sqs:SendMessage
                Resource: !GetAtt Failures.Arn
  Handler:
    Type: AWS::Lambda::Function
    Properties:
      Runtime: python3.13
      Handler: index.handler
      Role: !GetAtt HandlerRole.Arn
      Code:
        ZipFile: |
          def handler(event, context):
              for record in event["Records"]:
                  print(record["eventID"], record["eventName"])
              return {"batchItemFailures": []}
  Trigger:
    Type: AWS::Lambda::EventSourceMapping
    Properties:
      EventSourceArn: !GetAtt PeopleTable.StreamArn
      FunctionName: !Ref Handler
      StartingPosition: TRIM_HORIZON
      BatchSize: 100
      MaximumBatchingWindowInSeconds: 5
      MaximumRetryAttempts: 5
      MaximumRecordAgeInSeconds: 3600
      BisectBatchOnFunctionError: true
      FunctionResponseTypes:
        - ReportBatchItemFailures
      DestinationConfig:
        OnFailure:
          Destination: !GetAtt Failures.Arn

AWS CDK with Python

Within a CDK v2 application, this alternative creates the same relationship. Put the Python handler shown below in lambda/people/index.py and package its dependencies in that asset directory:

from aws_cdk import Duration, Stack
from aws_cdk import aws_dynamodb as dynamodb
from aws_cdk import aws_lambda as lambda_
from aws_cdk import aws_lambda_event_sources as sources
from aws_cdk import aws_sqs as sqs
from constructs import Construct


class PeopleStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs):
        super().__init__(scope, construct_id, **kwargs)
        table = dynamodb.TableV2(
            self, "People",
            partition_key=dynamodb.Attribute(
                name="PersonId", type=dynamodb.AttributeType.STRING
            ),
            dynamo_stream=dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,
        )
        function = lambda_.Function(
            self, "Handler",
            runtime=lambda_.Runtime.PYTHON_3_13,
            code=lambda_.Code.from_asset("lambda/people"),
            handler="index.handler",
        )
        failures = sqs.Queue(self, "Failures")
        function.add_event_source(sources.DynamoEventSource(
            table,
            starting_position=lambda_.StartingPosition.TRIM_HORIZON,
            batch_size=100,
            max_batching_window=Duration.seconds(5),
            retry_attempts=5,
            max_record_age=Duration.hours(1),
            bisect_batch_on_error=True,
            report_batch_item_failures=True,
            on_failure=sources.SqsDlq(failures),
        ))
from aws_cdk import Duration, Stack
from aws_cdk import aws_dynamodb as dynamodb
from aws_cdk import aws_lambda as lambda_
from aws_cdk import aws_lambda_event_sources as sources
from aws_cdk import aws_sqs as sqs
from constructs import Construct


class PeopleStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs):
        super().__init__(scope, construct_id, **kwargs)
        table = dynamodb.TableV2(
            self, "People",
            partition_key=dynamodb.Attribute(
                name="PersonId", type=dynamodb.AttributeType.STRING
            ),
            dynamo_stream=dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,
        )
        function = lambda_.Function(
            self, "Handler",
            runtime=lambda_.Runtime.PYTHON_3_13,
            code=lambda_.Code.from_asset("lambda/people"),
            handler="index.handler",
        )
        failures = sqs.Queue(self, "Failures")
        function.add_event_source(sources.DynamoEventSource(
            table,
            starting_position=lambda_.StartingPosition.TRIM_HORIZON,
            batch_size=100,
            max_batching_window=Duration.seconds(5),
            retry_attempts=5,
            max_record_age=Duration.hours(1),
            bisect_batch_on_error=True,
            report_batch_item_failures=True,
            on_failure=sources.SqsDlq(failures),
        ))

Event source mapping settings

AWS documents the mapping parameters separately from general Lambda asynchronous invocation options. Configure retry handling on the mapping, not just a function-level asynchronous dead-letter queue.

TRIM_HORIZON is usually the safer initial position: mapping creation can take time, and LATEST can miss writes during that startup interval. The new DynamoDB Streams timestamp-start capability is available to supported stream readers, but the Lambda DynamoDB mapping documentation still lists TRIM_HORIZON and LATEST. Do not assume every consumer accepts AT_TIMESTAMP.

Handle inserts, updates, deletes and TTL

This abbreviated insert event keeps Keys and NewImage consistent:

{
  "Records": [{
    "eventID": "example-event-1",
    "eventName": "INSERT",
    "eventSource": "aws:dynamodb",
    "dynamodb": {
      "Keys": {"PersonId": {"S": "Person-123"}},
      "NewImage": {
        "PersonId": {"S": "Person-123"},
        "FirstName": {"S": "John"},
        "LastName": {"S": "Doe"}
      },
      "SequenceNumber": "111",
      "StreamViewType": "NEW_AND_OLD_IMAGES"
    }
  }]
}
{
  "Records": [{
    "eventID": "example-event-1",
    "eventName": "INSERT",
    "eventSource": "aws:dynamodb",
    "dynamodb": {
      "Keys": {"PersonId": {"S": "Person-123"}},
      "NewImage": {
        "PersonId": {"S": "Person-123"},
        "FirstName": {"S": "John"},
        "LastName": {"S": "Doe"}
      },
      "SequenceNumber": "111",
      "StreamViewType": "NEW_AND_OLD_IMAGES"
    }
  }]
}

For a name update, eventName becomes MODIFY, OldImage.FirstName.S is John, and NewImage.FirstName.S is James. For deletion, eventName is REMOVE and only the old image is present. All events retain the key attributes.

JavaScript handler using SDK v3

Save as index.mjs, with handler index.handler, and bundle @aws-sdk/util-dynamodb. This handler logs the insert, name change or deletion and stops at the first failed record. Replace the log operation with an awaited, idempotent side effect:

import { unmarshall } from "@aws-sdk/util-dynamodb";

export const handler = async (event) => {
  for (const record of event.Records) {
    try {
      const data = record.dynamodb;
      const before = data.OldImage ? unmarshall(data.OldImage) : undefined;
      const after = data.NewImage ? unmarshall(data.NewImage) : undefined;
      const keys = unmarshall(data.Keys);
      switch (record.eventName) {
        case "INSERT":
          console.log("created", keys.PersonId, after.FirstName);
          break;
        case "MODIFY":
          console.log("updated", keys.PersonId, before.FirstName, after.FirstName);
          break;
        case "REMOVE":
          console.log("removed", keys.PersonId, before.FirstName);
          break;
      }
    } catch (error) {
      console.error("Record failed", record.eventID, error.name);
      return {
        batchItemFailures: [{ itemIdentifier: record.dynamodb.SequenceNumber }],
      };
    }
  }
  return { batchItemFailures: [] };
};
import { unmarshall } from "@aws-sdk/util-dynamodb";

export const handler = async (event) => {
  for (const record of event.Records) {
    try {
      const data = record.dynamodb;
      const before = data.OldImage ? unmarshall(data.OldImage) : undefined;
      const after = data.NewImage ? unmarshall(data.NewImage) : undefined;
      const keys = unmarshall(data.Keys);
      switch (record.eventName) {
        case "INSERT":
          console.log("created", keys.PersonId, after.FirstName);
          break;
        case "MODIFY":
          console.log("updated", keys.PersonId, before.FirstName, after.FirstName);
          break;
        case "REMOVE":
          console.log("removed", keys.PersonId, before.FirstName);
          break;
      }
    } catch (error) {
      console.error("Record failed", record.eventID, error.name);
      return {
        batchItemFailures: [{ itemIdentifier: record.dynamodb.SequenceNumber }],
      };
    }
  }
  return { batchItemFailures: [] };
};

This example logs names to demonstrate the images. In production, log event IDs and approved diagnostic fields rather than personal data or complete item images. SDK v3 replaces the unsupported aws-sdk v2 package.

Python handler with boto3

Boto3's TypeDeserializer converts DynamoDB JSON values; numeric attributes become Decimal. This version handles the same event types:

from boto3.dynamodb.types import TypeDeserializer

serializer = TypeDeserializer()


def decode(image):
    return {key: serializer.deserialize(value) for key, value in image.items()}


def handler(event, context):
    for record in event["Records"]:
        try:
            data = record["dynamodb"]
            keys = decode(data["Keys"])
            before = decode(data.get("OldImage", {}))
            after = decode(data.get("NewImage", {}))
            if record["eventName"] == "INSERT":
                print("created", keys["PersonId"], after.get("FirstName"))
            elif record["eventName"] == "MODIFY":
                print("updated", keys["PersonId"],
                      before.get("FirstName"), after.get("FirstName"))
            elif record["eventName"] == "REMOVE":
                print("removed", keys["PersonId"], before.get("FirstName"))
        except Exception as error:
            print("Record failed", record["eventID"], type(error).__name__)
            return {"batchItemFailures": [{
                "itemIdentifier": record["dynamodb"]["SequenceNumber"]
            }]}
    return {"batchItemFailures": []}
from boto3.dynamodb.types import TypeDeserializer

serializer = TypeDeserializer()


def decode(image):
    return {key: serializer.deserialize(value) for key, value in image.items()}


def handler(event, context):
    for record in event["Records"]:
        try:
            data = record["dynamodb"]
            keys = decode(data["Keys"])
            before = decode(data.get("OldImage", {}))
            after = decode(data.get("NewImage", {}))
            if record["eventName"] == "INSERT":
                print("created", keys["PersonId"], after.get("FirstName"))
            elif record["eventName"] == "MODIFY":
                print("updated", keys["PersonId"],
                      before.get("FirstName"), after.get("FirstName"))
            elif record["eventName"] == "REMOVE":
                print("removed", keys["PersonId"], before.get("FirstName"))
        except Exception as error:
            print("Record failed", record["eventID"], type(error).__name__)
            return {"batchItemFailures": [{
                "itemIdentifier": record["dynamodb"]["SequenceNumber"]
            }]}
    return {"batchItemFailures": []}

TTL deletes

A TTL expiration produces a REMOVE event with userIdentity.type equal to Service and userIdentity.principalId equal to dynamodb.amazonaws.com. Use those fields to distinguish expiration from an application delete. On global tables, only the Region that performed expiration marks the event this way. TTL deletion is asynchronous, typically within a few days; it is not a precise scheduler. See DynamoDB TTL.

Filter events before invoking Lambda

For a function that handles only inserts, save this object as filters.json:

{
  "Filters": [{
    "Pattern": "{\"eventName\":[\"INSERT\"]}"
  }]
}
{
  "Filters": [{
    "Pattern": "{\"eventName\":[\"INSERT\"]}"
  }]
}

aws lambda update-event-source-mapping \
  --uuid "$MAPPING_UUID" \
  --filter-criteria file://filters.json
aws lambda update-event-source-mapping \
  --uuid "$MAPPING_UUID" \
  --filter-criteria file://filters.json

To match a particular new name, the pattern itself would be:

{
  "eventName": ["MODIFY"],
  "dynamodb": {"NewImage": {"FirstName": {"S": ["James"]}}}
}
{
  "eventName": ["MODIFY"],
  "dynamodb": {"NewImage": {"FirstName": {"S": ["James"]}}}
}

Put that pattern into the mapping's string-valued Pattern field. Filtering uses DynamoDB's typed representation, not the plain objects produced by unmarshall. Numeric filter operators do not work on DynamoDB number strings. Filters cannot compare old and new values dynamically; compare them in your handler when you need a transition check.

Unmatched records are skipped, not saved for later. Changing a filter does not rewind a mapping. AWS's DynamoDB event filtering documentation covers the supported pattern structure.

Retries, partial batches and recovery

Without partial responses, throwing an error retries the whole batch, including records that already succeeded. With ReportBatchItemFailures, return the failed record's sequence number, not eventID. Lambda resumes from the lowest failed sequence number. Both handlers stop at the first failure so they do not perform later work that would immediately be replayed.

Bisecting helps isolate invocation failures; splitting a batch does not consume the retry quota. Retry-count limits apply to errors returned by an invoked function. Pre-invocation failures, such as throttling, are governed by record age and retention rather than that same function-error retry count.

When retry or age limits are reached, Lambda discards records and advances. A standard SQS or SNS failure destination receives invocation metadata and stream sequence information, not the full original batch. Recover those records before the stream's 24-hour retention expires. An S3 destination retains the full invocation payload, which supports later investigation and replay. Follow the failure-destination configuration and monitor delivery failures too.

Make downstream operations idempotent. For a database projection, store and conditionally compare a business version. For an external API, pass a stable idempotency key supported by that API. Merely recording an event ID before sending a notification can lose the notification if the function crashes between those steps.

Cross-account triggers, scaling and monitoring

Lambda supports a DynamoDB stream in another account through a resource-based policy on that stream. The function and stream must still be in the same Region. Grant the function execution role stream-read permission in its own account and allow that role in the stream policy. Follow AWS's cross-account Lambda setup. Recreated streams need a policy on their new ARN.

Design for up to two readers per shard on a single-Region table; AWS recommends one for global tables. Multiple mappings do not provide unlimited fan-out. Use one consumer to publish notifications through SNS, or use Kinesis Data Streams when the consumer requirements justify it.

Monitor IteratorAge, Lambda errors, duration, throttles and destination-delivery failures. Rising iterator age means processing is falling behind; investigate failures and downstream latency before raising concurrency. Avoid writing back to the source item without a filter or condition that prevents an invocation loop.

Standard Lambda trigger reads from DynamoDB Streams are free, while Lambda execution and downstream services are billed. Lambda Managed Instances are an exception to that free stream-read treatment. Test replayed records, missing images, malformed attributes, permission failures and a deliberately failing record before enabling production traffic.