Streams

Links#

1. Important Points#

DynamoDB Streams 是 DynamoDB table 的 change log。它适合把 table mutation 变成异步事件，但不适合做长期日志系统。

common use cases:
    update search index
    invalidate cache
    maintain aggregate / materialized view
    sync to another table / region / system
    trigger async workflow after order status changed
    audit important item changes
    publish domain events to EventBridge / SNS / SQS

not good for:
    long-term event retention
    replay from months ago
    high-fanout consumer group design
    replacing online reads

Stream record order:

same item:
    change order is preserved

different partition keys:
    do not rely on global order

2. Stream View Type#

View Type	Contains	Use Case
`KEYS_ONLY`	changed item keys only	lightweight notification, consumer reads latest item
`NEW_IMAGE`	item after change	index update, cache refresh
`OLD_IMAGE`	item before change	audit delete/update old value
`NEW_AND_OLD_IMAGES`	before and after	diff, audit, sync with delete/update context

Default recommendation:

NEW_AND_OLD_IMAGES:
    best for correctness
    more payload size
    easier to handle update/delete/audit

NEW_IMAGE:
    good when only current state matters

Enable stream:

aws dynamodb update-table \
  --table-name customer_order_status \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
  --region ap-northeast-1

3. Lambda Consumer#

Minimal Lambda handler:

import json
from boto3.dynamodb.types import TypeDeserializer

deserializer = TypeDeserializer()


def decode_image(image):
    if not image:
        return None
    return {k: deserializer.deserialize(v) for k, v in image.items()}


def lambda_handler(event, context):
    for record in event["Records"]:
        event_name = record["eventName"]  # INSERT / MODIFY / REMOVE
        keys = decode_image(record["dynamodb"].get("Keys"))
        old_item = decode_image(record["dynamodb"].get("OldImage"))
        new_item = decode_image(record["dynamodb"].get("NewImage"))
  
        print(json.dumps({
            "event_name": event_name,
            "keys": keys,
            "customer_id": (new_item or old_item or {}).get("customer_id"),
            "order_id": (new_item or old_item or {}).get("order_id"),
            "status": (new_item or old_item or {}).get("status"),
            "old_item": old_item,
            "new_item": new_item,
        }, default=str))

        # example:
        #   INSERT -> publish order-created event
        #   MODIFY -> publish order-status-changed event
        #   REMOVE -> remove order from read model

Lambda execution role policy:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ReadCustomerOrderStatusStream",
      "Effect": "Allow",
      "Action": [
        "dynamodb:DescribeStream",
        "dynamodb:GetRecords",
        "dynamodb:GetShardIterator",
        "dynamodb:ListStreams"
      ],
      "Resource": "arn:aws:dynamodb:ap-northeast-1:123456789012:table/customer_order_status/stream/*"
    },
    {
      "Sid": "WriteLambdaLogs",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:ap-northeast-1:123456789012:*"
    }
  ]
}

If event source mapping fails with Cannot access stream, attach this policy to the Lambda execution role used by customer-order-status-stream-worker.

Create event source mapping:

STREAM_ARN="$(aws dynamodb describe-table \
  --table-name customer_order_status \
  --region ap-northeast-1 \
  --query 'Table.LatestStreamArn' \
  --output text)"

aws lambda create-event-source-mapping \
  --function-name customer-order-status-stream-worker \
  --event-source-arn "${STREAM_ARN}" \
  --starting-position LATEST \
  --batch-size 100 \
  --maximum-batching-window-in-seconds 5 \
  --region ap-northeast-1

4. Best Practices#

idempotency:
    consumer must handle duplicate delivery
    use eventID / sequenceNumber / business version as dedupe key

batch failure:
    enable partial batch response when possible
    do not fail entire batch for one poison record
    send failed records to DLQ / destination when supported

latency:
    monitor IteratorAge
    scale Lambda concurrency only after checking downstream capacity

payload:
    choose stream view type intentionally
    large items increase Lambda payload and processing cost

side effects:
    use conditional writes when stream updates another DynamoDB table
    avoid endless loop if consumer writes back to same table

5. Monitoring#

Signal	Why
Lambda `IteratorAge`	consumer lag
Lambda `Errors`	handler failure
Lambda `Throttles`	reserved concurrency or account limit
Lambda `Duration`	processing cost and timeout risk
DLQ messages	poison events or downstream failure

Alert defaults:

IteratorAge:
    P1 when >= 300s for 10m

Errors:
    P1 when > 0 for 5m on critical stream

DLQ:
    P1 when visible messages >= 1

6. Decision#

use DynamoDB Streams when:
    consumer is close to the table
    Lambda trigger is enough
    24h-style short retention is acceptable
    no independent fanout stream platform is required

use Kinesis Data Streams when:
    multiple applications consume the same changes
    consumers need independent read positions
    retention / throughput / fanout control is important