Batch processing is the wrong tool when you need to react to data as it arrives. A fraud detection system that processes transactions in hourly batches is not a fraud detection system. An analytics dashboard that updates once a day is not analytics. A recommendation engine that learns from user behavior on a 24-hour delay is already behind.

Kinesis Data Streams gives you a managed real-time data pipeline that ingests millions of events per second, retains them for configurable periods, and lets multiple independent consumers read the same stream without interfering with each other. In this article I will walk through how Kinesis works, the producer and consumer patterns that hold up in production, and the Terraform configuration that matters.

How Kinesis Data Streams Works

A Kinesis stream is divided into shards. Each shard provides 1 MB per second of write throughput and 2 MB per second of read throughput. If you need more throughput, you add more shards. The total stream capacity is the sum of all its shards.

Every record written to a stream includes a partition key. Kinesis hashes the partition key to determine which shard the record goes to. Records with the same partition key always go to the same shard and are read in the order they were written. This is how you get per-entity ordering: use the user ID or device ID as the partition key and all events for that entity land on the same shard in sequence.

Records are retained in the stream for a configurable duration, defaulting to 24 hours and extendable up to 365 days. Consumers read from a shard using a shard iterator that tracks their position. Multiple consumer applications can read the same stream independently, each maintaining its own position. A Lambda function, a Kinesis Data Analytics application, and a custom ECS consumer can all process the same stream without any coordination between them.

Setting Up a Stream with Terraform

resource "aws_kinesis_stream" "events" {
  name             = "application-events"
  retention_period = 168

  stream_mode_details {
    stream_mode = "ON_DEMAND"
  }

  encryption_type = "KMS"
  kms_key_id      = aws_kms_key.kinesis.arn

  tags = {
    Environment = "production"
    ManagedBy   = "terraform"
  }
}

resource "aws_kms_key" "kinesis" {
  description             = "KMS key for Kinesis stream encryption"
  deletion_window_in_days = 30
  enable_key_rotation     = true
}

resource "aws_cloudwatch_log_group" "kinesis_errors" {
  name              = "/aws/kinesis/application-events"
  retention_in_days = 14
}

stream_mode = “ON_DEMAND” automatically scales shard count based on your throughput. You do not need to provision shards manually or manage shard splits and merges. On-demand mode costs slightly more per GB than provisioned shards at sustained high throughput, but for most workloads the operational simplicity is worth it. If your stream consistently processes more than 200 MB per second, evaluate provisioned mode with reserved shard capacity.

retention_period = 168 sets a 7-day retention window. The default 24-hour retention is often too short for production. If a consumer has a bug and is down for a day, you want the ability to replay events from before the outage. 7 days gives a comfortable reprocessing window without the cost of maximum retention.

Producing Records

Writing to Kinesis efficiently means batching your puts. PutRecords sends up to 500 records in a single API call and is far more efficient than individual PutRecord calls.

import boto3
import json
import time
from typing import List
from dataclasses import dataclass

@dataclass
class KinesisRecord:
    data: dict
    partition_key: str

class KinesisProducer:
    def __init__(self, stream_name: str, region: str = "us-east-1"):
        self.client      = boto3.client("kinesis", region_name=region)
        self.stream_name = stream_name
        self.buffer: List[KinesisRecord] = []
        self.max_batch_size = 500

    def add(self, data: dict, partition_key: str):
        self.buffer.append(KinesisRecord(data=data, partition_key=partition_key))
        if len(self.buffer) >= self.max_batch_size:
            self.flush()

    def flush(self):
        if not self.buffer:
            return

        records = [
            {
                "Data": json.dumps(record.data).encode("utf-8"),
                "PartitionKey": record.partition_key
            }
            for record in self.buffer
        ]

        self.buffer = []
        self._put_with_retry(records)

    def _put_with_retry(self, records: list, attempt: int = 0):
        if attempt > 3:
            raise RuntimeError(f"Failed to put {len(records)} records after 3 retries")

        response = self.client.put_records(
            StreamName=self.stream_name,
            Records=records
        )

        failed_count = response.get("FailedRecordCount", 0)
        if failed_count == 0:
            return

        failed_records = [
            records[i] for i, result in enumerate(response["Records"])
            if "ErrorCode" in result
        ]

        time.sleep(0.1 * (2 ** attempt))
        self._put_with_retry(failed_records, attempt + 1)


producer = KinesisProducer(stream_name="application-events")

def track_user_event(user_id: str, event_type: str, properties: dict):
    event = {
        "userId":    user_id,
        "eventType": event_type,
        "timestamp": time.time(),
        "properties": properties
    }
    producer.add(data=event, partition_key=user_id)

def on_request_end():
    producer.flush()

Retry logic on put_records is not optional. Even with on-demand mode, individual records in a batch can fail with ProvisionedThroughputExceededException. The response includes a FailedRecordCount and per-record error codes, so you can retry only the failed records with exponential backoff rather than resending the entire batch.

Using the user_id as the partition key ensures all events for a given user land on the same shard in order. If your stream consumers need per-user ordering, this is the right pattern. If ordering does not matter and you want even load distribution, use a random UUID as the partition key instead.

