DynamoDB Streams and Lambda Triggers

DynamoDB Streams is a fully managed, durable, and ordered data stream that captures changes to items in a DynamoDB table. When you enable streams on a table, DynamoDB records a sequence of stream records each time an item is created, updated, or deleted. Each stream record contains a before-and-after image of the item (configurable) and metadata such as the event type, timestamp, and sequence number. Streams are designed for high-throughput tables and provide exactly-once delivery within a shard.

When you enable a stream on a DynamoDB table, the service partitions the stream into shards. A shard is a container for stream records and is automatically created, managed, and deleted by DynamoDB. Each shard corresponds to a partition of the table. The number of shards scales with the number of table partitions. Stream records are written to shards in the order they occur within that partition. Each shard has a sequence of records identified by a sequence number. The stream endpoint is separate from the table endpoint, and you use the DynamoDB Streams API (e.g., DescribeStream, GetShardIterator, GetRecords) to read records.

Each stream record contains: - eventID: A globally unique identifier for the record. - eventName: INSERT, MODIFY, or REMOVE. - eventVersion: Stream record version (e.g., 1.0). - eventSource: aws:dynamodb. - awsRegion: Region of the table. - dynamodb: Contains: - Keys: Key attributes of the item. - NewImage: Entire item after the change (if StreamViewType is NEW_IMAGE or NEW_AND_OLD_IMAGES). - OldImage: Entire item before the change (if StreamViewType is OLD_IMAGE or NEW_AND_OLD_IMAGES). - SequenceNumber: Monotonically increasing number within a shard. - SizeBytes: Size of the record. - StreamViewType: The view type configured. - userIdentity: Details about the user if the request was made by an IAM user or role.

You choose a StreamViewType when enabling streams: - KEYS_ONLY: Only the key attributes of the modified item. - NEW_IMAGE: The entire item after it was modified. - OLD_IMAGE: The entire item before it was modified. - NEW_AND_OLD_IMAGES: Both the before and after images.

Default: KEYS_ONLY. The exam often tests the trade-offs: NEW_AND_OLD_IMAGES provides the most data but increases stream record size and cost.

You can enable streams via the AWS Management Console, AWS CLI, or CloudFormation. Using the AWS CLI:

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

The response includes the LatestStreamArn and StreamLabel. The Stream Arn format is: arn:aws:dynamodb:<region>:<account-id>:table/<table-name>/stream/<timestamp>

To read stream records manually: 1. Call DescribeStream to get shard IDs. 2. For each shard, call GetShardIterator with the shard ID and iterator type (e.g., TRIM_HORIZON, LATEST, AT_SEQUENCE_NUMBER, AFTER_SEQUENCE_NUMBER). 3. Call GetRecords with the shard iterator to retrieve a batch of records (max 1MB per call). 4. The response includes a NextShardIterator for pagination.

Iterator Types: - TRIM_HORIZON: Start from the oldest unexpired record. - LATEST: Start from the newest record (records added after this call). - AT_SEQUENCE_NUMBER: Start exactly at a given sequence number. - AFTER_SEQUENCE_NUMBER: Start after a given sequence number.

Lambda can be automatically triggered by DynamoDB Streams. You create an event source mapping that connects a Lambda function to a DynamoDB stream. The Lambda service polls the stream on your behalf, reading batches of records and invoking your function synchronously (event invocation).

Key Configuration Parameters: - Batch Size: Number of records to read from the stream per invocation (default 100, max 10,000). - Batch Window: Maximum time to wait for a full batch (default 0 seconds, max 300 seconds). - Starting Position: TRIM_HORIZON or LATEST (similar to iterator type). - Maximum Retry Attempts: Number of times to retry failed invocations (default -1, meaning infinite). - Maximum Record Age: Maximum age of a record before it is discarded (default -1, meaning infinite). - Bisect Batch on Function Error: If true, the batch is split in two and retried separately. - Destination: You can configure an SQS queue or SNS topic for records that are not processed after all retries.

The Lambda function's execution role must have permissions for: - dynamodb:DescribeStream - dynamodb:GetRecords - dynamodb:GetShardIterator - dynamodb:ListStreams

And optionally: - kms:Decrypt if the table uses AWS KMS encryption.

The Lambda event payload is a JSON object with an array of Records. Example:

