Streaming DynamoDB Changes to EventBridge: Why You Need EventBridge Pipes

The Problem: DynamoDB Streams Can’t Directly Target EventBridge

If you’ve worked with DynamoDB Streams and EventBridge, you might have expected to configure your DynamoDB table to automatically send stream records directly to an EventBridge event bus. Unfortunately, DynamoDB Streams do not natively support EventBridge as a direct target.

What DynamoDB Streams Support

DynamoDB Streams can only invoke:

AWS Lambda functions
Amazon Kinesis Data Streams

There’s no built-in CloudFormation property or AWS Console option to send DynamoDB Stream records directly to EventBridge. This is a significant limitation when you want to build event-driven architectures.

The Solution: EventBridge Pipes

AWS introduced EventBridge Pipes to bridge this gap. Pipes provide a managed service that connects point-to-point integrations between event sources (like DynamoDB Streams) and targets (like EventBridge).

Why Use Pipes Instead of Lambda?

Before Pipes, the common workaround was:

DynamoDB Stream → Lambda Function → EventBridge

With Pipes, you get:

DynamoDB Stream → EventBridge Pipe → EventBridge

Benefits of using Pipes:

✅ No Lambda function to maintain
✅ Built-in filtering and transformation
✅ Automatic retries and error handling
✅ Lower latency
✅ Lower cost (no Lambda invocation charges)
✅ Built-in CloudWatch logging

CloudFormation Implementation

Here’s a complete example of how to set up a DynamoDB Stream to EventBridge integration using Pipes:

1. DynamoDB Table with Stream Enabled

AuditLogsTable:
  Type: AWS::DynamoDB::Table
  Properties:
    TableName: audit-logs
    BillingMode: PAY_PER_REQUEST
    AttributeDefinitions:
      - AttributeName: PK
        AttributeType: S
      - AttributeName: SK
        AttributeType: S
    KeySchema:
      - AttributeName: PK
        KeyType: HASH
      - AttributeName: SK
        KeyType: RANGE
    StreamSpecification:
      StreamViewType: NEW_AND_OLD_IMAGES  # Captures both old and new item states

2. EventBridge Custom Event Bus

CustomEventBus:
  Type: AWS::Events::EventBus
  Properties:
    Name: my-event-bus
    DeadLetterConfig:
      Arn: !GetAtt EventBusDLQ.Arn

EventBusDLQ:
  Type: AWS::SQS::Queue
  Properties:
    QueueName: event-bus-dlq

3. EventBridge Pipe with Logging

AuditLogPipeLogGroup:
  Type: AWS::Logs::LogGroup
  Properties:
    LogGroupName: /aws/vendedlogs/pipes/audit-log-pipe
    RetentionInDays: 7

AuditLogPipe:
  Type: AWS::Pipes::Pipe
  Properties:
    Name: DynamoDBStreamToEventBridgePipe
    RoleArn: !GetAtt PipeExecutionRole.Arn
    
    # Enable detailed logging
    LogConfiguration:
      CloudwatchLogsLogDestination:
        LogGroupArn: !GetAtt AuditLogPipeLogGroup.Arn
      Level: TRACE  # Options: ERROR, INFO, TRACE
      IncludeExecutionData:
        - ALL  # Log all execution data including payloads

    # SOURCE: DynamoDB Stream
    Source: !GetAtt AuditLogsTable.StreamArn
    SourceParameters:
      DynamoDBStreamParameters:
        BatchSize: 1
        StartingPosition: LATEST  # Options: LATEST, TRIM_HORIZON
    
    # TARGET: EventBridge Bus
    Target: !GetAtt CustomEventBus.Arn
    TargetParameters:
      InputTemplate: |
        {
          "source": "<$.dynamodb.NewImage.source.S>",
          "detail-type": "<$.dynamodb.NewImage.eventType.S>",
          "detail": {
            "eventId": "<$.dynamodb.NewImage.eventId.S>",
            "entity": {
              "id": "<$.dynamodb.NewImage.entityId.S>",
              "type": "<$.dynamodb.NewImage.entityType.S>"
            },
            "metadata": {
              "timestamp": "<$.dynamodb.NewImage.timestamp.S>"
            }
          }
        }

4. IAM Role for Pipe Execution

PipeExecutionRole:
  Type: AWS::IAM::Role
  Properties:
    RoleName: AuditLogPipeExecutionRole
    AssumeRolePolicyDocument:
      Version: "2012-10-17"
      Statement:
        - Effect: Allow
          Principal:
            Service: pipes.amazonaws.com
          Action: sts:AssumeRole
    Policies:
      - PolicyName: AuditLogPipePolicy
        PolicyDocument:
          Version: "2012-10-17"
          Statement:
            # Read from DynamoDB Stream
            - Effect: Allow
              Action:
                - dynamodb:DescribeStream
                - dynamodb:GetRecords
                - dynamodb:GetShardIterator
                - dynamodb:ListStreams
              Resource: !GetAtt AuditLogsTable.StreamArn
            
            # Write events to EventBridge
            - Effect: Allow
              Action:
                - events:PutEvents
              Resource: "*"
            
            # Write logs to CloudWatch
            - Effect: Allow
              Action:
                - logs:CreateLogStream
                - logs:PutLogEvents
              Resource: !GetAtt AuditLogPipeLogGroup.Arn

5. EventBridge Rule to Capture Events

EventBusLogGroup:
  Type: AWS::Logs::LogGroup
  Properties:
    LogGroupName: /aws/events/my-event-bus
    RetentionInDays: 7

