How to Build an AWS Kinesis Streaming Sink to Parquet

The streaming-to-lakehouse pattern

An AWS Kinesis streaming pipeline that lands raw events in a data lake does one job: read a stream, batch the records, and write them as columnar Parquet files a query engine can scan. Kinesis Firehose is the managed version of this; a Kafka consumer writing Parquet is the open-source version. The logic is identical, so building it by hand teaches you exactly what Firehose or a Kafka Connect S3 Sink does under the hood.

In this project a Kafka topic carries clickstream events, each a JSON object like {"event_type": "click", "user_id": 42, "ts": "2024-01-15T09:31:00"}. The goal is to archive them to lake/events/date=YYYY-MM-DD/ so analysts can query with DuckDB, Spark, or Athena. Two rules make it production-grade: batch by date, and never lose an event.

Consume the topic with auto-commit OFF

The delivery guarantee lives in one setting. With enable_auto_commit=False you decide when an offset is marked done - and you only mark it done after the Parquet file is safely on disk. That is at-least-once delivery: a crash mid-batch re-reads a few events but drops none.

import json, os
from collections import defaultdict
from kafka import KafkaConsumer
import pyarrow as pa
import pyarrow.parquet as pq

BROKER = os.environ["KAFKA_BROKER"]
TOPIC = f"{os.environ['USER_ID']}-events"
BATCH_SIZE = 100

consumer = KafkaConsumer(
    TOPIC,
    bootstrap_servers=BROKER,
    group_id=f"{os.environ['USER_ID']}-parquet-sink",
    auto_offset_reset="earliest",
    enable_auto_commit=False,      # commit manually, after the write
    value_deserializer=lambda b: json.loads(b),
    consumer_timeout_ms=5000,      # iterator exits when the topic drains
)

consumer_timeout_ms=5000 is what lets the sink finish: when no message arrives for five seconds the for msg in consumer loop ends instead of blocking forever.

Buffer by date and flush to Parquet

Group events into an in-memory dict keyed by the date derived from each ts, then flush each date's list to its own Parquet file. pyarrow.Table.from_pylist builds a table straight from a list of dicts, and write_table(..., compression="snappy") handles the columnar encoding.

buffers = defaultdict(list)
last_offset = 0

def flush(date_str, events, end_offset):
    out_dir = f"lake/events/date={date_str}"
    os.makedirs(out_dir, exist_ok=True)
    out_path = f"{out_dir}/part-{end_offset:010d}.parquet"
    table = pa.Table.from_pylist(events)
    pq.write_table(table, out_path, compression="snappy")

for msg in consumer:
    last_offset = msg.offset
    date_str = msg.value["ts"][:10]      # "2024-01-15"
    buffers[date_str].append(msg.value)
    if sum(len(v) for v in buffers.values()) >= BATCH_SIZE:
        for d, evs in list(buffers.items()):
            if evs:
                flush(d, evs, last_offset)
                buffers[d] = []
        consumer.commit()               # commit AFTER the writes

Putting the offset in the filename (part-0000000149.parquet) keeps successive flushes from colliding when the same date appears in more than one batch.

Drain the tail and commit

When the loop exits on the timeout, some events are still buffered. Flush whatever remains, then commit one last time:

for d, evs in buffers.items():
    if evs:
        flush(d, evs, last_offset)
consumer.commit()
consumer.close()

Read it back to confirm every event landed and the partitions are correct:

python3 -c "import duckdb; print(duckdb.execute(\"SELECT date, count(*) FROM read_parquet('lake/events/**/*.parquet', hive_partitioning=1) GROUP BY date\").fetchall())"

That is the whole streaming sink: a topic in, three date-partitioned Parquet files out, with no lost events. Swap Kafka for a Kinesis stream and the exact same batch-then-write loop still applies.