How to capture data changes in DynamoDB using Streams and Lambda (Node.js + AWS CDK)

This content originally appeared on DEV Community and was authored by iAmSherif

One of the biggest advantages of serverless services is their event-driven nature. When something (an event) happens, another service reacts. DynamoDB is a serverless NoSQL database that fits perfectly into this pattern. It allows you to build scalable, event-driven applications that respond to changes in your data in near real time.

In this article, I’ll show you how to capture data changes in DynamoDB using DynamoDB Streams, and then use the stream as an event source to a Lambda function, which then logs the stream information to CloudWatch Logs, all implemented using Node.js and AWS CDK as Iac.

Why Capture Data Changes?

Before we dive into the implementation, let’s consider a few use cases where capturing data changes can be valuable:

Audit logging: Keep track of item-level changes for compliance or debugging.
Triggering side effects: Send notifications, update search indexes, or replicate data in real time.
Analytics: Track user behavior or system events as they happen.
Data synchronization: Sync changes between microservices or external systems.

How DynamoDB Streams Work

When you enable Streams on a DynamoDB table, DynamoDB records every item-level change (creates, updates, and deletes) in a time-ordered sequence.

Each stream record contains:

The primary key of the modified item
Metadata about the operation (INSERT, MODIFY, REMOVE)
Optional “before” and “after” images of the data (if configured)

These stream records are stored for up to 24 hours and can be read by AWS services like Lambda, which you can set up to react to each change automatically.

What We’ll Build

We’ll create:

A DynamoDB table with Streams enabled
A Lambda function that listens to the stream
Logging of the change details to CloudWatch Logs

So Let’s Get Started

1. Define a DynamoDB Table with Stream Enabled

import { AttributeType, StreamViewType, Table } from 'aws-cdk-lib/aws-dynamodb';

const table = new Table(this, 'MyTable', {
  partitionKey: { name: 'id', type: AttributeType.STRING },
  stream: StreamViewType.NEW_AND_OLD_IMAGES, // Enables streams with both before/after images
  // Other configuration
});

Stream View Types

The StreamViewType determines what data is recorded in the stream for each item change:

KEYS_ONLY – Only the primary key attributes are written to the stream.
NEW_IMAGE – The entire item, as it appears after the modification, is written to the stream.
OLD_IMAGE – The item as it appeared before the modification is written to the stream.
NEW_AND_OLD_IMAGES – Both the new and old versions of the item are included.

2. Create a Lambda Function to Handle Stream Events

import { Function, Runtime, Code } from 'aws-cdk-lib/aws-lambda';
import { LambdaIntegration } from 'aws-cdk-lib/aws-apigateway';
import * as path from 'path';

const streamHandler = new Function(this, 'StreamHandlerFunction', {
  runtime: Runtime.NODEJS_20_X,
  handler: 'index.handler',
  code: Code.fromAsset(path.join(__dirname, '../lambda')),
});

Sample index.js Lambda code:

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

export const handler = async (event) => {
  for (const record of event.Records) {
    const newImage = record.dynamodb.NewImage
      ? unmarshall(record.dynamodb.NewImage)
      : null;
    const oldImage = record.dynamodb.OldImage
      ? unmarshall(record.dynamodb.OldImage)
      : null;

    console.log('Event Type:', record.eventName);
    console.log('New Image:', JSON.stringify(newImage));
    console.log('Old Image:', JSON.stringify(oldImage));
  }

  return { statusCode: 200 };
};

Why do we use `unmarshall`?

When DynamoDB Streams triggers our Lambda function, it passes the item images in DynamoDB JSON format, which looks like this:

{
  "id": { "S": "asdf" },
  "uuid": { "S": "foo" },
  "status": { "S": "new" }
}

To work with standard JavaScript objects, we use the unmarshall utility from @aws-sdk/util-dynamodb. This transforms the DynamoDB JSON into a regular JavaScript object:

{
  "id": "asdf",
  "uuid": "foo",
  "status": "new"
}

This makes it easier to log or manipulate the data in our code.

3. Connect the Stream to Lambda

To allow our Lambda function to process events from the DynamoDB stream, we need to do two things:

Grant the Lambda the proper execution role permission to read from the stream
Register the DynamoDB table as an event source

// Grant the Lambda the proper execution role permission to read from the stream
table.grantStreamRead(streamHandler);

// Attach DynamoDB stream as an event source
streamHandler.addEventSource(new DynamoEventSource(table, {
  startingPosition: StartingPosition.LATEST,
  batchSize: 5,
  retryAttempts: 2,
}));

Configuration Options

startingPosition – Defines where the Lambda function should start reading the stream (e.g., LATEST for new records only).
batchSize – The maximum number of records to send to the function in a single invocation. The Lambda will receive all these records in the event payload.
retryAttempts – The maximum number of times Lambda will retry a failed batch before discarding or sending it to a dead-letter destination (if configured).

Conclusion

By using DynamoDB Streams, you can build powerful real-time event-driven systems with minimal operational overhead. In this example, we simply logged data changes to CloudWatch, but you can extend this to trigger notifications, sync systems, or persist audit logs to S3.

——————————————

For more articles, follow my social handles: