10. Streams and event-driven patterns

Chapter 1 said DynamoDB refuses to do anything whose cost it cannot predict — which rules out aggregates, search and fan-out reads. Streams are how you get those anyway: the table emits every change, and something else does the unpredictable work.

What a stream is

   DYNAMODB STREAMS
     • an ordered log of item-level changes
     • 24-hour retention
     • one shard per partition, roughly — ordering is
       guaranteed PER ITEM, not globally
     • read by Lambda, Kinesis Client Library, or the
       Streams API directly

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

   VIEW TYPES
     KEYS_ONLY            just the key of the changed item
     NEW_IMAGE            the item after the change
     OLD_IMAGE            the item before
     NEW_AND_OLD_IMAGES   both — the usual choice, since
                          diffs are what consumers want

A stream record

{
  "eventName": "MODIFY",
  "dynamodb": {
    "Keys":     {"PK": {"S": "ORDER#A917"}, "SK": {"S": "META"}},
    "OldImage": {"status": {"S": "pending"}, "total": {"N": "48.20"}},
    "NewImage": {"status": {"S": "paid"},    "total": {"N": "48.20"}},
    "SequenceNumber": "4421300000000016918",
    "ApproximateCreationDateTime": 1790000000
  }
}

   eventName
     INSERT   a new item
     MODIFY   an existing item changed
     REMOVE   deleted — by DeleteItem, by a
              transaction, or by TTL expiry

The REMOVE case includes TTL deletions, which is what makes TTL a genuine event source rather than just a cleanup mechanism (see below).

Guarantees, precisely

   ✓ EVERY change appears, exactly once, in the stream
   ✓ changes to ONE ITEM appear in the order they
     happened
   ✓ records survive 24 hours
   ✗ NO ordering between different items
   ✗ NO ordering between partitions
   ✗ consumers may see a record MORE THAN ONCE —
     delivery to your consumer is AT LEAST ONCE

Danger

The gap between “records are written exactly once” and “your consumer sees them at least once” is where the bugs live. A Lambda retry, a shard split, or a checkpoint failure re-delivers records. Every stream consumer must be idempotent, keyed on the item key plus the sequence number or a version attribute.

Consuming with Lambda

def handler(event, context):
    for record in event["Records"]:
        if record["eventName"] == "REMOVE":
            handle_delete(record["dynamodb"]["OldImage"])
        else:
            handle_upsert(record["dynamodb"]["NewImage"])

   HOW IT BEHAVES
     • Lambda polls the shards for you
     • batches of up to 10,000 records (you set the
       size and the batching window)
     • ONE concurrent invocation per shard by default;
       ParallelizationFactor raises it to 10, at the
       cost of per-key ordering across those lanes
     • A FAILED BATCH BLOCKS THAT SHARD and retries
       until it succeeds or the records expire

That last line is the operational trap. One poison record can stall a shard for 24 hours:

   MITIGATIONS — configure all three
     BisectBatchOnFunctionError   isolate the bad record
     MaximumRetryAttempts         bound the retries
     DestinationConfig (on-failure SQS/SNS)
                                  send failures somewhere
                                  a human will look

Kinesis Data Streams as an alternative

   DYNAMODB STREAMS          KINESIS DATA STREAMS
   24-hour retention         up to 365 days
   one consumer group,       many independent consumers
     effectively
   exactly-once write to     at-least-once write to
     the stream                the stream (duplicates
                               possible)
   simplest for Lambda       better for fan-out and for
                             feeding analytics

Both can be enabled; the choice is about retention and fan-out, not about the data.

The patterns

Materialised aggregates

   DynamoDB has no SUM. So maintain one:

   order item changes ──► stream ──► Lambda
                                       │
                                       └─► UpdateItem
                                           ADD dailyTotal :delta
                                           on a summary item

Use OldImage and NewImage to compute the delta, and make the update idempotent by tracking the last processed sequence number on the summary item.

Search, analytics and everything DynamoDB will not do

   table ──stream──► Lambda ──► Elasticsearch   (ch)
                            ──► S3 / data lake
                            ──► ClickHouse      (ch)
                            ──► Kafka           (ch)

This is the standard answer to “how do I do full-text search / reporting / joins on my DynamoDB data?”: you do not. You project the data into a system built for it, continuously. Elasticsearch for search, ClickHouse for analytics, Kafka as the bus in front of both.

The transactional outbox

