Streaming Data with Kinesis to S3 and Redshift

A stream is a place data passes through, not a place it lives. Kinesis keeps records for a retention period and then they are gone. For analytics the events have to land somewhere durable and queryable: usually Amazon S3, often Amazon Redshift as well. The path between stream and storage looks trivial on an architecture diagram and contains most of the decisions that determine whether the result is usable.

This guide assumes the stream exists; Amazon Kinesis explained covers streams, shards and partition keys.

The standard path: Firehose

Amazon Data Firehose is the delivery service. You point it at a source, which can be an existing Kinesis data stream, and at a destination, and it moves the data with no consumer code to write (Firehose docs: what is Amazon Data Firehose).

Its central mechanism is buffering. Firehose does not write one S3 object per event. It collects incoming records until a buffer size or a buffer interval is reached, then delivers the batch. For S3 the documented ranges are a size of 1 to 128 MiB and an interval of 60 to 900 seconds, and whichever condition is satisfied first triggers delivery (Firehose docs: buffering hints).

Those two numbers set a trade-off between how fresh the data in S3 is and how many files you create.

What buffering does to a real day

To make the trade-off concrete, we took the 77,033 New York taxi trips picked up on 15 January 2024, treated each as a JSON event arriving at its pickup time, and applied the size-or-time rule under three settings. This is a simulation of the rule in Python, not a Firehose run.

import json, statistics
import pandas as pd

trips = pd.read_parquet("yellow_tripdata_2024-01.parquet",
                        columns=["tpep_pickup_datetime", "PULocationID", "DOLocationID", "fare_amount", "payment_type"])
day = trips[trips["tpep_pickup_datetime"].dt.date == pd.Timestamp("2024-01-15").date()]
day = day.sort_values("tpep_pickup_datetime")

# Each trip becomes one JSON record, arriving at its pickup time
events = [(t.timestamp(), len(json.dumps({"at": t.isoformat(), "pu": int(pu), "do": int(do),
                                          "fare": float(fare), "pay": int(pay)}).encode()) + 1)
          for t, pu, do, fare, pay in day.itertuples(index=False, name=None)]
print(f"{len(events):,} records, {sum(size for _, size in events) / 2**20:.1f} MiB of JSON in the day")


def deliveries(buffer_mib, interval_s):
    """Flush when the buffer reaches the size or the interval, whichever comes first."""
    limit, objects, start, held = buffer_mib * 2**20, [], None, 0
    for at, size in events:
        if start is not None and at - start >= interval_s:      # interval elapsed before this record
            objects.append(held)
            start, held = None, 0
        if start is None:
            start = at
        held += size
        if held >= limit:
            objects.append(held)
            start, held = None, 0
    if held:
        objects.append(held)
    return objects


print()
print(f"{'buffer':>8} {'interval':>9} {'objects':>8} {'median size':>12} {'data up to':>11}")
for buffer_mib, interval_s in [(1, 60), (5, 300), (64, 900)]:
    objects = deliveries(buffer_mib, interval_s)
    print(f"{buffer_mib:>5} MiB {interval_s:>7} s {len(objects):>8,} {statistics.median(objects) / 1024:>9.0f} KiB "
          f"{interval_s / 60:>6.0f} min old")

77,033 records, 5.5 MiB of JSON in the day

  buffer  interval  objects  median size  data up to
    1 MiB      60 s    1,392         4 KiB      1 min old
    5 MiB     300 s      286        20 KiB      5 min old
   64 MiB     900 s       96        60 KiB     15 min old

The whole day is 5.5 MiB of JSON, so no buffer ever fills. The interval triggers every delivery:

• 60 seconds gives 1,392 objects with a median size of 4 KiB.
• 300 seconds gives 286 objects of about 20 KiB.
• 900 seconds, the maximum, still gives 96 objects of about 60 KiB.

This is the normal situation for all but the busiest streams, and it is the opposite of what query engines want. Athena's guidance is to avoid many small files, and Parquet's benefits disappear at these sizes. A stream of this volume produces from about three thousand to over forty thousand tiny objects a month, depending on the setting.

Two conclusions:

• Choose the longest interval the business can accept. Going from one minute to fifteen cut the object count by a factor of about fourteen.
• Plan a compaction step. Buffering alone will not produce well-sized files at modest volumes.

