AWS Kinesis Streaming: Producer and Consumer

Batch vs stream

Batch ETL processes data on a schedule (nightly), so insights lag a day. A stream processes events as they arrive. Kinesis is the managed stream: producers append records, consumers read them in near real-time, ordered within a shard.

The producer

import json, boto3
k = boto3.client("kinesis")
STREAM = "events"

def produce(events):
    for ev in events:
        k.put_record(
            StreamName=STREAM,
            Data=json.dumps(ev).encode(),
            PartitionKey=ev["user_id"],   # routes the record to a shard
        )

The partition key decides which shard a record lands on. Records with the same key go to the same shard and stay ordered - pick a key (e.g. user id) that groups events you need in order. For high throughput, put_records (plural) batches up to 500 records per call.

The consumer

Kinesis is pull-based per shard: get a shard iterator, then loop get_records:

def consume():
    shard_id = k.describe_stream(StreamName=STREAM)["StreamDescription"]["Shards"][0]["ShardId"]
    it = k.get_shard_iterator(StreamName=STREAM, ShardId=shard_id,
                              ShardIteratorType="TRIM_HORIZON")["ShardIterator"]
    while it:
        resp = k.get_records(ShardIterator=it, Limit=100)
        for rec in resp["Records"]:
            handle(json.loads(rec["Data"]))
        it = resp.get("NextShardIterator")

• ShardIteratorType - TRIM_HORIZON starts at the oldest record; LATEST only reads new ones; AT_TIMESTAMP from a point in time.
• Always advance with NextShardIterator - it's the cursor; the old iterator expires.

The concepts that matter

• Shards = throughput + ordering. Each shard handles a fixed write/read rate; add shards to scale. Ordering is guaranteed within a shard, not across the stream.
• Partition key controls shard placement and therefore ordering and hot-shard risk - spread keys to avoid one overloaded shard.
• Checkpointing - in production you track how far each consumer has read (the Kinesis Client Library / Lambda triggers do this) so a restart resumes, not replays.