What is Amazon Kinesis?
Amazon Kinesis is a fully managed, real-time data streaming platform provided by AWS. It enables you to ingest, process, analyze, and store streaming data at any scale. The platform consists of four core services that work together or independently to handle different aspects of your data streaming pipeline:
- Kinesis Data Streams — Capture and store streaming data as it arrives, then distribute it to multiple consumers for real-time processing.
- Kinesis Data Firehose — Ingest streaming data, optionally transform it, and deliver it directly to destinations like S3, Redshift, Elasticsearch, or Splunk with zero ongoing management.
- Kinesis Data Analytics — Run SQL queries and Apache Flink programs against streaming data to gain insights in real time.
- Kinesis Video Streams — Capture, process, and store video streams for analytics, machine learning, and playback.
Why Kinesis Matters
🚀 Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →In modern application architectures, the ability to react to data as it arrives—rather than waiting for batch processing windows—is a fundamental competitive advantage. Kinesis matters because it provides the infrastructure backbone for real-time use cases without requiring you to manage complex distributed systems like Apache Kafka clusters. Here is why developers choose Kinesis:
- Zero server management — AWS handles provisioning, patching, and scaling of the underlying infrastructure automatically.
- Elastic throughput scaling — You can increase or decrease shard capacity on demand without downtime, and with the new On-Demand capacity mode, Kinesis Data Streams scales automatically to match your workload.
- Durability and retention — Data is replicated across multiple availability zones and can be retained for up to 365 days (long-term retention), giving consumers plenty of time to catch up after outages.
- Exactly-once processing semantics — With enhanced fan-out consumers and proper checkpointing, you can achieve exactly-once processing guarantees for critical workloads.
- Deep AWS integration — Kinesis connects natively to Lambda, S3, DynamoDB, Redshift, OpenSearch, and more, enabling powerful serverless architectures.
Kinesis Data Streams: Core Concepts
Before diving into setup and code, you need to understand the fundamental building blocks of Kinesis Data Streams:
Shards
A shard is the base throughput unit of a Kinesis stream. Each shard provides:
- Write capacity: 1 MB/sec or 1,000 records/sec (whichever is hit first)
- Read capacity: 2 MB/sec (shared fan-out) or 2 MB/sec per consumer (enhanced fan-out)
The total stream capacity is the sum of all shard capacities. You split or merge shards to adjust throughput as your workload changes.
Partition Keys
Every record you write to Kinesis includes a partition key. Kinesis applies an MD5 hash to this key to determine which shard the record belongs to. By choosing the right partition keys, you can control data distribution across shards—ensuring that related records land on the same shard for ordered processing.
Sequence Numbers
Each record receives a unique, monotonically increasing sequence number within its shard. Consumers use sequence numbers to track progress and implement checkpointing for exactly-once processing.
Consumer Modes
- Shared fan-out — All consumers share the 2 MB/sec per shard read limit. Suitable for lightweight consumers or development.
- Enhanced fan-out — Each registered consumer gets its own 2 MB/sec per shard allocation. Ideal for production workloads with multiple independent consumers.
Setting Up Kinesis Data Streams
Let's walk through a complete setup—from creating your first stream to running a producer and consumer application. We will use the AWS CLI for infrastructure setup and Python (boto3) for application code.
Prerequisites
- An active AWS account with appropriate IAM permissions (we will cover the minimal IAM policy below).
- AWS CLI installed and configured (
aws configure). - Python 3.9+ with boto3 installed (
pip install boto3).
Step 1: Create an IAM Policy and Role
First, create a minimal IAM policy that allows your applications to interact with Kinesis. For a development setup, you can attach this policy directly to your user; for production, attach it to a role assumed by your EC2 instances, Lambda functions, or ECS tasks.
Create a file named kinesis-policy.json:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"kinesis:CreateStream",
"kinesis:DescribeStream",
"kinesis:PutRecord",
"kinesis:PutRecords",
"kinesis:GetRecords",
"kinesis:GetShardIterator",
"kinesis:ListShards",
"kinesis:ListStreams",
"kinesis:UpdateShardCount",
"kinesis:RegisterStreamConsumer",
"kinesis:DeregisterStreamConsumer",
"kinesis:SubscribeToShard",
"kinesis:DescribeStreamConsumer"
],
"Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/*"
}
]
}Replace the account ID and region with your actual values. Attach the policy using:
aws iam put-user-policy --user-name your-username \
--policy-name KinesisDeveloperPolicy \
--policy-document file://kinesis-policy.jsonStep 2: Create a Kinesis Data Stream
Create a stream with an initial shard count. For learning purposes, one shard is sufficient:
aws kinesis create-stream \
--stream-name user-activity-stream \
--shard-count 1 \
--stream-mode-details StreamMode=PROVISIONEDWait for the stream to become active before writing data:
aws kinesis describe-stream --stream-name user-activity-streamWhen the StreamStatus shows ACTIVE, you are ready to proceed.
Step 3: On-Demand Mode (Alternative)
If you prefer automatic capacity management, create an on-demand stream instead:
aws kinesis create-stream \
--stream-name user-activity-stream-ondemand \
--stream-mode-details StreamMode=ON_DEMANDOn-demand streams automatically scale between 4 MB/sec and 200 MB/sec (default quota) without you managing shards. This is ideal for variable or unpredictable workloads.
Building a Producer Application
A producer writes records to your Kinesis stream. The following Python example demonstrates both single-record writes and batched writes with proper error handling.
Basic Producer with PutRecord
import boto3
import json
import uuid
from datetime import datetime
import time
# Initialize the Kinesis client
kinesis_client = boto3.client('kinesis', region_name='us-east-1')
STREAM_NAME = 'user-activity-stream'
def put_single_record(payload: dict, partition_key: str) -> dict:
"""
Write a single record to the Kinesis stream.
Args:
payload: Dictionary to be serialized as JSON
partition_key: String that determines shard distribution
Returns:
Response from the PutRecord API call
"""
data_bytes = json.dumps(payload).encode('utf-8')
response = kinesis_client.put_record(
StreamName=STREAM_NAME,
Data=data_bytes,
PartitionKey=partition_key
)
print(f"Record sent. ShardId: {response['ShardId']}, "
f"SequenceNumber: {response['SequenceNumber']}")
return response
# Example usage: simulate user login events
user_event = {
"event_id": str(uuid.uuid4()),
"event_type": "user_login",
"user_id": "user_42",
"ip_address": "192.168.1.100",
"timestamp": datetime.utcnow().isoformat(),
"device": "mobile"
}
put_single_record(user_event, partition_key=user_event["user_id"])
High-Throughput Producer with PutRecords
For production workloads, batch your writes using PutRecords to reduce API overhead and increase throughput:
import boto3
import json
import uuid
from datetime import datetime
from typing import List, Dict
import time
kinesis_client = boto3.client('kinesis', region_name='us-east-1')
STREAM_NAME = 'user-activity-stream'
def generate_events(count: int) -> List[Dict]:
"""Generate sample events for testing."""
events = []
for i in range(count):
events.append({
"event_id": str(uuid.uuid4()),
"event_type": "page_view",
"user_id": f"user_{i % 100}",
"page_url": f"/products/{i % 50}",
"timestamp": datetime.utcnow().isoformat()
})
return events
def put_records_batch(records: List[Dict]) -> Dict:
"""
Write up to 500 records in a single PutRecords call.
Args:
records: List of dictionaries to write
Returns:
Response dict with failed record count and details
"""
entries = []
for record in records:
data_bytes = json.dumps(record).encode('utf-8')
entries.append({
'Data': data_bytes,
'PartitionKey': record.get('user_id', 'default-key')
})
# Split into batches of 500 (the API limit)
batch_size = 500
total_failed = 0
for i in range(0, len(entries), batch_size):
batch = entries[i:i + batch_size]
response = kinesis_client.put_records(
StreamName=STREAM_NAME,
Records=batch
)
failed_count = response.get('FailedRecordCount', 0)
total_failed += failed_count
if failed_count > 0:
print(f"Batch {i//batch_size}: {failed_count} records failed")
# Implement retry logic here for failed records
for idx, record_result in enumerate(response['Records']):
if record_result.get('ErrorCode'):
print(f" Record {idx} failed: {record_result['ErrorCode']} - "
f"{record_result.get('ErrorMessage')}")
print(f"Batch {i//batch_size}: Sent {len(batch)} records, "
f"{len(batch) - failed_count} succeeded")
return {"total_records": len(records), "total_failed": total_failed}
# Generate and send 1000 events
events = generate_events(1000)
result = put_records_batch(events)
print(f"\nFinal result: {result['total_records']} records attempted, "
f"{result['total_failed']} failed")
Error Handling and Retry Logic
Kinesis can throttle your writes if you exceed shard limits. Implement exponential backoff to handle throughput exceeded exceptions gracefully:
import boto3
import json
import time
from botocore.exceptions import ClientError
kinesis_client = boto3.client('kinesis', region_name='us-east-1')
STREAM_NAME = 'user-activity-stream'
def put_record_with_retry(payload: dict, partition_key: str, max_retries: int = 5) -> dict:
"""
Write a record with exponential backoff retry logic.
"""
data_bytes = json.dumps(payload).encode('utf-8')
base_delay = 0.5 # seconds
for attempt in range(max_retries):
try:
response = kinesis_client.put_record(
StreamName=STREAM_NAME,
Data=data_bytes,
PartitionKey=partition_key
)
return response
except ClientError as e:
error_code = e.response['Error']['Code']
if error_code == 'ProvisionedThroughputExceededException':
delay = base_delay * (2 ** attempt) # Exponential backoff
print(f"Throughput exceeded. Retrying in {delay:.2f}s "
f"(attempt {attempt + 1}/{max_retries})")
time.sleep(delay)
else:
# Non-throttling error — re-raise
raise
raise Exception(f"Failed to write record after {max_retries} attempts")
# Example with retry logic
event = {
"event_id": str(uuid.uuid4()),
"event_type": "purchase",
"user_id": "user_99",
"amount": 49.99,
"timestamp": datetime.utcnow().isoformat()
}
try:
response = put_record_with_retry(event, partition_key=event["user_id"])
print(f"Successfully wrote record: {response['SequenceNumber']}")
except Exception as e:
print(f"Fatal error after all retries: {e}")
Building a Consumer Application
Consumers read records from Kinesis and process them. You have two primary patterns: polling with GetRecords (shared fan-out) and push-based with enhanced fan-out using SubscribeToShard.
Polling Consumer with GetRecords
This is the classic consumer model. You obtain a shard iterator, then poll for records in a loop:
import boto3
import json
import time
from typing import List, Dict
kinesis_client = boto3.client('kinesis', region_name='us-east-1')
STREAM_NAME = 'user-activity-stream'
def get_shard_iterator(shard_id: str, iterator_type: str = 'LATEST') -> str:
"""
Obtain a shard iterator.
Iterator types:
- AT_SEQUENCE_NUMBER: Start at a specific sequence number
- AFTER_SEQUENCE_NUMBER: Start right after a specific sequence number
- TRIM_HORIZON: Start at the oldest record available (within retention window)
- LATEST: Start just after the most recent record (new data only)
"""
response = kinesis_client.get_shard_iterator(
StreamName=STREAM_NAME,
ShardId=shard_id,
ShardIteratorType=iterator_type
)
return response['ShardIterator']
def process_records(records: List[Dict]) -> None:
"""Process a batch of records — replace with your business logic."""
for record in records:
# Decode the data from base64-encoded bytes
data_bytes = record['Data']
# boto3 automatically decodes base64 for GetRecords
payload = json.loads(data_bytes.decode('utf-8') if isinstance(data_bytes, bytes) else data_bytes)
print(f"Processing event {payload.get('event_id')}: "
f"{payload.get('event_type')} by user {payload.get('user_id')}")
print(f" SequenceNumber: {record['SequenceNumber']}")
print(f" PartitionKey: {record['PartitionKey']}")
def poll_stream(iterator_type: str = 'TRIM_HORIZON', poll_interval: float = 1.0):
"""
Continuously poll a Kinesis stream for new records.
"""
# List all shards
shards_response = kinesis_client.list_shards(StreamName=STREAM_NAME)
shard_ids = [shard['ShardId'] for shard in shards_response['Shards']]
# Track iterators per shard
iterators = {}
for shard_id in shard_ids:
iterators[shard_id] = get_shard_iterator(shard_id, iterator_type)
checkpoint = {} # shard_id -> last sequence number processed
print(f"Starting to poll {len(shard_ids)} shard(s)...")
while True:
for shard_id, iterator in iterators.items():
try:
response = kinesis_client.get_records(
ShardIterator=iterator,
Limit=100 # Max 10,000 records per call
)
records = response.get('Records', [])
if records:
print(f"\nReceived {len(records)} record(s) from shard {shard_id}")
process_records(records)
# Update checkpoint
last_seq = records[-1]['SequenceNumber']
checkpoint[shard_id] = last_seq
# Update iterator for next poll
iterators[shard_id] = response['NextShardIterator']
# Sleep if iterator is exhausted (happens when stream is empty)
if response.get('MillisBehindLatest', 0) == 0 and not records:
time.sleep(poll_interval)
except Exception as e:
print(f"Error polling shard {shard_id}: {e}")
# Refresh iterator on error
time.sleep(5)
last_seq = checkpoint.get(shard_id)
if last_seq:
iterators[shard_id] = get_shard_iterator(
shard_id, 'AFTER_SEQUENCE_NUMBER', last_seq)
else:
iterators[shard_id] = get_shard_iterator(shard_id, 'LATEST')
time.sleep(poll_interval)
# Run the poller (press Ctrl+C to stop)
if __name__ == "__main__":
poll_stream(iterator_type='TRIM_HORIZON')
Enhanced Fan-Out Consumer with SubscribeToShard
Enhanced fan-out provides dedicated throughput per consumer and lower latency. It uses HTTP/2 streaming with a persistent connection:
import boto3
import json
kinesis_client = boto3.client('kinesis', region_name='us-east-1')
STREAM_NAME = 'user-activity-stream'
CONSUMER_NAME = 'my-enhanced-consumer'
# Step 1: Register an enhanced fan-out consumer (one-time setup)
def register_consumer():
try:
response = kinesis_client.register_stream_consumer(
StreamARN=f'arn:aws:kinesis:us-east-1:123456789012:stream/{STREAM_NAME}',
ConsumerName=CONSUMER_NAME
)
consumer_arn = response['Consumer']['ConsumerARN']
print(f"Registered consumer: {consumer_arn}")
return consumer_arn
except kinesis_client.exceptions.ResourceInUseException:
# Consumer already exists — retrieve its ARN
consumers = kinesis_client.list_stream_consumers(
StreamARN=f'arn:aws:kinesis:us-east-1:123456789012:stream/{STREAM_NAME}'
)
for consumer in consumers['Consumers']:
if consumer['ConsumerName'] == CONSUMER_NAME:
return consumer['ConsumerARN']
raise
consumer_arn = register_consumer()
# Step 2: Subscribe to a shard and process events
def subscribe_and_process(shard_id: str):
response = kinesis_client.subscribe_to_shard(
ConsumerARN=consumer_arn,
ShardId=shard_id,
StartingPosition={'Type': 'TRIM_HORIZON'}
)
# The response is an event stream (HTTP/2 streaming body)
event_stream = response['EventStream']
for event in event_stream:
if 'SubscribeToShardEvent' in event:
subscribe_event = event['SubscribeToShardEvent']
records = subscribe_event['Records']
for record in records:
payload = json.loads(record['Data'].decode('utf-8'))
print(f"[Enhanced Fan-Out] Event: {payload.get('event_type')} "
f"from user {payload.get('user_id')}")
print(f" SequenceNumber: {record['SequenceNumber']}")
# ContinuationSequenceNumber helps with resuming if needed
continuation_seq = subscribe_event.get('ContinuationSequenceNumber')
elif 'ResourceNotFoundException' in event:
print("Shard no longer exists — stopping subscription.")
break
# List shards and subscribe
shards = kinesis_client.list_shards(StreamName=STREAM_NAME)
for shard in shards['Shards']:
shard_id = shard['ShardId']
print(f"Subscribing to shard {shard_id}...")
subscribe_and_process(shard_id)
Setting Up Kinesis Data Firehose
Kinesis Data Firehose is the simplest way to deliver streaming data to destinations. It handles batching, compression, encryption, and retries automatically. Here is a complete setup that delivers data to an S3 bucket with JSON formatting and Parquet conversion.
Step 1: Create an S3 Bucket
aws s3 mb s3://my-firehose-delivery-bucket-2024 --region us-east-1Step 2: Create an IAM Role for Firehose
Firehose needs permission to write to S3 and optionally to invoke Lambda for transformations. Create a trust policy document firehose-trust.json:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "firehose.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}Create the role:
aws iam create-role \
--role-name FirehoseS3DeliveryRole \
--assume-role-policy-document file://firehose-trust.jsonAttach the necessary permissions:
aws iam put-role-policy \
--role-name FirehoseS3DeliveryRole \
--policy-name FirehoseS3Policy \
--policy-document '{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"s3:PutObject",
"s3:PutObjectAcl"
],
"Resource": "arn:aws:s3:::my-firehose-delivery-bucket-2024/*"
},
{
"Effect": "Allow",
"Action": [
"s3:ListBucket"
],
"Resource": "arn:aws:s3:::my-firehose-delivery-bucket-2024"
}
]
}'Step 3: Create the Firehose Delivery Stream
aws firehose create-delivery-stream \
--delivery-stream-name user-activity-firehose \
--delivery-stream-type DirectPut \
--s3-destination-configuration \
'{
"RoleARN": "arn:aws:iam::123456789012:role/FirehoseS3DeliveryRole",
"BucketARN": "arn:aws:s3:::my-firehose-delivery-bucket-2024",
"Prefix": "user-activity/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/",
"ErrorOutputPrefix": "errors/!{firehose:error-output-type}/",
"BufferingHints": {
"SizeInMBs": 5,
"IntervalInSeconds": 300
},
"CompressionFormat": "Snappy",
"EncryptionConfiguration": {
"NoEncryptionConfig": "NoEncryption"
},
"CloudWatchLoggingOptions": {
"Enabled": true,
"LogGroupName": "/aws/firehose/user-activity-firehose",
"LogStreamName": "s3-delivery"
}
}'Writing to Firehose
You can write to Firehose directly using the PutRecord or PutRecordBatch APIs. The Firehose API is nearly identical to Kinesis Data Streams but without shard management:
import boto3
import json
import uuid
from datetime import datetime
firehose_client = boto3.client('firehose', region_name='us-east-1')
DELIVERY_STREAM = 'user-activity-firehose'
def send_to_firehose(events: list):
"""Send a batch of events to Firehose delivery stream."""
records = []
for event in events:
data_bytes = json.dumps(event).encode('utf-8')
records.append({'Data': data_bytes})
response = firehose_client.put_record_batch(
DeliveryStreamName=DELIVERY_STREAM,
Records=records
)
failed = response.get('FailedPutCount', 0)
if failed > 0:
print(f"Warning: {failed} records failed to deliver")
# Implement retry for failed records
print(f"Sent {len(records)} records to Firehose, {failed} failed")
return response
# Example batch
events = [
{
"event_id": str(uuid.uuid4()),
"event_type": "page_view",
"user_id": "user_100",
"timestamp": datetime.utcnow().isoformat()
}
for _ in range(50)
]
send_to_firehose(events)
Kinesis Data Analytics: Real-Time SQL Processing
Kinesis Data Analytics lets you run SQL queries on streaming data without provisioning servers. You define an analytics application that reads from a Kinesis stream (or Firehose), executes SQL, and outputs results to another stream or Firehose delivery stream.
Creating an Analytics Application
First, ensure you have both a source stream and a destination stream (or Firehose). Then create the analytics application:
aws kinesisanalyticsv2 create-application \
--application-name user-activity-analytics \
--runtime-environment SQL-1 \
--service-execution-role arn:aws:iam::123456789012:role/KinesisAnalyticsRole \
--application-configuration '{
"SqlApplicationConfiguration": {
"Inputs": [{
"NamePrefix": "SOURCE_SQL_STREAM",
"KinesisStreamsInput": {
"ResourceARN": "arn:aws:kinesis:us-east-1:123456789012:stream/user-activity-stream"
},
"InputSchema": {
"RecordColumns": [
{"Name": "event_id", "SqlType": "VARCHAR(36)", "Mapping": "$.event_id"},
{"Name": "event_type", "SqlType": "VARCHAR(32)", "Mapping": "$.event_type"},
{"Name": "user_id", "SqlType": "VARCHAR(64)", "Mapping": "$.user_id"},
{"Name": "timestamp", "SqlType": "TIMESTAMP", "Mapping": "$.timestamp"}
],
"RecordFormat": {
"RecordFormatType": "JSON"
},
"RecordEncoding": "UTF-8"
}
}],
"Outputs": [{
"Name": "DESTINATION_SQL_STREAM",
"KinesisFirehoseOutput": {
"ResourceARN": "arn:aws:firehose:us-east-1:123456789012:deliverystream/user-activity-firehose"
},
"DestinationSchema": {
"RecordFormatType": "JSON"
}
}]
}
}'Example SQL for Real-Time Aggregation
Here is a sample SQL query that creates a sliding window aggregation to count events per user over 5-minute intervals:
-- Create a pump to continuously output results
CREATE OR REPLACE PUMP user_activity_pump AS
INSERT INTO DESTINATION_SQL_STREAM
SELECT STREAM
user_id,
event_type,
COUNT(*) AS event_count,
COUNT(DISTINCT event_id) AS unique_events,
FLOOR(SOURCE_SQL_STREAM.ROWTIME TO MINUTE) AS window_start
FROM SOURCE_SQL_STREAM
GROUP BY
user_id,
event_type,
FLOOR(SOURCE_SQL_STREAM.ROWTIME TO MINUTE),
STEP(FLOOR(SOURCE_SQL_STREAM.ROWTIME TO MINUTE) BY INTERVAL '5' MINUTE)
HAVING COUNT(*) > 2; -- Only output if more than 2 events in windowComplete Architecture: Kinesis + Lambda + DynamoDB
One of the most powerful patterns is connecting Kinesis to Lambda for serverless stream processing. Here is a complete example that reads from Kinesis, enriches events, and writes to DynamoDB.
Lambda Function Code
import boto3
import json
import base64
import os
from datetime import datetime
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['DYNAMODB_TABLE'])
def lambda_handler(event, context):
"""
Lambda handler for Kinesis stream events.
The event contains a batch of records from the stream.
Each record's 'data' field is base64-encoded.
"""
processed_count = 0
failed_records = []
for record in event['Records']:
try:
# Decode the Kinesis data
kinesis_data = record['kinesis']
payload_bytes = base64.b64decode(kinesis_data['data'])
payload = json.loads(payload_bytes)
# Extract metadata
partition_key = kinesis_data['partitionKey']
sequence_number = kinesis_data['sequenceNumber']
# Enrich the event
payload['processed_at'] = datetime.utcnow().isoformat()
payload['source_shard'] = kinesis_data['shardID']
# Write to DynamoDB
table.put_item(Item={
'event_id': payload['event_id'],
'event_type': payload['event_type'],
'user_id': payload['user_id'],
'timestamp': payload['timestamp'],
'processed_at': payload['processed_at'],
'full_payload': json.dumps(payload)
})
processed_count += 1
except Exception as e:
print(f"Error processing record: {e}")
failed_records.append(record['eventID'])
print(f"Processed {processed_count} records successfully, "
f"{len(failed_records)} failed")
# Return batch item failures for partial batch retry
return {
"batchItemFailures": [
{"itemIdentifier": failed_id} for failed_id in failed_records
]
}
CloudFormation Snippet for Lambda Event Source Mapping
# Snippet for your CloudFormation or CDK template
LambdaKinesisEventSource:
Type: AWS::Lambda::EventSourceMapping
Properties:
EventSourceArn: !Ref KinesisStreamArn
FunctionName: !Ref LambdaFunctionArn
StartingPosition: TRIM_HORIZON
BatchSize: 100
MaximumBatchingWindowInSeconds: 30
ParallelizationFactor: 1
MaximumRetryAttempts: 3
DestinationConfig:
OnFailure:
Destination: !Ref DeadLetterQueueArn
Monitoring and Observability
Kinesis exposes critical metrics through CloudWatch that you must monitor to ensure stream health:
Key Metrics to Watch
- UserRecords PutRecords.Success — Track write throughput and detect producer issues.
- WriteProvisionedThroughputExceeded — Indicates throttling; scale up shards or switch to on-demand mode.
- GetRecords.Success / GetRecords.Records — Monitor consumer activity.
- ReadProvisionedThroughputExceeded — Consumer throttling; switch to enhanced fan-out.
- MillisBehindLatest — Consumer lag; the primary indicator of processing delay. Alert if this exceeds your SLA threshold.
- IteratorAgeMilliseconds — Age of the oldest record not yet processed. Critical for detecting stalled consumers.
CloudWatch Alarm for Consumer Lag
aws cloudwatch put-metric-alarm \
--alarm-name kinesis-consumer-lag \
--alarm-description "Alert when consumer falls behind by more than 5 minutes" \
--namespace AWS/Kinesis \
--metric-name MillisBehindLatest \
--statistic Maximum \
--period 300 \
--evaluation-periods 2 \
--threshold 300000 \
--comparison-operator GreaterThanThreshold \
--dimensions Name=StreamName,Value=user-activity-stream \
--alarm-actions arn:aws:sns:us-east-1:123456789012:ops-alertsBest Practices
1. Choose the Right Capacity Mode
Use on-demand mode for variable or unpredictable workloads, new applications where you do not yet know the throughput profile, or when operational simplicity is a priority. Use provisioned mode when you have predictable, steady-state throughput and want cost optimization (provisioned can be cheaper at consistent high volume).
2. Design Your Partition Key Carefully
The partition key determines both shard distribution and record ordering. A good partition key distributes records evenly across shards while keeping related records together. Avoid using timestamps alone—they tend to create hot shards during peak periods. Instead, combine a high-cardinality identifier (