Kinesis Family

Real-time data streaming, near-real-time archival, and windowed aggregation on AWS

Introduction

SQS and SNS handle discrete messages between services. Kinesis handles continuous data flows: millions of events per second from IoT sensors, clickstream analytics, application logs, or financial transactions. The Kinesis family gives you four services that together cover real-time ingestion, durable archival, windowed aggregation, and video streaming.

In this lesson you'll build a sensor telemetry pipeline: synthetic warehouse sensors produce readings into a Kinesis Data Stream, Firehose archives the raw events to S3, and a Lambda with a 60-second tumbling window aggregates count, average, min, and max per sensor group and writes the results to DynamoDB.

1. The Kinesis Family

Kinesis Data Streams (KDS)

Durable, ordered, real-time stream. Producers write records and multiple independent consumer groups each read at their own pace. Records persist for 24 hours by default (up to 365 days) so consumers can replay data. Capacity is measured in shards.

Kinesis Data Firehose

Fully managed near-real-time delivery service. No consumer code required. Buffers records and delivers them to S3, Redshift, OpenSearch, Splunk, Snowflake, or HTTP endpoints. Minimum latency is 60 seconds (buffering).

Managed Apache Flink

Formerly Kinesis Data Analytics. Run Apache Flink applications on streaming data for stateful processing, time-windowed aggregations, and anomaly detection. AWS manages the Flink cluster. Always-on (not scale-to-zero).

Kinesis Video Streams

Ingest, store, and process streaming video from cameras and IoT devices. Integrates with Amazon Rekognition for ML-based video analysis. Typically used for security cameras, smart home, and industrial vision.

Sensor Telemetry Pipeline

Figure 1: Sensor telemetry pipeline. KDS fans out to the Lambda aggregator and Firehose simultaneously.

2. Kinesis Data Streams: Core Concepts

Kinesis Data Streams (KDS) is a durable, ordered, real-time stream. Producers write records and they become available to consumers within milliseconds. Unlike SQS, records are not deleted after consumption, they persist for a configurable retention period and multiple independent consumer groups can each read the same stream at their own pace.

Shards

A shard is the unit of capacity. Each shard provides:

1 MB/s write throughput (up to 1,000 records/s)
2 MB/s read throughput, shared across all standard consumers on that shard
Enhanced Fan-Out: 2 MB/s dedicated per registered consumer per shard, bypassing the shared read limit

Capacity Modes

Provisioned: set a fixed shard count, pay per shard-hour. You control scaling manually.
On-Demand: AWS auto-scales shards based on throughput, you pay per GB in and out. No capacity planning needed.

Records and Retention

Each record contains:

Partition key: determines which shard the record lands on; records with the same partition key land on the same shard in order
Sequence number: assigned by Kinesis, unique and monotonically increasing per shard
Data blob: up to 1 MB of arbitrary bytes (the Lambda ESM base64-encodes it before delivering to your handler)

Default retention is 24 hours. Extend to up to 365 days to replay historical data.

Multiple Consumers

Unlike SQS, consuming a record does not delete it. Multiple consumer groups each maintain their own position (iterator) in the stream and read independently:

A Lambda aggregator reads for windowed analytics
Firehose reads the same records for archival to S3
A third consumer could feed an OpenSearch index

All three consume the same stream without interfering with each other.

CDK: create a provisioned Kinesis Data Stream with 1 shard and a 24-hour retention window. Swap shard_count for the stream_mode=kinesis.StreamMode.ON_DEMAND property to let AWS manage capacity automatically.

stream = kinesis.Stream(self, "SensorEventStream",
    shard_count=1,
    retention_period=cdk.Duration.hours(24),
)

3. Kinesis Data Firehose

Firehose is a fully managed delivery service. You configure a source and a destination, and Firehose handles batching, compression, encryption, and retry automatically. There is no consumer code to write or infrastructure to manage.

Sources

Kinesis Data Stream (KinesisStreamAsSource)
Direct PUT from SDK
Amazon MSK (Managed Kafka)

Destinations

Amazon S3
Amazon Redshift
Amazon OpenSearch
Splunk, Snowflake, HTTP endpoint

Buffer Settings

By size: 1-128 MB
By time: 60-900 seconds
Delivers on whichever threshold is hit first
Minimum latency: 60 seconds

Firehose is near-real-time, not real-time

The 60-second minimum buffer means records arriving at 12:00:00 may not land in S3 until 12:01:00 or later. If your use case requires sub-second latency, consume directly from Kinesis Data Streams with a Lambda or KCL application instead.

CDK: create a Firehose delivery stream that reads from the Kinesis stream and archives raw events to S3 under the raw/ prefix. Records are buffered for up to 60 seconds or 1 MB before delivery, whichever comes first. The L1 CfnDeliveryStreamis used here because the L2 Firehose constructs still live in an alpha module (aws-kinesisfirehose-alpha) and may introduce breaking changes.

