Kinesis Data Streams

Kinesis Data Streams

Kinesis Data Streams (KDS) is a scalable real-time data streaming service.

Core Concepts

Create Stream

# Create stream
aws kinesis create-stream \
  --stream-name my-stream \
  --shard-count 4

# Describe stream
aws kinesis describe-stream --stream-name my-stream

# Put record
aws kinesis put-record \
  --stream-name my-stream \
  --data '{"eventId": "123", "timestamp": "2024-01-15T10:30:00Z", "value": 42.5}' \
  --partition-key event-123

# Get records
aws kinesis get-shard-iterator \
  --stream-name my-stream \
  --shard-id shardId-000000000000 \
  --shard-iterator-type TRIM_HORIZON

aws kinesis get-records --shard-iterator <iterator> --limit 100

Shard Calculation

Shard = max(Incoming MB/s, Outgoing MB/s/2)

Example:
- 5 MB/s incoming, 10 MB/s outgoing
- Shards = max(5, 10/2) = 5 shards

- 200 records/s, 1 KB each = 0.2 MB/s incoming
- 1000 records/s, 0.5 KB each = 0.5 MB/s outgoing
- Shards = max(0.2, 0.5/2) = 1 shard

Kinesis Data Firehose

Kinesis Data Firehose captures, transforms, and delivers streaming data.

Create Delivery Stream

# Create delivery stream to S3
aws firehose create-delivery-stream \
  --delivery-stream-name my-firehose \
  --delivery-stream-type DirectPut \
  --extended-s3-destination-configuration '{
    "RoleARN": "arn:aws:iam::xxx:role/firehose-role",
    "BucketARN": "arn:aws:s3:::my-bucket",
    "Prefix": "firehose/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/",
    "ErrorOutputPrefix": "errors/",
    "BufferingHints": {
      "SizeInMBs": 64,
      "IntervalInSeconds": 300
    },
    "CompressionFormat": "GZIP",
    "ProcessingConfiguration": {
      "Enabled": true,
      "Processors": [{
        "Type": "Lambda",
        "Parameters": [{
          "ParameterName": "LambdaArn",
          "ParameterValue": "arn:aws:lambda:us-east-1:xxx:function:transform"
        }]
      }]
    }
  }'

# Create delivery stream to Redshift
aws firehose create-delivery-stream \
  --delivery-stream-name my-redshift-firehose \
  --redshift-destination-configuration '{
    "RoleARN": "arn:aws:iam::xxx:role/firehose-role",
    "ClusterJDBCURL": "jdbc:redshift://my-cluster.xxxx.us-east-1.redshift.amazonaws.com:5439/mydb",
    "CopyCommand": {
      "DataTableName": "events",
      "DataTableColumns": "event_id event_type timestamp data"
    },
    "Username": "admin",
    "Password": "mypassword"
  }'

Firehose Destinations

Kinesis Data Analytics

Analyze streaming data with SQL.

Create Application

# Create analytics application
aws kinesisanalytics create-application \
  --application-name my-analytics \
  --runtime-environment SQL-1_0 \
  --input-configurations '[{
    "Id": "1",
    "InputSchema": {
      "RecordColumns": [
        {"Name": "event_type", "Type": "VARCHAR"},
        {"Name": "value", "Type": "DECIMAL"},
        {"Name": "timestamp", "Type": "TIMESTAMP"}
      ],
      "RecordFormat": {
        "RecordFormatType": "JSON",
        "MappingParameters": {
          "JSONMappingParameters": {
            "RecordRowPath": "$"
          }
        }
      }
    },
    "KinesisStreamsInput": {
      "ResourceARN": "arn:aws:kinesis:us-east-1:xxx:stream/my-stream"
    }
  }]'

# Add SQL code
aws kinesisanalytics update-application \
  --application-name my-analytics \
  --application-update '{
    "ApplicationCodeUpdate": {
      "ApplicationCode": "CREATE OR REPLACE STREAM \"DESTINATION_SQL_STREAM\" (event_type VARCHAR(50), avg_value DECIMAL(10,2)); CREATE OR REPLACE PUMP \"STREAM_PUMP\" AS INSERT INTO \"DESTINATION_SQL_STREAM\" SELECT STREAM event_type, AVG(value) OVER (PARTITION BY event_type RANGE INTERVAL '\''5 minute'\'' PRECEDING) FROM \"SOURCE_SQL_STREAM\" GROUP BY event_type;"
    }
  }'

