DynamoDB Streams to SQS, and SQS to DynamoDB

Send DynamoDB stream events to SQS with EventBridge Pipes for managed routing, or with Lambda when you need custom transformation. To write SQS messages into DynamoDB, use a Lambda SQS event source mapping and an idempotent database write. The queue buffers work and controls processing pace; it does not remove the need to handle duplicates, failed batches and retention limits.

Table of contents

• SQS versus DynamoDB
• DynamoDB Streams and SQS
• DynamoDB Streams to SQS
• SQS to DynamoDB
• FAQ

SQS versus DynamoDB

SQS is a queue for work awaiting processing. DynamoDB is a database for durable items retrieved through keys, queries and indexes. SQS provides receive, visibility and acknowledgement behavior; it is not a key-addressable cache or a searchable table.

Common reasons to combine them include moving slow work off an interactive request, buffering an ingestion spike, limiting concurrent database writers and retrying a temporarily unavailable downstream service. A queue helps only if consumers recover before retention expires and eventually drain the backlog.

Multiple workers reading one queue compete for messages; they do not each receive a copy. For fan-out to independent consumers, use SNS with a queue per subscriber, or another routing service. Do not use one shared queue when every service must process every event.

Standard and FIFO queues

The FIFO diagram illustrates ordering, not a guarantee that a worker's side effect happens once. A worker can crash after writing to DynamoDB but before the message is deleted, causing another delivery. FIFO send deduplication covers retries within a five-minute interval; it is not a permanent event ledger. See FIFO deduplication.

SQS message retention defaults to four days and can be configured up to 14 days. The maximum message size is now 1 MiB, replacing the older 256 KB ceiling. A queue may be configured with a smaller maximum. Check current SQS message quotas, especially before relying on an old FIFO throughput figure.

DynamoDB Streams and SQS

DynamoDB Streams captures inserts, updates and deletes for 24 hours. It gives the bridge a record of what changed without asking the application to write to DynamoDB and SQS separately. Those two independent application calls can leave a gap if one succeeds and the other fails.

A stream record is still not a permanent message archive. Monitor the bridge so it does not fall behind the stream's retention window. Once a message reaches SQS, its queue retention applies independently. Increasing SQS retention does not rescue stream records that were never forwarded.

The examples use a source table named Orders with a string partition key id, and a separate destination table named ReceivedChanges with string key eventId. Keep the destinations separate to avoid triggering the same pipeline repeatedly.

DynamoDB Streams to SQS

1. Enable the source stream and create the queue

Choose New and old images in the table's Exports and streams settings, or run the following for a table without a stream:

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

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

QUEUE_URL=$(aws sqs create-queue --queue-name order-changes \
  --query QueueUrl --output text)

QUEUE_ARN=$(aws sqs get-queue-attributes --queue-url "$QUEUE_URL" \
  --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)
aws dynamodb update-table \
  --table-name Orders \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES

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

QUEUE_URL=$(aws sqs create-queue --queue-name order-changes \
  --query QueueUrl --output text)

QUEUE_ARN=$(aws sqs get-queue-attributes --queue-url "$QUEUE_URL" \
  --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)

Run the examples in one Region. Keys-only streams cannot provide item images, and a deletion has no NewImage even when both-image mode is enabled. Choose either the pipe route or the Lambda route below for this queue; enabling both publishes duplicate business events.

2. Forward with EventBridge Pipes

Create a pipe execution role trusted by pipes.amazonaws.com. Grant it dynamodb:DescribeStream, dynamodb:GetRecords and dynamodb:GetShardIterator on the stream, plus sqs:SendMessage on the destination queue. Configure encryption-key permissions if required. Set PIPE_ROLE_ARN to that role's ARN.

This example uses one record per target invocation and transforms the stream record into a compact message. It assumes the source key is a string named id:

aws pipes create-pipe \
  --name orders-stream-to-sqs \
  --role-arn "$PIPE_ROLE_ARN" \
  --source "$STREAM_ARN" \
  --target "$QUEUE_ARN" \
  --source-parameters '{"DynamoDBStreamParameters":{"StartingPosition":"TRIM_HORIZON","BatchSize":1,"MaximumRetryAttempts":5,"MaximumRecordAgeInSeconds":3600}}' \
  --target-parameters '{"InputTemplate":"{\"eventId\": <$.eventID>, \"operation\": <$.eventName>, \"key\": {\"id\": <$.dynamodb.Keys.id.S>}}"}'
aws pipes create-pipe \
  --name orders-stream-to-sqs \
  --role-arn "$PIPE_ROLE_ARN" \
  --source "$STREAM_ARN" \
  --target "$QUEUE_ARN" \
  --source-parameters '{"DynamoDBStreamParameters":{"StartingPosition":"TRIM_HORIZON","BatchSize":1,"MaximumRetryAttempts":5,"MaximumRecordAgeInSeconds":3600}}' \
  --target-parameters '{"InputTemplate":"{\"eventId\": <$.eventID>, \"operation\": <$.eventName>, \"key\": {\"id\": <$.dynamodb.Keys.id.S>}}"}'

