PreviewLab7 hours

Streams & Shards

Provision Kinesis streams, write and read records, manage shards, and consume with Lambda and enhanced fan-out.

Kinesis vs SQS/SNS mindset shift: records are never deleted by consumers. The stream is an immutable log. Consumers track their own position via iterators or checkpoints. Multiple independent consumers can read the same data without interfering with each other.

Prerequisites

0 of 5 checked

Stream Basics: Create, Put, Get

Goal

Create a Kinesis stream. Put records into it. Retrieve them using the shard iterator model. Understand why retrieval feels different from SQS, and why that's intentional.

Estimated time: 1–2 hours

Create a stream

What's happening here

A Kinesis stream is an ordered, immutable log divided into shards. Each shard is an independent unit of capacity: 1 MB/s write, 2 MB/s read, up to 1000 records/s write. With 1 shard, your entire stream is capped at those limits. Scaling = adding shards. StreamModeDetails=ON_DEMAND removes the need to pre-provision shard count. AWS scales shards automatically based on throughput. The alternative is PROVISIONED mode where you specify shard count explicitly. ON_DEMAND is better for learning (no capacity planning); PROVISIONED is better for cost-predictable production workloads.

Kinesis vs SQS: SQS is a queue (messages are consumed and deleted. Kinesis is a log) records persist for the retention period regardless of how many consumers read them. Multiple consumers can independently read the same records from the same stream.

bash

mkdir kinesis-lab && cd kinesis-lab

export STREAM_NAME=kinesis-lab-stream

aws kinesis create-stream \
  --stream-name $STREAM_NAME \
  --stream-mode-details StreamMode=ON_DEMAND \
  --region $AWS_REGION

# Wait for stream to become ACTIVE
aws kinesis wait stream-exists --stream-name $STREAM_NAME --region $AWS_REGION

export STREAM_ARN=$(aws kinesis describe-stream-summary \
  --stream-name $STREAM_NAME \
  --region $AWS_REGION \
  --query 'StreamDescriptionSummary.StreamARN' --output text)

echo "Stream ARN: $STREAM_ARN"

Inspect the stream:

bash

aws kinesis describe-stream-summary \
  --stream-name $STREAM_NAME \
  --region $AWS_REGION \
  --query 'StreamDescriptionSummary.{Status:StreamStatus,Mode:StreamModeDetails.StreamMode,Shards:OpenShardCount,RetentionHours:RetentionPeriodHours}'

Checkpoint: Status: ACTIVE, RetentionPeriodHours: 24.

Put records

What's happening here: Every record requires a partition key: an arbitrary string you choose. Kinesis hashes it (MD5) and maps it to a shard. Records with the same partition key always go to the same shard: this is how ordering is guaranteed. Records with different partition keys may land on different shards.

put-record is synchronous, it returns only after the record is durably written to the stream across multiple AZs. The response includes ShardId (which shard received it) and SequenceNumber (the record's unique, monotonically increasing position within that shard).

put-records (batch) sends up to 500 records in one API call. Each record has its own success/failure in the response (check FailedRecordCount. A batch with 3 failed records returns 200) you must check per-record status.

bash

# Put a single record
aws kinesis put-record \
  --stream-name $STREAM_NAME \
  --partition-key "customer-001" \
  --data "$(echo '{"event": "invoice.created", "id": "inv-001", "amount": 9900}' | base64)" \
  --region $AWS_REGION

# Put records with different partition keys
for i in $(seq 1 5); do
  aws kinesis put-record \
    --stream-name $STREAM_NAME \
    --partition-key "customer-00$i" \
    --data "$(echo "{\"event\": \"invoice.created\", \"id\": \"inv-00$i\", \"amount\": $((i * 1000))}" | base64)" \
    --region $AWS_REGION \
    --query '{ShardId:ShardId,SequenceNumber:SequenceNumber}'
done

Batch put:

bash

aws kinesis put-records \
  --stream-name $STREAM_NAME \
  --records \
    'Data='"$(echo '{"id":"inv-010","amount":5000}' | base64)"',PartitionKey=customer-010' \
    'Data='"$(echo '{"id":"inv-011","amount":7500}' | base64)"',PartitionKey=customer-011' \
    'Data='"$(echo '{"id":"inv-012","amount":3200}' | base64)"',PartitionKey=customer-012' \
  --region $AWS_REGION | jq '{FailedRecordCount: .FailedRecordCount, Records: [.Records[] | {ShardId,SequenceNumber}]}'

FailedRecordCount: 0. Each record shows ShardId and SequenceNumber. Notice records with different partition keys may land on different shards in PROVISIONED mode, in ON_DEMAND with one effective shard they likely share one.

Checkpoint: records written, FailedRecordCount: 0.

Get records via the iterator model

What's happening here: Reading from Kinesis is a two-step process:

Get a shard iterator: a cursor pointing to a position in a shard
Call get-records using that iterator, returns up to 10 MB of records and a next iterator for the following position

Iterator types:

TRIM_HORIZON: start from the oldest record in the shard (full replay)
LATEST: start from new records only (skip existing)
AT_SEQUENCE_NUMBER / AFTER_SEQUENCE_NUMBER: precise position
AT_TIMESTAMP: start from a specific time

Unlike SQS, getting records does not remove them. The iterator just moves forward. The same records can be read by a second consumer from scratch by getting a new TRIM_HORIZON iterator.

Shard iterator expiry

an iterator expires after 5 minutes of inactivity. If you don't call get-records within 5 minutes, you must get a new iterator.

bash

# List shards first
aws kinesis list-shards \
  --stream-name $STREAM_NAME \
  --region $AWS_REGION \
  --query 'Shards[*].{ShardId:ShardId,Start:SequenceNumberRange.StartingSequenceNumber}'

# Get the first shard ID
export SHARD_ID=$(aws kinesis list-shards \
  --stream-name $STREAM_NAME \
  --region $AWS_REGION \
  --query 'Shards[0].ShardId' --output text)

echo "Shard ID: $SHARD_ID"

# Get a TRIM_HORIZON iterator (read from the beginning)
export ITERATOR=$(aws kinesis get-shard-iterator \
  --stream-name $STREAM_NAME \
  --shard-id $SHARD_ID \
  --shard-iterator-type TRIM_HORIZON \
  --region $AWS_REGION \
  --query 'ShardIterator' --output text)

echo "Iterator (first 80 chars): ${ITERATOR:0:80}..."

Read records:

bash

# Get records using the iterator
RESPONSE=$(aws kinesis get-records \
  --shard-iterator $ITERATOR \
  --limit 10 \
  --region $AWS_REGION)

# Decode and display records
echo $RESPONSE | jq '.Records[] | {SequenceNumber: .SequenceNumber, PartitionKey: .PartitionKey, Data: (.Data | @base64d)}'

# How many records returned?
echo "Records returned: $(echo $RESPONSE | jq '.Records | length')"  

# Milliseconds behind latest (lag indicator)
echo "MillisBehindLatest: $(echo $RESPONSE | jq '.MillisBehindLatest')"

# Save the next iterator
export NEXT_ITERATOR=$(echo $RESPONSE | jq -r '.NextShardIterator')

MillisBehindLatest: 0 means you've caught up to the tip of the shard. This is the Kinesis equivalent of queue depth, but measured in time lag, not message count. NextShardIterator is your cursor for the next get-records call.

Read again with the next iterator (should return empty):

bash

RESPONSE2=$(aws kinesis get-records \
  --shard-iterator $NEXT_ITERATOR \
  --limit 10 \
  --region $AWS_REGION)

echo "Records: $(echo $RESPONSE2 | jq '.Records | length')"
echo "MillisBehindLatest: $(echo $RESPONSE2 | jq '.MillisBehindLatest')"

0 records, MillisBehindLatest: 0. No more records. But if you put another record now and use the next iterator, it will appear.

Checkpoint: read all records, observed MillisBehindLatest: advanced the iterator.

Streams & Shards

Prerequisites

Stream Basics: Create, Put, Get

Create a stream

Inspect the stream:

Put records

Batch put:

Get records via the iterator model

Read records:

Read again with the next iterator (should return empty):

Replay from the beginning (the key Kinesis superpower)