Real-Time Analytics SQL

-- Sliding window aggregation
SELECT 
  event_type,
  COUNT(*) as event_count,
  AVG(value) as avg_value,
  MAX(value) as max_value,
  TUMBLE_START(time, INTERVAL '1' HOUR) as window_start
FROM "SOURCE_SQL_STREAM"
GROUP BY event_type, TUMBLE(time, INTERVAL '1' HOUR);

-- Anomaly detection
SELECT 
  event_type,
  value,
  AVG(value) OVER (PARTITION BY event_type RANGE INTERVAL '5' MINUTE PRECEDING) as moving_avg,
  CASE 
    WHEN value > 2 * AVG(value) OVER (PARTITION BY event_type RANGE INTERVAL '5' MINUTE PRECEDING)
    THEN 'ANOMALY'
    ELSE 'NORMAL'
  END as status
FROM "SOURCE_SQL_STREAM";

Kinesis Producers and Consumers

Producer (Python)

import boto3
import json
from datetime import datetime

kinesis = boto3.client('kinesis')

def put_record(stream_name, data, partition_key):
    response = kinesis.put_record(
        StreamName=stream_name,
        Data=json.dumps(data).encode(),
        PartitionKey=partition_key
    )
    return response['SequenceNumber']

# Batch put
def put_records(stream_name, records):
    response = kinesis.put_records(
        StreamName=stream_name,
        Records=[
            {
                'Data': json.dumps(r).encode(),
                'PartitionKey': r['partition_key']
            } for r in records
        ]
    )
    return response['FailedRecordCount']

Consumer (Python)

import boto3
import json

kinesis = boto3.client('kinesis')

def consume_stream(stream_name, shard_id):
    # Get iterator
    response = kinesis.get_shard_iterator(
        StreamName=stream_name,
        ShardId=shard_id,
        ShardIteratorType='LATEST'
    )
    shard_iterator = response['ShardIterator']
    
    # Read records
    while True:
        response = kinesis.get_records(
            ShardIterator=shard_iterator,
            Limit=100
        )
        
        for record in response['Records']:
            data = json.loads(record['Data'])
            process_record(data)
        
        shard_iterator = response['NextShardIterator']
        if not shard_iterator:
            break

# KCL (Kinesis Client Library) for production
# Uses checkpointing, load balancing, and fault tolerance

Kinesis Best Practices

Stream Configuration

# Monitor shard utilization
aws cloudwatch get-metric-statistics \
  --namespace AWS/Kinesis \
  --metric-name IncomingBytes \
  --dimensions Name=StreamName,Value=my-stream \
  --start-time $(date -u -d '1 hour ago') \
  --end-time $(date -u) \
  --period 300 \
  --statistics Sum

# Reshard (increase capacity)
aws kinesis update-shard-count \
  --stream-name my-stream \
  --target-shard-count 8 \
  --scaling-type UNIFORM_SCALING

Architecture Patterns

┌─────────────────────────────────────────────────────────────┐
│              Real-Time Analytics Architecture                │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Producers ──▶ Kinesis Data Streams                        │
│                   │                                        │
│         ┌─────────┼─────────┐                             │
│         │         │         │                             │
│         ▼         ▼         ▼                             │
│    Kinesis Data Firehose (Fan-out)                        │
│         │         │         │                             │
│         ▼         ▼         ▼                             │
│      S3 Data    Redshift   Elasticsearch                  │
│       Lake      Warehouse   (Search)                       │
│                                                             │
│    Kinesis Data Analytics (Real-time SQL)                  │
│         │                                                  │
│         ▼                                                  │
│    Lambda ──▶ Alerts/Actions                               │
└─────────────────────────────────────────────────────────────┘

Best Practices

1. Choose shard count based on throughput needs
2. Use partition keys for even distribution
3. Enable encryption at rest and in transit
4. Use Firehose for simplified delivery to S3/Redshift
5. Implement retries with backoff for producers
6. Use KCL for production consumers
7. Monitor shard utilization and reshard as needed
8. Set up alarms for iterator age and get records latency