The transform sends keys and operation type, not full images. Expand it only with fields available for the event types you route. An input transform does not execute arbitrary JavaScript or automatically deserialize all DynamoDB values. If a worker re-reads the table using a key, it sees current state, which may differ from the state when the event occurred.

Configure a dead-letter destination before production use, and verify its permissions. If you recreate the DynamoDB stream, create a new pipe using the new ARN; a pipe's source cannot simply be changed. AWS's DynamoDB source documentation covers startup, filtering and retry behavior.

For a FIFO queue target, configure message group and deduplication parameters deliberately. Use a group associated with the entity whose ordering you need. A single global group serializes the entire workload, while unrelated group IDs can break an entity's intended order.

3. Alternative: forward with Lambda and SDK v3

Use a nodejs24.x function with handler index.handler in index.mjs, bundling @aws-sdk/client-sqs and @aws-sdk/util-dynamodb. Set QUEUE_URL. Grant stream-read permissions, sqs:SendMessage on the queue and CloudWatch Logs permissions.

import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";
import { unmarshall } from "@aws-sdk/util-dynamodb";

const sqs = new SQSClient({});

export const handler = async (event) => {
  for (const record of event.Records) {
    try {
      await sqs.send(new SendMessageCommand({
        QueueUrl: process.env.QUEUE_URL,
        MessageBody: JSON.stringify({
          eventId: `${record.eventSourceARN}#${record.eventID}`,
          operation: record.eventName,
          key: unmarshall(record.dynamodb.Keys),
        }),
      }));
    } catch (error) {
      console.error("Forward failed", record.eventID, error.name);
      return {
        batchItemFailures: [{ itemIdentifier: record.dynamodb.SequenceNumber }],
      };
    }
  }
  return { batchItemFailures: [] };
};
import { SQSClient, SendMessageCommand } from "@aws-sdk/client-sqs";
import { unmarshall } from "@aws-sdk/util-dynamodb";

const sqs = new SQSClient({});

export const handler = async (event) => {
  for (const record of event.Records) {
    try {
      await sqs.send(new SendMessageCommand({
        QueueUrl: process.env.QUEUE_URL,
        MessageBody: JSON.stringify({
          eventId: `${record.eventSourceARN}#${record.eventID}`,
          operation: record.eventName,
          key: unmarshall(record.dynamodb.Keys),
        }),
      }));
    } catch (error) {
      console.error("Forward failed", record.eventID, error.name);
      return {
        batchItemFailures: [{ itemIdentifier: record.dynamodb.SequenceNumber }],
      };
    }
  }
  return { batchItemFailures: [] };
};

This publisher targets a standard queue. For FIFO, also supply MessageGroupId and MessageDeduplicationId appropriate to your ordering and retry model. Keep payloads within the queue's configured size rather than assuming both full item images always fit.

Connect the function with a stream event source mapping, using TRIM_HORIZON, ReportBatchItemFailures, bounded retry and age settings, and a failure destination. The DynamoDB trigger guide gives complete setup examples. The handler processes the full batch and stops at the first failed stream record.

A successful send followed by a lost response can create a duplicate on retry. Deduplicate in the consumer by the stable domain event ID. SendMessageBatch can reduce request overhead, but inspect its per-entry failures even when the API call itself succeeds.

4. Test stream-to-queue delivery

Write, update and delete a test order. Check that all operation types appear in the queue, then test a batch containing several changes. A manual ReceiveMessage temporarily hides a message under its visibility timeout; it does not acknowledge it. Avoid competing with a live production consumer during inspection.

Test a failed send and verify that the bridge retries, retains recovery information and eventually resumes. Alarm on stream lag and delivery failures. If downstream processing is unavailable, confirm that the queue's oldest-message age remains within the retention window.

SQS to DynamoDB

1. Prepare the worker

Create ReceivedChanges with string partition key eventId. Deploy a nodejs24.x Lambda function named sqs-to-dynamodb, set TABLE_NAME=ReceivedChanges, and bundle @aws-sdk/client-dynamodb and @aws-sdk/lib-dynamodb.

The execution role needs sqs:ReceiveMessage, sqs:DeleteMessage, sqs:GetQueueAttributes and dynamodb:PutItem on the relevant resources, plus logs and any necessary KMS permissions. The queue and function must be in the same Region.

2. Handle the complete batch and make writes idempotent

This standard-queue handler validates the message, stores it once under its event ID, and returns the SQS message IDs of records that failed:

import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";

const ddb = DynamoDBDocumentClient.from(new DynamoDBClient({}));