client.transact_write_items(TransactItems=[
  {"Put": {"TableName": "AppData", "Item": order_item}},
  {"Put": {"TableName": "AppData", "Item": {
      "PK": {"S": f"OUTBOX#{event_id}"}, "SK": {"S": "EVENT"},
      "type": {"S": "OrderPlaced"}, "payload": {"S": payload},
      "expiresAt": {"N": str(int(time.time()) + 604800)}}}},
])

The order and the event are written atomically (chapter 9); the stream then delivers the outbox item to whatever publishes it. This is the same pattern the Debezium guide describes for relational databases — here it needs no extra infrastructure, because change capture is already in the table.

Fan-out on write

   a post is created ──► stream ──► Lambda writes one
                                     item per follower's
                                     timeline partition

   reads become a single Query per user; the expensive
   work happens once, asynchronously, at write time.

Keeping denormalised copies in step

Chapter 8 duplicated the customer name onto every order. The stream is how you repair those copies when the name changes — asynchronously, in batches, without the write path paying for it.

Designing a consumer that survives contact with production

def handler(event, context):
    failures = []
    for record in event["Records"]:
        try:
            process(record)
        except Exception:
            failures.append({"itemIdentifier": record["eventID"]})
    # partial batch response: only the failures are retried
    return {"batchItemFailures": failures}

   PARTIAL BATCH RESPONSES
     without them, ONE bad record fails the whole
     batch, and the whole batch is retried — including
     the records that already succeeded.
     with them, only the failures come back.

   ReportBatchItemFailures must be enabled on the
   event source mapping for this to take effect.

Idempotency in the consumer

   THE RECORD CARRIES ITS OWN DEDUPLICATION KEY

     eventID              unique per record
     SequenceNumber       monotonic PER SHARD
     NewImage.version     if you maintain one

   PATTERN: store the last processed sequence number
   on the target item, and skip records at or below
   it. Cheap, and it survives re-delivery.

table.update_item(
    Key=summary_key,
    UpdateExpression="ADD total :delta SET lastSeq = :seq",
    ConditionExpression="attribute_not_exists(lastSeq) OR lastSeq < :seq",
    ExpressionAttributeValues={":delta": delta, ":seq": seq})

Watching a stream

   IteratorAge      how far behind the consumer is.
                    THE metric to alarm on — if it
                    approaches 24 hours, records are
                    about to be lost forever.
   Throttles        the consumer writing back to
                    DynamoDB faster than capacity
                    allows — a feedback loop  (ch 14)
   Errors / Retries per function, per shard

An IteratorAge climbing steadily means the consumer cannot keep up; the fixes are a larger batch window, a higher ParallelizationFactor (at the cost of per-key ordering), or less work per record.

TTL as an event source

aws dynamodb update-time-to-live --table-name AppData \
  --time-to-live-specification "Enabled=true, AttributeName=expiresAt"

   • expiresAt is a UNIX epoch SECOND, as a number
   • deletion is FREE — no write capacity     (ch 6)
   • it is BEST EFFORT: usually within hours, not
     minutes. expired items can still be returned by
     reads until they are removed, so filter on
     expiresAt in queries that must not see them.
   • the delete appears in the STREAM as a REMOVE with
     userIdentity.principalId = "dynamodb.amazonaws.com"

That last detail turns TTL into a scheduler: write an item that expires when something should happen, and let the stream record trigger it. Session expiry, delayed retries, “notify me in 30 days” — all of it without a cron job.

What to take from this chapter

• Streams are change data capture built into the table: every item-level change, 24-hour retention, with NEW_AND_OLD_IMAGES the usual view type.
• Ordering is guaranteed per item, never globally, and consumer delivery is at least once — so every consumer must be idempotent.
• With Lambda, a failed batch blocks its shard. Configure BisectBatchOnFunctionError, MaximumRetryAttempts and an on-failure destination from the start.
• Kinesis Data Streams is the alternative when you need long retention or many independent consumers.
• The core patterns: materialised aggregates, projection into search and analytics systems, the transactional outbox, fan-out on write, and repairing denormalised copies.
• The honest answer to search and reporting is project the data elsewhere continuously — Elasticsearch, ClickHouse, S3, Kafka.
• TTL deletes are free, best-effort in timing, still visible to reads until removed, and appear in the stream — which makes TTL a usable scheduler.

Next: global tables and backups — multi-region replication, last-writer-wins conflict resolution, point-in-time recovery, and exports.