Consuming with Lambda

resource "aws_lambda_event_source_mapping" "kinesis_consumer" {
  event_source_arn                   = aws_kinesis_stream.events.arn
  function_name                      = aws_lambda_function.process_events.arn
  starting_position                  = "LATEST"
  batch_size                         = 100
  maximum_batching_window_in_seconds = 5
  parallelization_factor             = 3
  bisect_batch_on_function_error     = true
  maximum_retry_attempts             = 3

  destination_config {
    on_failure {
      destination_arn = aws_sqs_queue.kinesis_dlq.arn
    }
  }

  filter_criteria {
    filter {
      pattern = jsonencode({
        data = {
          eventType = ["user_action", "purchase", "page_view"]
        }
      })
    }
  }
}

parallelization_factor = 3 allows Lambda to process 3 batches from each shard concurrently. Without this, each shard is processed by a single Lambda invocation at a time. With on-demand mode and unpredictable shard counts, increasing parallelization helps keep processing latency low during traffic bursts.

bisect_batch_on_function_error = true is one of the most valuable settings here. When a Lambda invocation fails, instead of retrying the entire batch, Kinesis bisects it and retries each half separately. This continues until individual failing records are isolated. Combined with a DLQ destination, poison records get routed to the dead letter queue rather than blocking the shard indefinitely.

The filter_criteria means Lambda is only invoked for records matching the specified event types. Records that do not match are acknowledged automatically without invoking Lambda, which reduces cost and invocation count significantly for streams carrying multiple event types.

The Consumer Function

import base64
import json
import logging

logger = logging.getLogger()

def lambda_handler(event, context):
    processed  = 0
    failed_ids = []

    for record in event["Records"]:
        sequence_number = record["kinesis"]["sequenceNumber"]
        try:
            payload = json.loads(
                base64.b64decode(record["kinesis"]["data"]).decode("utf-8")
            )
            process_event(payload)
            processed += 1
        except Exception as e:
            logger.error(f"Failed to process record {sequence_number}: {e}")
            failed_ids.append({"itemIdentifier": record["kinesis"]["sequenceNumber"]})

    logger.info(f"Processed {processed} records, {len(failed_ids)} failures")

    return {"batchItemFailures": [
        {"itemIdentifier": sid} for sid in failed_ids
    ]}

def process_event(payload: dict):
    event_type = payload.get("eventType")

    if event_type == "purchase":
        handle_purchase(payload)
    elif event_type == "user_action":
        handle_user_action(payload)
    elif event_type == "page_view":
        handle_page_view(payload)
    else:
        logger.warning(f"Unknown event type: {event_type}")

Kinesis records arrive base64 encoded. Always decode before parsing as JSON. This catches a significant number of runtime errors in new consumer implementations where the developer forgets the base64 step.

Monitoring a Production Stream

resource "aws_cloudwatch_metric_alarm" "iterator_age_high" {
  alarm_name          = "kinesis-iterator-age-high"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 3
  metric_name         = "GetRecords.IteratorAgeMilliseconds"
  namespace           = "AWS/Kinesis"
  period              = 60
  extended_statistic  = "p99"
  threshold           = 60000
  alarm_description   = "Kinesis consumer is more than 60 seconds behind the stream"
  alarm_actions       = [aws_sns_topic.alerts.arn]

  dimensions = {
    StreamName = aws_kinesis_stream.events.name
  }
}

resource "aws_cloudwatch_metric_alarm" "write_throttling" {
  alarm_name          = "kinesis-write-throttling"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 2
  metric_name         = "WriteProvisionedThroughputExceeded"
  namespace           = "AWS/Kinesis"
  period              = 60
  statistic           = "Sum"
  threshold           = 100
  alarm_description   = "Kinesis stream is experiencing write throttling"
  alarm_actions       = [aws_sns_topic.alerts.arn]

  dimensions = {
    StreamName = aws_kinesis_stream.events.name
  }
}

GetRecords.IteratorAgeMilliseconds is the most important Kinesis health metric. It measures how far behind the consumer is from the tip of the stream. If this climbs past a few minutes, your consumers cannot keep up with the incoming data rate. The p99 statistic catches lagging shards that the average would miss.

Closing Thoughts

Kinesis Data Streams is the right foundation for any system that needs to react to data in real time rather than after the fact. The shard model gives you predictable throughput and ordering guarantees. The retention window gives you replay capability. Multiple independent consumers mean you can build multiple downstream systems without coupling them to each other.

Start with on-demand mode, use sensible retention, implement retry logic in your producers, and watch your iterator age. Those four things will keep your stream healthy through growth and traffic spikes.

Enjoy the cloud.

Osama

---

#AWS #Kinesis #RealTimeData #StreamProcessing #DataEngineering #CloudArchitecture #EventDrivenArchitecture #Terraform #InfrastructureAsCode #AWSLambda #CloudNative #AmazonWebServices #SolutionsArchitect #CloudComputing #BackendEngineering #BigData #TechBlog #DataPipeline #MLOps #CloudInfrastructure