{
  "Records": [
    {
      "eventID": "1",
      "eventName": "INSERT",
      "eventVersion": "1.0",
      "eventSource": "aws:dynamodb",
      "awsRegion": "us-east-1",
      "dynamodb": {
        "Keys": {
          "Id": {
            "N": "101"
          }
        },
        "NewImage": {
          "Id": {
            "N": "101"
          },
          "Message": {
            "S": "Hello"
          }
        },
        "SequenceNumber": "111",
        "SizeBytes": 26,
        "StreamViewType": "NEW_AND_OLD_IMAGES"
      },
      "eventSourceARN": "arn:aws:dynamodb:us-east-1:123456789012:table/ExampleTable/stream/2015-06-27T00:48:05.899"
    }
  ]
}

When a Lambda invocation fails (returns an error or times out), the Lambda service retries the batch of records. The retry behavior is governed by: - Maximum Retry Attempts: By default, retries are infinite. You can set a finite number. - Maximum Record Age: Once a record's age exceeds this value (in seconds), it is discarded. Default is infinite. - Bisect Batch on Function Error: If enabled, the failed batch is split into two halves, each retried separately. This isolates problematic records. - Destination: You can specify a dead-letter queue or SNS topic for records that fail after all retries.

CloudWatch metrics for DynamoDB Streams include: - ReturnedRecordCount: Number of records returned by GetRecords. - ReturnedBytes: Bytes returned. - UserErrors: Errors from user requests. - SystemErrors: Internal errors.

Lambda monitoring includes: - IteratorAge: How old the last record processed is (in milliseconds). High age indicates the function is falling behind. - Throttles: When Lambda invocations are throttled (concurrency limit).

DynamoDB TTL automatically deletes expired items. These deletions generate stream records with eventName = "REMOVE" (if streams are enabled). The stream record includes the item's content before deletion (if StreamViewType is OLD_IMAGE or NEW_AND_OLD_IMAGES). This is commonly used to trigger cleanup actions after item expiration.

A global e-commerce company needs to replicate order data from a primary DynamoDB table in us-east-1 to a secondary table in eu-west-1 for disaster recovery and low-latency reads. They enable DynamoDB Streams on the primary table with StreamViewType=NEW_AND_OLD_IMAGES. A Lambda function is triggered by the stream, which writes each change to the secondary table using the DynamoDB PutItem or UpdateItem API. The function is deployed in the same region as the primary table to minimize latency. They configure the Lambda function with a batch size of 500 and a batch window of 10 seconds to balance throughput and cost. They monitor the IteratorAge metric; if it exceeds 5 seconds, they increase Lambda concurrency or reduce batch size. They also implement idempotency by using conditional writes (e.g., ConditionExpression: attribute_not_exists(version) or version < new_version) to handle retries. This setup replicates millions of orders per day with sub-second latency.

A social media platform tracks user interactions (likes, comments, shares) stored in a DynamoDB table. They want to update a leaderboard in Amazon ElastiCache in real-time. They enable streams with StreamViewType=NEW_IMAGE on the interactions table. A Lambda function processes each stream record and increments a counter in Redis (e.g., ZINCRBY). They use a batch size of 1000 and a batch window of 5 seconds. The function handles spikes in traffic (e.g., during a viral post) by scaling Lambda concurrency. They configure the Lambda function with a reserved concurrency of 500 to avoid throttling. They also set MaximumRetryAttempts to 3 and MaximumRecordAge to 60 seconds to avoid stale data. If a record fails after retries, they send it to an SQS dead-letter queue for manual inspection. This pipeline processes 10,000 events per second with an average latency of 200ms.

A multinational corporation maintains user profiles in DynamoDB tables in multiple regions for low latency. They use DynamoDB Streams and Lambda to replicate changes across regions. However, conflicts can arise when the same item is updated in two regions simultaneously. They implement a last-writer-wins strategy using a timestamp attribute. The Lambda function checks the timestamp in the NewImage against the current item in the target table. If the incoming change is older, it is ignored. They also use conditional writes to ensure atomic updates. They configure the stream with StreamViewType=NEW_AND_OLD_IMAGES to compare timestamps. They set the Lambda function timeout to 30 seconds and batch size to 100. They monitor CloudWatch metrics for errors and iterator age. This setup handles 50 million user profiles with 99.99% uptime.