Shaping the data on the way in

Firehose can do three useful things before writing:

• Convert to Parquet. It can convert JSON input to Apache Parquet or ORC before storing it in S3. Other input formats must first be turned into JSON by a Lambda transformation (Firehose docs: format conversion). The conversion needs a schema, which comes from a table in the Glue Data Catalog.
• Partition dynamically. Dynamic partitioning groups records by keys within the data and delivers them to matching S3 prefixes, so queries can prune (Firehose docs: dynamic partitioning). Use a date derived from the event, and keep the key coarse: every additional partition value divides the already small files further.
• Transform with Lambda. A function can clean, enrich or filter records in flight. Keep it light; heavy logic belongs downstream.

Even without these, land the raw events. A raw zone of untouched JSON is the only thing that lets you reprocess when a transformation turns out to be wrong.

Into Redshift: two routes

Route 1: Firehose with a Redshift destination. Firehose delivers the data to S3 first and then issues a Redshift COPY command to load it. This is batch loading on a short cycle, and it needs a table whose columns match the data. Loading data into Redshift with COPY explains what makes that command efficient.

Route 2: Redshift streaming ingestion. Redshift itself can be the stream's consumer. Data flows directly from Kinesis Data Streams, or from Amazon MSK, into a materialized view, with no landing area in S3 in between (Redshift docs: streaming ingestion). The setup is a few SQL statements, written here from the documentation and not run:

CREATE EXTERNAL SCHEMA kds
FROM KINESIS
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftStreamRole';

CREATE MATERIALIZED VIEW trips_stream AUTO REFRESH YES AS
SELECT approximate_arrival_timestamp,
       partition_key,
       JSON_PARSE(kinesis_data) AS payload
FROM kds."trips";

The routes are not exclusive. A common design sends one stream both ways: Firehose to S3 for the durable, replayable record, and streaming ingestion for the low-latency view.

Duplicates and order

Firehose uses at-least-once semantics. The guide notes that delivery retries can introduce duplicates for destination types other than S3, Iceberg tables and Snowflake (Firehose docs: data delivery), which includes Redshift. Producers can also send the same event twice after a timeout. So:

• Give every event an ID at the source.
• Deduplicate in the warehouse on that ID, in the model that reads the landed table.
• Do not rely on arrival order across shards. Carry an event timestamp and order by it.

Batch vs streaming discusses late and out-of-order data further.

The batch job streaming still needs

Almost every streaming landing zone ends up with a scheduled job behind it:

1. Compact the day's small objects into a few large Parquet files per partition.
2. Deduplicate on the event ID.
3. Register the partition in the Glue Data Catalog.
4. Expire the small originals with a lifecycle rule once the compacted copy is checked.

That job is ordinary batch ETL in Glue or Athena. It is what turns a stream's exhaust into a table people can query cheaply; see Parquet and compression for cheaper Athena queries and partitioning data in S3.

A build order

1. Create the data stream and a producer that includes an event ID and event time.
2. Add a Firehose stream reading it, delivering raw JSON to an S3 prefix, with the longest acceptable interval.
3. Query the raw prefix with Athena to confirm data is arriving and well formed.
4. Add Parquet conversion and a date partition, or leave that to the compaction job.
5. Add the nightly compaction and deduplication job.
6. Only then add the Redshift route the use case needs.

Each step is observable before the next is added, which matters with streams: problems that would be obvious in a batch file are easy to miss in a flow.

Real-time streaming pipelines with Amazon Kinesis, S3 and Redshift are covered in the Data Engineering course.

Sources

• What is Amazon Data Firehose? (Amazon Data Firehose Developer Guide) — AWS Documentation
• Configure backup settings (Amazon Data Firehose Developer Guide) — AWS Documentation
• Convert input data format in Amazon Data Firehose (Amazon Data Firehose Developer Guide) — AWS Documentation
• Partition streaming data in Amazon Data Firehose (Amazon Data Firehose Developer Guide) — AWS Documentation
• Understand data delivery in Amazon Data Firehose (Amazon Data Firehose Developer Guide) — AWS Documentation
• Streaming ingestion to a materialized view (Amazon Redshift Database Developer Guide) — AWS Documentation