export const handler = async (event) => {
  const batchItemFailures = [];
  for (const record of event.Records) {
    try {
      const message = JSON.parse(record.body);
      if (typeof message.eventId !== "string" || message.eventId.length === 0) {
        throw new Error("Message requires a non-empty eventId");
      }
      try {
        await ddb.send(new PutCommand({
          TableName: process.env.TABLE_NAME,
          Item: { eventId: message.eventId, payload: message },
          ConditionExpression: "attribute_not_exists(eventId)",
        }));
      } catch (error) {
        if (error.name !== "ConditionalCheckFailedException") throw error;
        // This immutable event is already stored.
      }
    } catch (error) {
      console.error("Message failed", record.messageId, error.name);
      batchItemFailures.push({ itemIdentifier: record.messageId });
    }
  }
  return { batchItemFailures };
};
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";

const ddb = DynamoDBDocumentClient.from(new DynamoDBClient({}));

export const handler = async (event) => {
  const batchItemFailures = [];
  for (const record of event.Records) {
    try {
      const message = JSON.parse(record.body);
      if (typeof message.eventId !== "string" || message.eventId.length === 0) {
        throw new Error("Message requires a non-empty eventId");
      }
      try {
        await ddb.send(new PutCommand({
          TableName: process.env.TABLE_NAME,
          Item: { eventId: message.eventId, payload: message },
          ConditionExpression: "attribute_not_exists(eventId)",
        }));
      } catch (error) {
        if (error.name !== "ConditionalCheckFailedException") throw error;
        // This immutable event is already stored.
      }
    } catch (error) {
      console.error("Message failed", record.messageId, error.name);
      batchItemFailures.push({ itemIdentifier: record.messageId });
    }
  }
  return { batchItemFailures };
};

This assumes an event ID always means the same immutable event. For business-state updates, design a version condition instead of blindly replacing items. A valid SQS message can still exceed DynamoDB's 400 KB item limit, so validate the stored representation and use object-storage references for larger payloads.

For FIFO queues, stop after the first failure and return that message plus all unprocessed messages as failures. Continuing past a failed message can violate the intended processing order. AWS describes this distinction in SQS partial batch responses.

3. Configure visibility, redrive and the mapping

Set the source queue visibility timeout to at least six times the function timeout, plus any configured batching window. Configure the source queue's redrive policy to move repeatedly failing messages to a dead-letter queue. This is separate from a DynamoDB stream mapping's on-failure destination.

aws lambda create-event-source-mapping \
  --function-name sqs-to-dynamodb \
  --event-source-arn "$QUEUE_ARN" \
  --batch-size 10 \
  --function-response-types ReportBatchItemFailures
aws lambda create-event-source-mapping \
  --function-name sqs-to-dynamodb \
  --event-source-arn "$QUEUE_ARN" \
  --batch-size 10 \
  --function-response-types ReportBatchItemFailures

An SQS mapping has no stream StartingPosition. Lambda polls the queue and deletes messages after successful processing. With partial responses enabled, successful messages can be acknowledged while failed ones reappear for retry. Follow the Lambda SQS configuration guide for visibility and redrive settings.

4. Test queue-to-table delivery

aws sqs send-message \
  --queue-url "$QUEUE_URL" \
  --message-body '{"eventId":"test-change-001","operation":"INSERT","key":{"id":"order-001"}}'

aws dynamodb get-item \
  --table-name ReceivedChanges \
  --key '{"eventId":{"S":"test-change-001"}}' \
  --consistent-read
aws sqs send-message \
  --queue-url "$QUEUE_URL" \
  --message-body '{"eventId":"test-change-001","operation":"INSERT","key":{"id":"order-001"}}'

aws dynamodb get-item \
  --table-name ReceivedChanges \
  --key '{"eventId":{"S":"test-change-001"}}' \
  --consistent-read

Wait for the worker before reading. Send the same event again and verify one stored item. Send malformed JSON and verify retry and DLQ behavior, then test a mixed batch with valid and invalid messages. Catching every exception and returning success would silently acknowledge bad messages, so do not suppress unexpected errors.

Capacity, monitoring and operational limits

A queue smooths write demand but does not reduce the total DynamoDB writes required for the same work. Limit consumer concurrency when draining too quickly would throttle the destination. Conversely, a permanently underpowered consumer eventually fills its retention window and loses messages.

Monitor SQS ApproximateAgeOfOldestMessage, visible-message count and DLQ depth; Lambda duration, errors and throttles; and destination-table throttling. For a stream bridge, also monitor iterator age or the pipe's execution and delivery telemetry. See DynamoDB monitoring for cause-specific throttle diagnosis.

Standard Lambda reads from DynamoDB Streams are free, except on Lambda Managed Instances. SQS requests, Lambda execution, Pipes usage, DynamoDB writes and logging have their own charges. A queue is not a cache: use a suitable cache or read model when the problem is repeated reads, rather than asynchronous writes.