Amazon Kinesis Explained: Real-Time Streaming

Batch pipelines move data that has finished arriving. Streaming handles data that never finishes: clicks, payments, sensor readings, log lines. Amazon Kinesis is AWS's own family of services for that. The core service, Kinesis Data Streams, collects and processes large streams of data records in real time (Kinesis docs: what is Kinesis Data Streams).

If you have met Apache Kafka, the model will be familiar: a durable, ordered log that many producers write to and many consumers read from at their own pace. Apache Kafka explained covers that model in general. This article covers the AWS version and its vocabulary.

The family

• Kinesis Data Streams: the stream itself. You write records and read them back.
• Amazon Data Firehose: a delivery service. It takes streaming data and loads it into S3, Redshift, OpenSearch and other destinations, with no consumer code.
• Managed Service for Apache Flink: stream processing with Flink, for windows, joins and aggregations.
• Kinesis Video Streams: video, a separate concern.

Most of the concepts worth learning belong to Data Streams.

Streams, shards and records

The terminology page defines the parts (Kinesis docs: concepts):

• A stream is a set of shards. A shard is an ordered sequence of records and a fixed unit of capacity.
• A record has three parts: a partition key, a sequence number assigned by Kinesis, and a data blob of up to 1 MB. Kinesis does not inspect or change the blob.
• Retention: records stay readable for 24 hours by default, extendable up to 365 days. Reading does not remove them.

Each shard supports up to 1 MB or 1,000 records per second for writes, and up to 2 MB per second for reads. A stream's capacity is the sum of its shards. Going over a shard's limit produces throttling errors, not slower writes.

The partition key decides everything

Every record must carry a partition key. Kinesis applies an MD5 hash to it, and the result determines which shard receives the record. Two consequences:

• Ordering. Records with the same key go to the same shard, and within a shard order is preserved. That is how you keep all events for one customer in sequence.
• Load. If some keys are far more common than others, their shards get far more traffic.

To see the second point, we replayed 2,000 real New York taxi trips, in pickup order, into a four-shard stream, twice: once keyed by pickup zone and once keyed by a running event number. This uses moto, a local emulator of the Kinesis API, so the placement shown is the emulator's implementation of the documented hashing.

import json, os
from collections import Counter
import boto3
import pyarrow.compute as pc
import pyarrow.parquet as pq
from moto import mock_aws

os.environ.update(AWS_ACCESS_KEY_ID="testing", AWS_SECRET_ACCESS_KEY="testing",
                  AWS_DEFAULT_REGION="us-east-1")

# 2,000 real trips, in pickup order, to play the part of live events
trips = pq.read_table("yellow_tripdata_2024-01.parquet",
                      columns=["tpep_pickup_datetime", "PULocationID", "fare_amount"])
trips = trips.filter(pc.greater_equal(trips["tpep_pickup_datetime"], pc.strptime("2024-01-15 08:00:00", "%Y-%m-%d %H:%M:%S", "us")))
events = trips.sort_by("tpep_pickup_datetime").slice(0, 2000).to_pylist()


@mock_aws
def main():
    kinesis = boto3.client("kinesis")
    kinesis.create_stream(StreamName="trips", ShardCount=4)

    def send(key_of):
        placed = Counter()
        for i in range(0, len(events), 500):                    # PutRecords takes up to 500 records
            batch = [{"Data": json.dumps({"zone": e["PULocationID"], "fare": e["fare_amount"],
                                          "at": e["tpep_pickup_datetime"].isoformat()}).encode(),
                      "PartitionKey": key_of(n, e)} for n, e in enumerate(events[i:i + 500], start=i)]
            response = kinesis.put_records(StreamName="trips", Records=batch)
            placed.update(r["ShardId"][-1] for r in response["Records"])
        return [placed[s] for s in "0123"]

    print("records per shard, 4 shards, 2,000 events")
    print("  key = pickup zone :", send(lambda n, e: str(e["PULocationID"])))
    print("  key = event number:", send(lambda n, e: str(n)))

    # Read one shard from the beginning: records come back in the order they were written
    shard = kinesis.describe_stream(StreamName="trips")["StreamDescription"]["Shards"][0]["ShardId"]
    iterator = kinesis.get_shard_iterator(StreamName="trips", ShardId=shard,
                                          ShardIteratorType="TRIM_HORIZON")["ShardIterator"]
    records = kinesis.get_records(ShardIterator=iterator, Limit=3)["Records"]
    print()
    print("first three records in", shard)
    for r in records:
        print("  key", r["PartitionKey"].rjust(3), "|", r["Data"].decode())
    zones = Counter(str(e["PULocationID"]) for e in events)
    print()
    print("distinct zones:", len(zones), "| busiest three:", zones.most_common(3))