# The CDK L2 Firehose constructs live in an alpha module (aws-kinesisfirehose-alpha).
# For stability, use the L1 CfnDeliveryStream directly:
delivery_stream = firehose.CfnDeliveryStream(self, "RawEventsDeliveryStream",
    delivery_stream_type="KinesisStreamAsSource",
    kinesis_stream_source_configuration=firehose.CfnDeliveryStream.KinesisStreamSourceConfigurationProperty(
        kinesis_stream_arn=stream.stream_arn,
        role_arn=firehose_role.role_arn,
    ),
    extended_s3_destination_configuration=firehose.CfnDeliveryStream.ExtendedS3DestinationConfigurationProperty(
        bucket_arn=raw_bucket.bucket_arn,
        role_arn=firehose_role.role_arn,
        prefix="raw/",
        buffering_hints=firehose.CfnDeliveryStream.BufferingHintsProperty(
            interval_in_seconds=60,
            size_in_m_bs=1,
        ),
    ),
)

4. Lambda Tumbling Windows

Windowed aggregation: grouping events by time and computing statistics was the main use case for Kinesis Data Analytics. That SQL-based service is now deprecated. The supported path forward is Managed Apache Flink, but Flink is always-on (not scale-to-zero), takes minutes to start and stop, and requires packaging a Flink application. For most windowed aggregation needs, Lambda tumbling windows give you identical semantics with zero infrastructure overhead.

How Tumbling Windows Work

Lambda Event Source Mapping reads batches from the stream continuously
Your handler accumulates values in a state dict returned from each invocation
Lambda carries state forward across all batches within the window
When the window closes, Lambda sets isFinalInvokeForWindow: True and passes the completed window.start / window.end
Your handler flushes the aggregate (writes to DynamoDB) and returns state: {} to reset

Lambda Tumbling Windows vs Managed Flink

	Lambda Windows	Managed Flink
Billing	Per invocation	Per KPU-hour (always-on)
Scale to zero	Yes	No
Start/stop time	Milliseconds	Minutes
Language	Any Lambda runtime	Java/Python Flink app
State limit	1 MB	Unlimited (S3-backed)
Best for	Simple windowed agg	Complex stateful pipelines

CDK: attach the Lambda aggregator to the Kinesis stream with a 60-second tumbling window. The tumbling_window parameter on KinesisEventSource is what enables state carry-forward and the final-invoke flush.

aggregator_fn.add_event_source(
    lambda_event_sources.KinesisEventSource(stream,
        starting_position=lambda_.StartingPosition.LATEST,
        batch_size=100,
        max_batching_window=cdk.Duration.seconds(10),
        tumbling_window=cdk.Duration.seconds(60),
        retry_attempts=2,
    )
)

5. Kinesis vs SQS vs SNS

Feature	Kinesis Data Streams	SQS	SNS
Pattern	Streaming (ordered per shard)	Queue (worker pool)	Pub/Sub (push)
Consumers	Multiple independent groups	One consumer group	N subscribers
Replay	Yes - up to 365 days	No - deleted on consume	No
Ordering	Per shard - guaranteed	FIFO queue only	Not guaranteed
Max message size	1 MB	256 KB	256 KB
Latency	~70 ms	~1 s (polling)	Near-instant push
Managed delivery	No - consumer pulls	No - consumer polls	Yes - push to endpoint
Best for	Real-time analytics, IoT, logs	Task queues, job workers	Fan-out notifications

Decision Guide

Use Kinesis Data Streams when:

Multiple independent consumers need the same data
You need to replay historical events
Ordered processing within a partition matters
Records exceed 256 KB or arrive at very high throughput

Use Kinesis Data Firehose when:

You need to archive streaming data to S3, Redshift, or OpenSearch
60-second delivery latency is acceptable
You want delivery without writing consumer code

Use SQS when:

Decoupling services with a worker pool pattern
Exactly-once processing with FIFO queues
Backpressure control - workers scale to queue depth

Use SNS when:

Broadcasting one event to many subscribers at once
Fan-out to Lambda + SQS + email + HTTP simultaneously
Push-based delivery to endpoints without polling

6. Code Example: Sensor Telemetry Pipeline

The three building blocks of the pipeline: a boto3 producer that sends synthetic sensor batches, a Lambda handler that accumulates windowed aggregates, and the CDK stack that wires everything together.

import json
import random
import sys
import time
from datetime import datetime, timezone

import boto3

kinesis = boto3.client("kinesis")

SENSORS = [
    {"sensorType": "temperature", "location": "warehouse-a", "unit": "celsius", "range": (18.0, 26.0)},
    {"sensorType": "temperature", "location": "warehouse-b", "unit": "celsius", "range": (16.0, 24.0)},
    {"sensorType": "temperature", "location": "loading-dock", "unit": "celsius", "range": (10.0, 30.0)},
    {"sensorType": "humidity",    "location": "warehouse-a", "unit": "percent", "range": (35.0, 55.0)},
    {"sensorType": "humidity",    "location": "warehouse-b", "unit": "percent", "range": (30.0, 60.0)},
]


def make_reading(sensor_id: int, sensor: dict) -> dict:
    low, high = sensor["range"]
    return {
        "sensorId": f"sensor-{sensor_id:03d}",
        "sensorType": sensor["sensorType"],
        "location": sensor["location"],
        "reading": round(random.uniform(low, high), 2),
        "unit": sensor["unit"],
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }


def send_batch(stream_name: str, batch_num: int) -> int:
    records = []
    for offset, sensor in enumerate(SENSORS):
        reading = make_reading(batch_num * len(SENSORS) + offset, sensor)
        records.append({
            "Data": json.dumps(reading).encode(),
            # All readings from the same sensor type + location land on the same
            # shard, preserving ordered delivery within each group
            "PartitionKey": f"{reading['sensorType']}#{reading['location']}",
        })
    response = kinesis.put_records(StreamName=stream_name, Records=records)
    return response["FailedRecordCount"]


stream_name = sys.argv[1]
for batch_num in range(24):   # ~2 min at 5s intervals, covers two 60s windows
    failed = send_batch(stream_name, batch_num)
    print(f"Batch {batch_num + 1}: {len(SENSORS)} readings sent, {failed} failed")
    time.sleep(5)

import base64
import json
import logging
import os
from decimal import Decimal

import boto3

logger = logging.getLogger()
logger.setLevel(logging.INFO)
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])


def handler(event: dict, context) -> dict:
    # Lambda injects 'state', 'window', and 'isFinalInvokeForWindow' into the
    # event when a tumbling_window is configured on the Event Source Mapping.
    # On regular batches, state carries forward; on the final invocation it's flushed.
    state = event.get("state") or {}
    records = event.get("Records", [])
    is_final = event.get("isFinalInvokeForWindow", False)

    logger.info("Aggregating %d record(s), isFinalInvokeForWindow=%s", len(records), is_final)

    for record in records:
        # Kinesis records are base64-encoded by the Event Source Mapping
        reading_event = json.loads(base64.b64decode(record["kinesis"]["data"]))
        key = f"{reading_event['sensorType']}#{reading_event['location']}"
        reading = float(reading_event["reading"])

        bucket = state.setdefault(key, {
            "count": 0, "sum": 0.0, "min": reading, "max": reading,
        })
        bucket["count"] += 1
        bucket["sum"]   += reading
        bucket["min"]    = min(bucket["min"], reading)
        bucket["max"]    = max(bucket["max"], reading)

    # Return updated state so Lambda carries it into the next batch
    if not is_final:
        return {"state": state}

    # Final invocation: flush one row per sensor group to DynamoDB, reset state
    window = event["window"]
    for key, agg in state.items():
        avg = agg["sum"] / agg["count"]
        table.put_item(Item={
            "sensorKey":   key,
            "windowStart": window["start"],
            "windowEnd":   window["end"],
            "count":       agg["count"],
            "avgReading":  Decimal(str(round(avg, 2))),
            "minReading":  Decimal(str(agg["min"])),
            "maxReading":  Decimal(str(agg["max"])),
        })
        logger.info(
            "Window %s..%s - %s: count=%d avg=%.2f",
            window["start"], window["end"], key, agg["count"], avg,
        )

    return {"state": {}}

import os
import aws_cdk as cdk
from aws_cdk import (
    aws_dynamodb as dynamodb,
    aws_iam as iam,
    aws_kinesis as kinesis,
    aws_kinesisfirehose as firehose,
    aws_lambda as lambda_,
    aws_lambda_event_sources as lambda_event_sources,
    aws_s3 as s3,
)
from constructs import Construct

LAMBDAS_DIR = os.path.join(os.path.dirname(__file__), "..", "lambdas")