EventBusLogGroupPolicy:
  Type: AWS::Logs::ResourcePolicy
  Properties:
    PolicyName: EventBridgeToCWLogsPolicy
    PolicyDocument: !Sub |
      {
        "Version": "2012-10-17",
        "Statement": [
          {
            "Effect": "Allow",
            "Principal": {
              "Service": ["events.amazonaws.com", "delivery.logs.amazonaws.com"]
            },
            "Action": ["logs:CreateLogStream", "logs:PutLogEvents"],
            "Resource": "${EventBusLogGroup.Arn}"
          }
        ]
      }

EventBusLogRule:
  Type: AWS::Events::Rule
  Properties:
    Name: catch-all-rule
    EventBusName: !Ref CustomEventBus
    State: ENABLED
    EventPattern:
      source:
        - prefix: ""  # Catch all events
    Targets:
      - Arn: !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:${EventBusLogGroup}'
        Id: CloudWatchLogTarget

Understanding the InputTemplate

The InputTemplate in the Pipe configuration transforms DynamoDB Stream records into EventBridge events.

DynamoDB Stream Record Format

DynamoDB Stream records use a specific format:

{
  "dynamodb": {
    "NewImage": {
      "eventId": { "S": "evt-123" },
      "source": { "S": "audit.service" },
      "eventType": { "S": "UserCreated" }
    },
    "OldImage": { ... }
  }
}

Notice the type indicators like S (String), N (Number), BOOL (Boolean), etc.

JSONPath Transformation

The InputTemplate uses JSONPath to extract values:

<$.dynamodb.NewImage.eventId.S> extracts the string value
Paths must be wrapped in quotes: "<$.path>"
The result is a properly formatted EventBridge event

Important Notes on EventBridgeEventBusParameters

You cannot use dynamic values in EventBridgeEventBusParameters:

# ❌ THIS DOESN'T WORK
TargetParameters:
  EventBridgeEventBusParameters:
    DetailType: "<$.dynamodb.NewImage.eventType.S>"  # JSONPath not supported here
    Source: "<$.dynamodb.NewImage.source.S>"          # JSONPath not supported here

Instead, include them in InputTemplate:

# ✅ THIS WORKS
TargetParameters:
  InputTemplate: |
    {
      "source": "<$.dynamodb.NewImage.source.S>",
      "detail-type": "<$.dynamodb.NewImage.eventType.S>",
      "detail": { ... }
    }

Key Limitations and Gotchas

1. Global Secondary Index Limitations

DynamoDB GSI can only have one HASH key and one RANGE key:

# ❌ INVALID - Multiple RANGE keys
GlobalSecondaryIndexes:
  - IndexName: MyIndex
    KeySchema:
      - AttributeName: entityType
        KeyType: HASH
      - AttributeName: entityId
        KeyType: RANGE
      - AttributeName: eventType
        KeyType: RANGE  # ❌ ERROR!

2. InputTemplate Must Be Valid JSON

# ❌ INVALID - Missing quotes around JSONPath
"detail": <$.dynamodb.NewImage>

# ✅ VALID - Properly quoted
"detail": "<$.dynamodb.NewImage>"

3. EventBridge Rules Cannot Directly Target CloudWatch Logs

You need a CloudWatch Logs resource policy to allow EventBridge to write logs.

4. Pipe Logging Levels

ERROR: Only errors
INFO: Errors and general info
TRACE: Everything including full payloads

Use TRACE during development, INFO or ERROR in production.

Testing Your Setup

1. Insert Data into DynamoDB

aws dynamodb put-item \
  --table-name audit-logs \
  --item '{
    "PK": {"S": "USER#123"},
    "SK": {"S": "2025-11-29T10:00:00Z"},
    "eventId": {"S": "evt-123"},
    "source": {"S": "user.service"},
    "eventType": {"S": "UserCreated"},
    "entityId": {"S": "user-123"},
    "entityType": {"S": "User"},
    "timestamp": {"S": "2025-11-29T10:00:00Z"}
  }'

2. Check Pipe Logs

aws logs tail /aws/vendedlogs/pipes/audit-log-pipe --follow

3. Check EventBridge Logs

aws logs tail /aws/events/my-event-bus --follow

4. Check Dead Letter Queue

aws sqs receive-message \
  --queue-url https://sqs.REGION.amazonaws.com/ACCOUNT/event-bus-dlq \
  --max-number-of-messages 10

Monitoring and Debugging

CloudWatch Metrics for Pipes

InvocationCount: Number of times the pipe was invoked
InvocationFailedCount: Number of failed invocations
InvocationThrottledCount: Number of throttled invocations
InvocationLatency: Time taken to process events

Common Issues

Events not appearing:

Check DynamoDB Stream is enabled with NEW_AND_OLD_IMAGES
Verify Pipe has correct IAM permissions
Check Pipe CloudWatch logs for errors
Verify EventBridge rule is enabled
Check EventBridge DLQ for failed events

InputTemplate errors:

Ensure JSON is valid
Wrap JSONPath expressions in quotes
Check DynamoDB attribute types match your paths (.S for strings, .N for numbers)

Conclusion

EventBridge Pipes provide a powerful, serverless way to connect DynamoDB Streams to EventBridge without the overhead of Lambda functions. While there are some limitations and gotchas (especially around InputTemplate formatting and IAM permissions), the benefits of reduced latency, cost, and operational complexity make Pipes the preferred solution for DynamoDB-to-EventBridge integrations.

The key takeaway: DynamoDB Streams cannot directly emit to EventBridge—you need EventBridge Pipes to bridge the gap.