Kinesis Data Streams

good for:
    many independent consumers
    analytics / lakehouse ingestion
    custom stream processing
    fanout to multiple applications
    longer retention than DynamoDB Streams
    consumer reads with checkpoint / replay control

not for:
    simplest one Lambda trigger
    synchronous business transaction
    low-volume app where Streams is enough

choose DynamoDB Streams:
    one or two consumers
    Lambda trigger is enough
    short retention acceptable

choose Kinesis Data Streams:
    data platform needs CDC
    consumer teams are independent
    stream retention and replay matter
    throughput needs explicit management

aws kinesis create-stream \
  --stream-name orders-ddb-cdc \
  --stream-mode-details StreamMode=ON_DEMAND \
  --region ap-east-1

TABLE_ARN="arn:aws:dynamodb:ap-east-1:111122223333:table/orders"
STREAM_ARN="arn:aws:kinesis:ap-east-1:111122223333:stream/orders-ddb-cdc"

aws dynamodb enable-kinesis-streaming-destination \
  --table-name orders \
  --stream-arn "${STREAM_ARN}" \
  --region ap-east-1

aws dynamodb describe-kinesis-streaming-destination \
  --table-name orders \
  --region ap-east-1

pip install boto3
export AWS_REGION=ap-east-1
export STREAM_NAME=orders-ddb-cdc

import json
import os
import time

import boto3


REGION = os.getenv("AWS_REGION", "ap-east-1")
STREAM_NAME = os.getenv("STREAM_NAME", "orders-ddb-cdc")

kinesis = boto3.client("kinesis", region_name=REGION)


def first_shard_id() -> str:
    resp = kinesis.describe_stream_summary(StreamName=STREAM_NAME)
    stream_arn = resp["StreamDescriptionSummary"]["StreamARN"]
    shards = kinesis.list_shards(StreamARN=stream_arn)["Shards"]
    return shards[0]["ShardId"]


def read_latest_records():
    shard_id = first_shard_id()
    iterator = kinesis.get_shard_iterator(
        StreamName=STREAM_NAME,
        ShardId=shard_id,
        ShardIteratorType="LATEST",
    )["ShardIterator"]

    while iterator:
        resp = kinesis.get_records(ShardIterator=iterator, Limit=100)
        for record in resp["Records"]:
            payload = json.loads(record["Data"].decode("utf-8"))
            print(json.dumps(payload, ensure_ascii=False, indent=2))
        iterator = resp.get("NextShardIterator")
        time.sleep(1)


if __name__ == "__main__":
    read_latest_records()

python read_kinesis.py

capacity:
    use on-demand stream mode unless workload is predictable
    monitor write/read throttles
    check downstream consumers before increasing producer volume

consumer design:
    one app per consumer group / checkpoint namespace
    make processing idempotent
    store checkpoint outside business table when using custom consumer

schema:
    version event payload
    keep consumer tolerant to new attributes
    do not assume global order across different partition keys

security:
    table role can write to stream through AWS-managed integration
    consumers get kinesis:DescribeStream, ListShards, GetShardIterator, GetRecords
    encrypt stream with KMS for sensitive data

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kinesis:DescribeStream",
        "kinesis:DescribeStreamSummary",
        "kinesis:ListShards",
        "kinesis:GetShardIterator",
        "kinesis:GetRecords"
      ],
      "Resource": "arn:aws:kinesis:ap-east-1:111122223333:stream/orders-ddb-cdc"
    }
  ]
}

IteratorAge:
    P1 when p95 >= 300s for 10m

Read/Write throttle:
    P1 when Sum > 0 for 5m

Consumer errors:
    P1 when errors > 0 for 5m