Loading Academy Hub...
Build real-time streaming architectures on AWS: Amazon Kinesis Data Streams (shards, enhanced fan-out), Amazon MSK (Managed Streaming for Kafka), Lambda batching windows, and streaming directly to S3.
[!NOTE] AWS provides two primary event streaming services:
- Amazon Kinesis Data Streams: Fully managed, AWS-native serverless streaming with per-shard capacity (1MB/sec write, 2MB/sec read).
- Amazon MSK (Managed Streaming for Apache Kafka): Dedicated Apache Kafka clusters for multi-cloud compatibility and complex open-source ecosystem tools.
1 MB/sec (or 1,000 records/sec) ingress and 2 MB/sec egress (5 reads/sec shared across standard consumers). If multiple independent Spark/Flink/Lambda consumers read the same stream, enable Enhanced Fan-Out (EFO) so each consumer receives a dedicated 2 MB/sec HTTP/2 push pipe per shard.200 MB/s+, but a skewed PartitionKey (e.g., country="US" generating 80% traffic) still trips ProvisionedThroughputExceededException on a single shard. Always salt high-cardinality partition keys (e.g., tenant_id + "_" + hash(event_id) % N) when downstream ordering only matters per sub-key.log.cleanup.policy=compact), sub-10ms tail latency, or multi-cloud portability.Review the production code, company interview Q&As, incident traces, and comprehension quiz inside this lesson.
import base64
import json
import boto3
s3 = boto3.client('s3')
def lambda_handler(event, context):
processed_records = []
failed_records = []
for record in event['Records']:
record_id = record['kinesis']['sequenceNumber']
try:
payload = json.loads(base64.b64decode(record['kinesis']['data']).decode('utf-8'))
# Cleanse and validate payload
if 'transaction_id' in payload:
processed_records.append(payload)
else:
failed_records.append({'itemIdentifier': record_id})
except Exception as e:
failed_records.append({'itemIdentifier': record_id})
# Send clean records to S3 Micro-batch buffer
if processed_records:
s3.put_object(
Bucket='enterprise-lakehouse-prod',
Key=f"bronze/streaming/{event['Records'][0]['kinesis']['partitionKey']}.json",
Body=json.dumps(processed_records)
)
# Return partial batch failure response to Kinesis checkpointing
return {'batchItemFailures': failed_records}com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException: Rate exceeded for shard shardId-000000000000 in stream orders-stream.