Introduction to Scaling Kinesis
Amazon Kinesis Data Streams is a powerful service for collecting, processing, and analyzing real-time streaming data at massive scale. However, moving from a working prototype to a production-ready system requires careful planning around throughput, partitioning, error handling, and cost management. This tutorial walks you through the key concepts and practical steps to scale your Kinesis implementation effectively.
What is Kinesis Data Streams?
Kinesis Data Streams is a managed service that can capture gigabytes of data per second from hundreds of thousands of sources. The data is stored in shards, which are the base throughput units of a stream. Each shard can ingest up to 1 MB per second and 1,000 records per second, and supports up to 2 MB per second of read throughput.
Why Scaling Matters
When you build a prototype, you typically work with a small number of shards and low traffic. In production, your stream must handle unpredictable traffic spikes, maintain ordering guarantees, and process data with low latency. Poorly scaled streams lead to ProvisionedThroughputExceededException errors, increased latency, and potential data loss if producers cannot retry effectively.
Understanding Shards and Partition Keys
The fundamental unit of scalability in Kinesis is the shard. When you create a stream, you specify the number of shards. Records are distributed across shards based on their partition key. Kinesis applies a hash function to the partition key to determine which shard receives the record.
Choosing the Right Partition Key
Your partition key strategy directly impacts how evenly data is distributed across shards. A poor choice can cause hot shards, where one shard receives disproportionate traffic while others remain underutilized.
- High cardinality keys (e.g., user IDs, session IDs) distribute data evenly across shards.
- Low cardinality keys (e.g., country codes, device types) can cause hot shards if one value dominates.
- Composite keys combine multiple fields to increase cardinality while preserving logical grouping.
For example, if you need to group events by customer but some customers generate far more traffic than others, consider appending a random suffix to the partition key:
import hashlib
import random
def get_partition_key(customer_id: str, bucket_count: int = 10) -> str:
"""Distribute a high-traffic customer across multiple partition keys."""
suffix = random.randint(0, bucket_count - 1)
return f"{customer_id}#{suffix}"
# When consuming, you can aggregate across all buckets for a given customer
partition_key = get_partition_key("cust_12345")
This approach preserves ordering within each bucket while spreading load across shards. Consumers must then aggregate data from all buckets to reconstruct the full picture for a customer.
Producer Scaling Strategies
Producers are the applications that put records into Kinesis. At scale, producers must handle retries, batching, and backpressure gracefully.
Using the Kinesis Producer Library (KPL)
The Kinesis Producer Library (KPL) is an advanced library that handles batching, retries, and aggregation automatically. It can significantly increase throughput by combining multiple records into a single Kinesis record, reducing the per-record overhead.
from amazon_kclpy import kcl
import boto3
# Example using the AWS SDK with batching
client = boto3.client('kinesis', region_name='us-east-1')
def put_records_batch(stream_name: str, records: list) -> dict:
"""Batch put records into Kinesis with error handling."""
formatted = [
{
'Data': record['data'],
'PartitionKey': record['partition_key']
}
for record in records
]
response = client.put_records(
Records=formatted,
StreamName=stream_name
)
# Handle failed records
if response['FailedRecordCount'] > 0:
failed_records = []
for idx, record in enumerate(response['Records']):
if 'ErrorCode' in record:
failed_records.append(records[idx])
# Retry failed records with exponential backoff
return retry_failed_records(failed_records, stream_name)
return response
def retry_failed_records(failed_records: list, stream_name: str, max_retries: int = 3):
"""Retry failed records with exponential backoff."""
import time
for attempt in range(max_retries):
time.sleep(2 ** attempt)
try:
return put_records_batch(stream_name, failed_records)
except Exception as e:
if attempt == max_retries - 1:
raise e
Handling ProvisionedThroughputExceededException
When a shard exceeds its write capacity, Kinesis throws a ProvisionedThroughputExceededException. Your producer must handle this with exponential backoff and jitter:
import time
import random
import boto3
from botocore.exceptions import ClientError
client = boto3.client('kinesis')
def put_record_with_retry(stream_name, data, partition_key, max_retries=5):
"""Put a single record with exponential backoff and jitter."""
base_delay = 0.1
for attempt in range(max_retries):
try:
response = client.put_record(
StreamName=stream_name,
Data=data,
PartitionKey=partition_key
)
return response
except ClientError as e:
error_code = e.response['Error']['Code']
if error_code == 'ProvisionedThroughputExceededException':
# Exponential backoff with jitter
delay = base_delay * (2 ** attempt) + random.uniform(0, 0.1)
time.sleep(delay)
else:
raise e
raise Exception(f"Failed after {max_retries} retries")
Consumer Scaling Strategies
Consumers read and process records from Kinesis. There are two main approaches: the Kinesis Client Library (KCL) and enhanced fan-out consumers.
Kinesis Client Library (KCL)
KCL manages consumer instances, handles shard assignment, and provides checkpointing. When you add consumer instances, KCL automatically rebalances shard assignments across the fleet. This is the recommended approach for most production workloads.
from amazon_kclpy import kcl
from amazon_kclpy.app import RecordProcessorBase
class StreamProcessor(RecordProcessorBase):
def initialize(self, shard_id):
self.shard_id = shard_id
self.checkpoint_interval = 60
self.next_checkpoint = time.time() + self.checkpoint_interval
def process_records(self, records, checkpointer):
for record in records:
try:
# Process the record
data = record.get_binary_data()
partition_key = record.partition_key
sequence_number = record.sequence_number
self.handle_record(data, partition_key, sequence_number)
except Exception as e:
# Log error but continue processing other records
print(f"Error processing record: {e}")
continue
# Periodically checkpoint
if time.time() > self.next_checkpoint:
checkpointer.checkpoint()
self.next_checkpoint = time.time() + self.checkpoint_interval
def handle_record(self, data, partition_key, sequence_number):
"""Implement your business logic here."""
pass
def run_consumer():
"""Configure and run a KCL consumer."""
config = kcl.KCLConfig(
stream_name='my-production-stream',
application_name='my-consumer-app',
region='us-east-1',
initial_position='LATEST'
)
processor = StreamProcessor()
kcl_process = kcl.KCLProcess(config, processor)
kcl_process.run()
Enhanced Fan-Out Consumers
By default, all consumers of a stream share the 2 MB per second read throughput per shard. With enhanced fan-out, each consumer gets its own dedicated 2 MB per second read throughput per shard. This is essential when you have multiple downstream consumers that each need high throughput.
import boto3
client = boto3.client('kinesis')
def register_consumer(stream_name: str, consumer_name: str) -> dict:
"""Register an enhanced fan-out consumer."""
response = client.register_stream_consumer(
StreamARN=f'arn:aws:kinesis:us-east-1:123456789012:stream/{stream_name}',
ConsumerName=consumer_name
)
return response['Consumer']
def subscribe_to_shard(consumer_arn: str, shard_id: str):
"""Subscribe to a shard using enhanced fan-out."""
response = client.subscribe_to_shard(
ConsumerARN=consumer_arn,
ShardId=shard_id,
StartingPosition={'Type': 'LATEST'}
)
for event in response['EventStream']:
records = event.get('Records', {}).get('Records', [])
for record in records:
process_record(record)
Monitoring and Auto-Scaling
Production systems require proactive monitoring. CloudWatch provides several key metrics for Kinesis streams:
- WriteProvisionedThroughputExceeded — indicates producers are being throttled.
- ReadProvisionedThroughputExceeded — indicates consumers are being throttled.
- GetRecords.IteratorAgeMilliseconds — indicates how far behind consumers are; high values mean consumers cannot keep up.
- IncomingBytes / IncomingRecords — actual throughput being written to the stream.
Setting Up CloudWatch Alarms
import boto3
cloudwatch = boto3.client('cloudwatch')
def create_throttling_alarm(stream_name: str):
"""Create an alarm for write throttling."""
cloudwatch.put_metric_alarm(
AlarmName=f'{stream_name}-write-throttling',
ComparisonOperator='GreaterThanThreshold',
EvaluationPeriods=2,
MetricName='WriteProvisionedThroughputExceeded',
Namespace='AWS/Kinesis',
Period=60,
Statistic='Sum',
Threshold=1,
AlarmDescription='Alert when write throttling occurs',
Dimensions=[
{
'Name': 'StreamName',
'Value': stream_name
}
],
AlarmActions=[
'arn:aws:sns:us-east-1:123456789012:stream-alerts'
]
)
def create_iterator_age_alarm(stream_name: str):
"""Create an alarm for high iterator age (consumer lag)."""
cloudwatch.put_metric_alarm(
AlarmName=f'{stream_name}-consumer-lag',
ComparisonOperator='GreaterThanThreshold',
EvaluationPeriods=3,
MetricName='GetRecords.IteratorAgeMilliseconds',
Namespace='AWS/Kinesis',
Period=60,
Statistic='Average',
Threshold=300000, # 5 minutes
AlarmDescription='Alert when consumer falls behind by more than 5 minutes',
Dimensions=[
{
'Name': 'StreamName',
'Value': stream_name
}
],
AlarmActions=[
'arn:aws:sns:us-east-1:123456789012:stream-alerts'
]
)
On-Demand vs. Provisioned Mode
Kinesis offers two capacity modes. In provisioned mode, you manually specify the number of shards and must scale them yourself. In on-demand mode, Kinesis automatically scales based on traffic, accommodating traffic spikes without manual intervention.
import boto3
client = boto3.client('kinesis')
def create_on_demand_stream(stream_name: str):
"""Create a stream in on-demand capacity mode."""
client.create_stream(
StreamName=stream_name,
StreamModeDetails={
'StreamMode': 'ON_DEMAND'
}
)
def update_shard_count(stream_name: str, target_shards: int):
"""Scale a provisioned stream to a target shard count."""
client.update_shard_count(
StreamName=stream_name,
TargetShardCount=target_shards,
ScalingType='UNIFORM_SCALING'
)
On-demand mode is ideal for workloads with unpredictable traffic patterns, while provisioned mode offers more control and can be more cost-effective for steady-state workloads.
Best Practices for Production
1. Design for Idempotency
Network failures and retries can cause duplicate records. Consumers should be idempotent, meaning processing the same record multiple times produces the same result. Use sequence numbers or a deduplication cache:
import redis
dedup_cache = redis.Redis(host='localhost', port=6379)
def process_record_idempotent(record):
"""Process a record with deduplication using sequence number."""
seq_num = record['SequenceNumber']
# Check if already processed
if dedup_cache.exists(seq_num):
print(f"Record {seq_num} already processed, skipping")
return
# Process the record
handle_business_logic(record['Data'])
# Mark as processed with TTL of 24 hours
dedup_cache.setex(seq_num, 86400, "1")
2. Implement Proper Checkpointing
Checkpointing allows consumers to resume from where they left off after a failure. Checkpoint after processing batches rather than after every single record to balance reliability with performance:
def process_records_with_checkpoint(records, checkpointer):
"""Process records and checkpoint at appropriate intervals."""
batch_size = 100
processed_count = 0
for record in records:
process_record(record)
processed_count += 1
# Checkpoint every batch_size records
if processed_count % batch_size == 0:
checkpointer.checkpoint(record)
# Final checkpoint for remaining records
if processed_count % batch_size != 0:
checkpointer.checkpoint()
3. Use Dead Letter Queues
Not all records can be processed successfully. Route unprocessable records to a dead letter queue (DLQ) for later investigation rather than blocking the consumer:
import boto3
import json
sqs = boto3.client('sqs')
DLQ_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/stream-dlq'
def process_with_dlq(record, max_attempts=3):
"""Process a record, sending failures to a DLQ."""
try:
result = process_record(record)
return result
except Exception as e:
# Send to DLQ with error context
sqs.send_message(
QueueUrl=DLQ_URL,
MessageBody=json.dumps({
'record_data': record['Data'].decode('utf-8'),
'partition_key': record['PartitionKey'],
'sequence_number': record['SequenceNumber'],
'error': str(e),
'timestamp': record['ApproximateArrivalTimestamp'].isoformat()
}),
MessageAttributes={
'ErrorType': {'StringValue': type(e).__name__, 'DataType': 'String'}
}
)
print(f"Sent failed record to DLQ: {e}")
4. Plan for Resharding
When scaling provisioned streams, resharding operations (split or merge) take time and temporarily reduce throughput on affected shards. Plan resharding during low-traffic periods and avoid frequent resharding operations. After a split or merge, Kinesis needs a few seconds to stabilize before the next operation.
def split_hot_shard(stream_name: str, shard_id: str, new_hash_key: str):
"""Split a shard to increase capacity for a hot partition."""
client.split_shard(
StreamName=stream_name,
ShardToSplit=shard_id,
NewStartingHashKey=new_hash_key
)
# Wait for stream to become active again
waiter = client.get_waiter('stream_exists')
waiter.wait(StreamName=stream_name)
def merge_cold_shards(stream_name: str, shard_id_1: str, shard_id_2: str):
"""Merge two adjacent shards to reduce costs on underutilized partitions."""
client.merge_shards(
StreamName=stream_name,
ShardToMerge=shard_id_1,
AdjacentShardToMerge=shard_id_2
)
5. Secure Your Stream
Use IAM policies to restrict access to your Kinesis streams. Follow the principle of least privilege:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"kinesis:PutRecord",
"kinesis:PutRecords"
],
"Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/production-stream",
"Condition": {
"IpAddress": {
"aws:SourceIp": ["10.0.0.0/8"]
}
}
},
{
"Effect": "Allow",
"Action": [
"kinesis:GetRecords",
"kinesis:GetShardIterator",
"kinesis:DescribeStream"
],
"Resource": "arn:aws:kinesis:us-east-1:123456789012:stream/production-stream"
}
]
}Additionally, enable server-side encryption using AWS KMS to protect sensitive data at rest:
def enable_encryption(stream_name: str, kms_key_id: str):
"""Enable server-side encryption on a Kinesis stream."""
client.start_stream_encryption(
StreamName=stream_name,
EncryptionType='KMS',
KeyId=kms_key_id
)
Cost Optimization
As you scale, costs can grow quickly. Here are strategies to keep costs under control:
- Right-size your shards: Monitor actual throughput and adjust shard counts to match demand. Over-provisioning wastes money.
- Use on-demand mode for variable workloads: On-demand mode charges per GB ingested and per GB retrieved, which can be cheaper than maintaining excess provisioned shards.
- Optimize record size: Smaller records have higher per-record overhead. Use KPL aggregation to combine small records into larger payloads.
- Limit enhanced fan-out consumers: Each enhanced fan-out consumer incurs additional charges. Only use it when shared throughput is insufficient.
- Implement data retention wisely: Kinesis charges per shard-hour for data retention beyond 24 hours. Set retention periods based on actual recovery needs.
Conclusion
Scaling Amazon Kinesis from a prototype to a production system involves much more than simply adding shards. It requires thoughtful partition key design to avoid hot shards, robust producer and consumer patterns that handle failures gracefully, comprehensive monitoring to detect issues before they impact users, and cost-conscious decisions about capacity modes and consumer types. By following the strategies and code patterns outlined in this tutorial—idempotent processing, proper checkpointing, dead letter queues, automated scaling, and security best practices—you can build a resilient, high-throughput streaming pipeline that handles real-world traffic with confidence. Remember that scaling is an iterative process: start with solid fundamentals, monitor continuously, and adjust your architecture as your traffic patterns evolve.