class StreamingPipelineStack(cdk.Stack):
    def __init__(self, scope: Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # --- Kinesis Data Stream ---
        stream = kinesis.Stream(self, "SensorEventStream",
            shard_count=1,
            retention_period=cdk.Duration.hours(24),
        )

        # --- Firehose: archive raw events to S3 ---
        # The L2 Firehose constructs still live in an alpha module; use L1 for stability.
        raw_bucket = s3.Bucket(self, "RawEventsBucket",
            removal_policy=cdk.RemovalPolicy.DESTROY,
            auto_delete_objects=True,
        )

        firehose_role = iam.Role(self, "FirehoseDeliveryRole",
            assumed_by=iam.ServicePrincipal("firehose.amazonaws.com"),
        )
        stream.grant_read(firehose_role)
        raw_bucket.grant_read_write(firehose_role)

        delivery_stream = firehose.CfnDeliveryStream(self, "RawEventsDeliveryStream",
            delivery_stream_type="KinesisStreamAsSource",
            kinesis_stream_source_configuration=firehose.CfnDeliveryStream.KinesisStreamSourceConfigurationProperty(
                kinesis_stream_arn=stream.stream_arn,
                role_arn=firehose_role.role_arn,
            ),
            extended_s3_destination_configuration=firehose.CfnDeliveryStream.ExtendedS3DestinationConfigurationProperty(
                bucket_arn=raw_bucket.bucket_arn,
                role_arn=firehose_role.role_arn,
                prefix="raw/",
                error_output_prefix="errors/",
                buffering_hints=firehose.CfnDeliveryStream.BufferingHintsProperty(
                    interval_in_seconds=60,
                    size_in_m_bs=1,
                ),
            ),
        )
        delivery_stream.node.add_dependency(firehose_role)

        # --- DynamoDB: per-window aggregates ---
        table = dynamodb.Table(self, "AggregatesTable",
            partition_key=dynamodb.Attribute(name="sensorKey",    type=dynamodb.AttributeType.STRING),
            sort_key=dynamodb.Attribute(     name="windowStart",  type=dynamodb.AttributeType.STRING),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=cdk.RemovalPolicy.DESTROY,
        )

        # --- Lambda with tumbling-window Event Source Mapping ---
        aggregator_fn = lambda_.Function(self, "AggregatorFn",
            runtime=lambda_.Runtime.PYTHON_3_12,
            handler="aggregator.handler",
            code=lambda_.Code.from_asset(os.path.join(LAMBDAS_DIR, "aggregator")),
            environment={"TABLE_NAME": table.table_name},
            timeout=cdk.Duration.seconds(60),
        )
        table.grant_write_data(aggregator_fn)

        # tumbling_window groups all batches within a 60s window; Lambda carries
        # state forward across batches and does a final flush when the window closes
        aggregator_fn.add_event_source(
            lambda_event_sources.KinesisEventSource(stream,
                starting_position=lambda_.StartingPosition.LATEST,
                batch_size=100,
                max_batching_window=cdk.Duration.seconds(10),
                tumbling_window=cdk.Duration.seconds(60),
                retry_attempts=2,
            )
        )

        # --- Outputs ---
        cdk.CfnOutput(self, "StreamName",             value=stream.stream_name)
        cdk.CfnOutput(self, "RawEventsBucketName",    value=raw_bucket.bucket_name)
        cdk.CfnOutput(self, "DeliveryStreamName",     value=delivery_stream.ref)
        cdk.CfnOutput(self, "AggregatesTableName",    value=table.table_name)
        cdk.CfnOutput(self, "AggregatorFunctionName", value=aggregator_fn.function_name)

Bonus: Full Working Project

The complete deployable pipeline from this lesson is available as a GitLab repository. Clone, deploy, and watch real windowed aggregates appear in DynamoDB within minutes.

The project includes:

StreamingPipelineStack - CDK stack: Kinesis stream, Firehose to S3, Lambda with tumbling-window ESM, DynamoDB aggregates table
aggregator.py - Lambda handler: base64 decode, per-batch state accumulation, final-window flush to DynamoDB
produce_events.py - sends 24 batches of 5 sensor readings over ~2 minutes to cover two complete 60-second windows
read_aggregates.py - scans DynamoDB and prints count, avg, min, max per sensor group per window in a formatted table

Full source code:gitlab.com/alek.cora.glez/aws/kinesis

Key Takeaways

KDS for real-time streaming, Firehose for managed delivery - KDS gives you ordered records with replay and multiple independent consumers; Firehose delivers the same stream to S3 or Redshift without a line of consumer code
Shards are the capacity unit - 1 MB/s write and 2 MB/s read per shard; choose Provisioned for predictable load or On-Demand to skip capacity planning
KDS vs SQS: streams allow replay and parallel consumer groups - SQS queues are destructive reads for one worker group, KDS is non-destructive and retains records for up to 365 days
Firehose minimum latency is 60 seconds - it is near-real-time, not real-time; for sub-second needs consume directly from KDS with Lambda or KCL
Lambda tumbling windows replace Kinesis Analytics SQL - same windowed aggregation semantics (accumulate state, flush on window close), scale-to-zero billing, pure Python, no Flink cluster required

Back to "Step Functions"

AWS FundamentalsLesson 17

Next: "Big Data & Analytics"