main()

records per shard, 4 shards, 2,000 events
  key = pickup zone : [601, 585, 614, 200]
  key = event number: [509, 472, 498, 521]

first three records in shardId-000000000000
  key 170 | {"zone": 170, "fare": 5.1, "at": "2024-01-15T08:00:01"}
  key 262 | {"zone": 262, "fare": 16.3, "at": "2024-01-15T08:00:03"}
  key 138 | {"zone": 138, "fare": 65.0, "at": "2024-01-15T08:00:06"}

distinct zones: 101 | busiest three: [('230', 177), ('186', 108), ('132', 92)]

Keyed by event number, the four shards receive between 472 and 521 records each: an even spread. Keyed by pickup zone, three shards receive about 600 each and one receives 200. There are only 101 distinct zones in the sample, and they are far from equally busy; the busiest zone alone accounts for 177 of the 2,000 events.

At this scale the imbalance is harmless. At production volume it is the classic hot shard: one shard reaches its write limit and throttles while the others sit half idle, and adding shards does not help if the traffic is concentrated in a few keys.

The trade-off is real, though. The event-number key balances perfectly and destroys per-zone ordering, because consecutive events for one zone land on different shards. Choose the key from what must stay in order:

The middle of the output shows the reading side: asking for a shard's records from the beginning, TRIM_HORIZON, returns them in the order they were written.

Capacity modes

A stream runs in one of two kinds of mode (Kinesis docs: choosing a mode):

On-demand removes shard arithmetic and is the sensible default while a workload's shape is unknown. It does not remove the hot-key problem: the partition key still decides how evenly traffic spreads.

Consumers

A stream can have several independent consumers, each reading at its own position. There are two ways to read:

• Shared throughput. Consumers poll, and share each shard's 2 MB per second between them. Fine for one or two consumers.
• Enhanced fan-out. Each registered consumer gets its own dedicated 2 MB per second per shard, pushed to it, so consumers do not contend (Kinesis docs: enhanced fan-out).

Common consumers are a Lambda function, which receives batches of records through an event source mapping; an application built on the Kinesis Client Library, which tracks its position for each shard; a Flink application; and Firehose.

Delivery is at-least-once in practice: after a failure a consumer resumes from its last checkpoint and may see some records again. Design consumers so that reprocessing a record does no harm.

Firehose: when you only need delivery

Many streaming requirements are not "process each event" but "get these events into S3 every few minutes". That is Firehose. AWS describes it as a fully managed service for delivering real-time streaming data to destinations, with no applications to write (Firehose docs: what is Amazon Data Firehose). It buffers incoming data by size or time and writes batches to the destination, and it can read from an existing data stream.

The distinction to hold on to: Data Streams stores records for consumers to read and re-read. Firehose delivers them and keeps nothing for replay. Streaming data with Kinesis to S3 and Redshift builds that delivery path.

Kinesis or Kafka?

Kinesis is the simpler choice when everything is on AWS and nobody wants to operate brokers. Kafka, self-managed or through Amazon MSK, suits teams that need Kafka's ecosystem, its client compatibility, or portability beyond AWS. Amazon MSK covers the managed Kafka option.

Before either, check that streaming is needed at all. Batch vs streaming measures what each level of freshness costs.

Mistakes to avoid

• A low-cardinality or skewed partition key, creating a hot shard.
• One constant partition key, which sends everything to a single shard.
• Assuming exactly-once delivery in consumers.
• Ignoring throttling errors in producers instead of retrying with backoff.
• Using a stream where Firehose would do.
• Forgetting retention. After the retention period the records are gone; land raw events in S3 if they must be kept.

Real-time streaming with Amazon Kinesis and Apache Kafka is covered in the Data Engineering course, which includes a real-time streaming pipeline project.

Sources

• What is Amazon Kinesis Data Streams? (Kinesis Data Streams Developer Guide) — AWS Documentation
• Amazon Kinesis Data Streams terminology and concepts (Kinesis Data Streams Developer Guide) — AWS Documentation
• Choose the right mode to stream in (Kinesis Data Streams Developer Guide) — AWS Documentation
• Develop enhanced fan-out consumers with dedicated throughput (Kinesis Data Streams Developer Guide) — AWS Documentation
• What is Amazon Data Firehose? (Amazon Data Firehose Developer Guide) — AWS Documentation
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission