Retail Events Streaming Pipeline

Stream retail order events through Kafka and land them as Parquet files.

Introduction

30 Second Summary

Every time you tap "Buy" on a retail site, that single click triggers a chain of events. The order, payment, inventory update, and shipping notification all need to flow through the system in real time without losing a single message.

In this project, you will build a local streaming ingestion pipeline that simulates a retail store generating order events in real time. You will publish those events to Apache Kafka, consume them with Python, and land them as analytics-ready Parquet files.

What You'll Build

Start your producer, watch realistic order events stream into Kafka, then see your consumer batch them into compact Parquet files you can query by column without loading the full dataset.

By the end of this project, you'll have:

  • A live event producer you can start and watch order events flow into Kafka in real time, with partition assignment and offset tracking visible in your terminal.
  • A Parquet consumer pipeline that batches events and writes compact files with embedded schema and columnar compression, ready for downstream batch processing.
  • A side-by-side format comparison proving Parquet is 3-5x smaller than JSON, showing the embedded schema, and demonstrating single-column reads without loading the full file.
  • Secret Mission: Add a clickstream event topic with partition-key routing and observe Kafka consumer group rebalancing across multiple consumer instances.

Are there any prerequisites?

This project assumes you have Docker Desktop installed, Python 3.10+ installed, and VS Code ready to go. Basic familiarity with Python and the command line is all you need to get started.

Before We Start

Before We Start...

Before we dive in, let's lock in what you're about to build and why it matters. This step is your moment to commit to the goal before any hands-on work begins.

Set Up Kafka and Your Python Environment

Before you can stream any events, you need two things: a message broker that accepts and stores events, and a Python environment that can talk to it.

In this step, you will spin up Apache Kafka inside a Docker container, then set up your Python packages. By the end, you will have a running broker ready to receive order events from the producer you build in Step 2.

In this step, get ready to:
  • Create a Docker Compose file and start Kafka in KRaft mode.
  • Set up a Python virtual environment and install project dependencies.
  • Verify Kafka connectivity by creating and listing a test topic.
Create your project and start Kafka

You need a project folder on your Desktop and a running Kafka broker before anything else can happen.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir retail-pipeline
cd retail-pipeline
  • Confirm the folder exists by running:
pwd

You should see a path ending in retail-pipeline.

  • Open this folder in VS Code by running:
code .

VS Code didn't open?

If you see command not found, open VS Code manually.

  • Press Cmd+Space (macOS) or the Windows key (Windows) and type Visual Studio Code.
  • Click File in the top menu bar, then Open Folder.
  • Navigate to Desktop and select the retail-pipeline folder.

help me open my project folder in VS Code

  • In VS Code, open the integrated terminal by pressing Ctrl+` (backtick key, top-left of keyboard).

Now you have a terminal inside VS Code that is already pointed at your retail-pipeline folder.

Kafka 4.x uses KRaft mode, which means a single container handles both brokering and controller duties. No ZooKeeper container is needed.

  • In the VS Code Explorer sidebar, click the new file icon and name it docker-compose.yml.
  • Paste the following configuration into docker-compose.yml:
services:
  kafka:
    image: apache/kafka:4.3.1
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_NUM_PARTITIONS: 3

What does this configuration do?

  • apache/kafka:4.3.1 pulls the official Kafka image with CLI tools included.
  • KAFKA_PROCESS_ROLES: broker,controller runs both roles in one container (KRaft combined mode).
  • KAFKA_LISTENERS binds to all interfaces inside the container, while KAFKA_ADVERTISED_LISTENERS tells your Python client to connect via localhost:9092.
  • KAFKA_NUM_PARTITIONS: 3 sets the default partition count for new topics.
  • Save docker-compose.yml with Cmd+S (macOS) or Ctrl+S (Windows).
  • Start the Kafka container by running this in your VS Code terminal:
docker compose up -d
  • Confirm the container is running:
docker ps

You should see a container named kafka with status Up and port 9092 mapped.

Container not running or port conflict?

  • If you see no containers, check that Docker Desktop is running on your machine.
  • If port 9092 is already in use, stop whatever process is using it or change the port mapping in docker-compose.yml.
  • Run docker compose logs kafka to see the container's startup output for specific errors.

help me fix my Kafka Docker container that won't start

✔️ Awesome, I've got everything!

Great. Make sure you have saved docker-compose.yml and your Kafka container shows Up in docker ps.

ⓧ I'd like to double check the full code

services:
  kafka:
    image: apache/kafka:4.3.1
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_NUM_PARTITIONS: 3
Set up your Python environment

A virtual environment keeps this project's packages isolated from the rest of your system.

  • Create and activate a virtual environment by running these commands in your VS Code terminal:
python3 -m venv venv
source venv/bin/activate

What does this do?

python3 -m venv venv creates a venv folder containing a self-contained Python installation. source venv/bin/activate activates it so pip install puts packages in that folder instead of your global Python.

Your terminal prompt should now show (venv) at the beginning, confirming activation.

On Windows and not using macOS?

Replace the second command with:

venv\Scripts\activate

help me activate my Python virtual environment on Windows

  • Create a requirements.txt file in the VS Code Explorer sidebar (click the new file icon) and paste this:
confluent-kafka==2.15.0
Faker==40.36.0
pyarrow==25.0.0

What are these packages?

  • confluent-kafka is a high-performance Python client for Kafka, backed by the C library librdkafka.
  • Faker generates realistic fake data (names, addresses, UUIDs) for your simulated order events.
  • pyarrow reads and writes Parquet files with schema enforcement and columnar compression.
  • Save requirements.txt and install the packages by running:
pip install -r requirements.txt

You should see output ending with Successfully installed followed by the three package names and their versions.

Installation failed?

  • If you see a permissions error, make sure your virtual environment is activated (look for (venv) in your prompt).
  • If confluent-kafka fails to build, you may need to install system-level build tools. On macOS, run xcode-select --install first.

help me install confluent-kafka when pip install fails

✔️ Awesome, I've got everything!

Great. Confirm you saved requirements.txt and all three packages installed successfully.

ⓧ I'd like to double check the full code

confluent-kafka==2.15.0
Faker==40.36.0
pyarrow==25.0.0
Verify Kafka connectivity

Your container is running, but you need to confirm Kafka is actually accepting commands from the outside.

  • Create a test topic inside the Kafka container by running:
docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test-topic --partitions 1 --replication-factor 1

You should see a message confirming the topic was created.

What does this command do?

docker exec kafka runs a command inside your running container. kafka-topics.sh is Kafka's built-in CLI tool for managing topics. The --bootstrap-server localhost:9092 flag tells it which broker to talk to.

Before you run the next command, predict what you will see. What topic name should appear in the list?

  • List all topics to confirm Kafka processed your request:
docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

You should see test-topic printed in the output. This proves your Kafka broker is running, accepting connections, and storing topic metadata.

Not seeing the topic or getting connection errors?

  • If you get a connection refused error, wait 10-15 seconds and retry. Kafka can take a moment to finish its startup sequence.
  • If docker exec says the container is not running, go back and run docker compose up -d again.
  • Run docker compose logs kafka and look for lines mentioning KRaft or Broker to confirm the broker started.

help me troubleshoot my Kafka broker connection

Your Kafka broker is running and your Python packages are installed. Next up, you will build the event producer that generates realistic retail order data and streams it into Kafka in real time.

Build and Run the Event Producer

In Step 1 you stood up a Kafka broker and installed your Python packages. Now you have infrastructure, but nothing is flowing through it.

Without events arriving in Kafka, there is nothing for a downstream consumer to process. In this step you will write a Python producer that simulates a retail store placing orders in real time, giving your pipeline its first live data source.

In this step, get ready to:
  • Write a producer that generates realistic order events with Faker and publishes them to Kafka.
  • Run the producer and watch events arrive in real time using Kafka's built-in console consumer.
Write the event data model

The producer needs two things: a catalog of products to sell and a function that assembles a random order from that catalog. You will build both now.

  • Create a new file called producer.py in your project folder. Add the imports and product catalog by pasting this code:
import json
import time
import random
from datetime import datetime, timezone

from faker import Faker
from confluent_kafka import Producer

fake = Faker()

PRODUCTS = [
    {"id": "PROD-001", "name": "Wireless Mouse", "category": "Electronics", "price": 29.99},
    {"id": "PROD-002", "name": "USB-C Hub", "category": "Electronics", "price": 49.99},
    {"id": "PROD-003", "name": "Standing Desk Mat", "category": "Office", "price": 39.99},
    {"id": "PROD-004", "name": "Mechanical Keyboard", "category": "Electronics", "price": 89.99},
    {"id": "PROD-005", "name": "Notebook Set", "category": "Office", "price": 12.99},
    {"id": "PROD-006", "name": "LED Desk Lamp", "category": "Home", "price": 34.99},
    {"id": "PROD-007", "name": "Coffee Mug Warmer", "category": "Home", "price": 19.99},
    {"id": "PROD-008", "name": "Webcam HD", "category": "Electronics", "price": 59.99},
]

What does this code do?

  • The imports bring in Faker for generating realistic fake data and confluent-kafka's Producer class for publishing messages.
  • PRODUCTS is a list of 8 sample retail products, each with an ID, name, category, and price. The producer will pick randomly from this catalog for every order.
  • Below the PRODUCTS list, add the event generation function and the delivery callback:
def generate_order_event():
    # Pick a random product and quantity
    product = random.choice(PRODUCTS)
    quantity = random.randint(1, 5)
    return {
        "order_id": fake.uuid4(),
        "customer_id": f"CUST-{random.randint(1000, 9999)}",
        "customer_name": fake.name(),
        "product_id": product["id"],
        "product_name": product["name"],
        "category": product["category"],
        "quantity": quantity,
        "unit_price": product["price"],
        "total_amount": round(product["price"] * quantity, 2),
        "order_timestamp": datetime.now(timezone.utc).isoformat(),
        "shipping_address": fake.address().replace("\n", ", "),
    }


def delivery_report(err, msg):
    # Kafka calls this once per message to confirm delivery
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}")

What does this code do?

  • generate_order_event() assembles a single order with a UUID, a random customer, a random product, a calculated total, a UTC timestamp, and a shipping address.
  • delivery_report() is a callback that Kafka invokes after each message is acknowledged. It prints the topic, partition number, and offset so you can see exactly where each event lands.
  • Save producer.py.
  • Verify the data model works by running this quick test in your terminal:
python -c "from producer import generate_order_event; import json; print(json.dumps(generate_order_event(), indent=2))"

You should see a single JSON object printed with fields like order_id, customer_name, total_amount, and shipping_address.

Seeing an ImportError?

  • Make sure your virtual environment is activated. You should see the environment name in your terminal prompt.
  • Confirm you are running the command from the same folder where producer.py lives.
  • help me debug an ImportError when running my producer
Add the producer loop and run it

The data model is proven. Now you need the loop that continuously generates events and publishes them to Kafka.

  • At the bottom of producer.py, below delivery_report(), add the main function and entry point:
def main():
    # Connect to the local Kafka broker
    conf = {"bootstrap.servers": "localhost:9092"}
    producer = Producer(conf)

    print("Starting order event producer...")
    print("Press Ctrl+C to stop.\n")

    try:
        event_count = 0
        while True:
            event = generate_order_event()
            producer.produce(
                topic="orders",
                key=event["customer_id"],
                value=json.dumps(event),
                callback=delivery_report,
            )
            producer.poll(0)
            event_count += 1
            if event_count % 10 == 0:
                print(f"  [{event_count} events produced]")
            time.sleep(random.uniform(0.1, 0.5))
    except KeyboardInterrupt:
        print(f"\nStopping producer. Total events: {event_count}")
    finally:
        producer.flush()


if __name__ == "__main__":
    main()

What does this code do?

  • Producer(conf) creates a Kafka producer connected to your local broker on port 9092.
  • The while True loop generates an event, serializes it as JSON, and publishes it to the orders topic. The message key is customer_id so Kafka routes all events for the same customer to the same partition.
  • producer.poll(0) triggers delivery callbacks without blocking. Every 10 events, a progress counter prints so you can see the producer is alive.
  • When you press Ctrl+C, the KeyboardInterrupt is caught, and producer.flush() ensures all buffered messages are sent before the process exits.
  • Save producer.py.
  • Start the producer by running:
python producer.py

You should see delivery confirmations scrolling in your terminal, showing the topic name, partition number (0, 1, or 2), and an incrementing offset for each message.

  • Let it run for about 10 seconds, then press Ctrl+C to stop it.

Seeing a connection refused error?

  • Make sure your Kafka Docker container is still running. Run docker ps and confirm you see a container named kafka with status Up.
  • If the container stopped, restart it with docker compose up -d from your project folder and wait 5 seconds before retrying.
  • help me fix a Kafka connection refused error in my Python producer
Verify events in Kafka with the console consumer

The delivery confirmations prove messages left your producer. Now you will read them back directly from Kafka to confirm they actually arrived in the topic.

  • Open a second terminal in VS Code by pressing Ctrl+Shift+` (backtick).
  • Before you run the next command, predict: will you see raw JSON objects or something else?
  • Read the first 5 events from the orders topic by running:
docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --from-beginning --max-messages 5

You should see 5 JSON order events printed to your terminal. Each line is a complete JSON object with fields like order_id, customer_id, product_name, and total_amount.

What do the partition and offset numbers mean?

When you ran the producer, each delivery confirmation showed something like Delivered to orders [1] @ offset 7. The number in brackets is the partition (0, 1, or 2 since you configured 3 partitions).

The offset is the sequential position of that message within its partition. Because you use customer_id as the message key, Kafka hashes the key to decide which partition receives the message. All orders from the same customer always land on the same partition, preserving their ordering.

This key-based routing is essential for downstream consumers that need to process events for a single customer in order.

Seeing 0 messages or a timeout?

  • If you never started producer.py, the topic exists but has no data. Run python producer.py in your first terminal for 10 seconds, stop it, then retry the consumer command.
  • Check that the topic name matches exactly. Run docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list to see available topics.
  • help me troubleshoot why kafka-console-consumer shows no messages

✔️ Awesome, I've got everything!

Great. Double check you saved producer.py and confirm it runs without errors.

ⓧ I'd like to double check the full code

Scroll up and confirm your producer.py contains these sections in order:

  • Imports (json, time, random, datetime, Faker, confluent_kafka Producer) plus fake = Faker().
  • The PRODUCTS list with 8 product dictionaries.
  • The generate_order_event() function returning a dictionary with 11 fields.
  • The delivery_report() callback.
  • The main() function with the infinite loop and if __name__ == "__main__" guard at the bottom.

Your pipeline now has a live data source pumping realistic order events into Kafka. Next up, you will build a consumer that reads these events and writes them to disk as JSON files.

Build a JSON Consumer

Your producer is generating order events into Kafka. Now you need something on the other side to read those events and store them on disk.

The simplest approach is to write a consumer that dumps raw JSON files. In this step, you will build that consumer, run it alongside the producer, and observe how JSON performs as a landing format for streaming data.

In this step, get ready to:
  • Write a Kafka consumer that buffers events and flushes JSON Lines files.
  • Run the producer and consumer together and inspect the output files.
Write the consumer helper functions

The consumer needs two helpers: one to connect to Kafka with the right configuration, and one to write a batch of events to a file on disk.

  • Create a new file called consumer_json.py in your project folder, and add the imports, constants, and helper functions:
import json
import os

from confluent_kafka import Consumer, KafkaError

OUTPUT_DIR = "output/json"
BATCH_SIZE = 50


def create_consumer():
    conf = {
        "bootstrap.servers": "localhost:9092",
        "group.id": "order-json-writer",
        "auto.offset.reset": "earliest",
        "enable.auto.commit": True,
    }
    return Consumer(conf)


def write_json_batch(events, batch_number):
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    filename = f"orders_batch_{batch_number:04d}.jsonl"
    filepath = os.path.join(OUTPUT_DIR, filename)

    with open(filepath, "w") as f:
        for event in events:
            f.write(json.dumps(event) + "\n")

    file_size = os.path.getsize(filepath)
    print(f"Wrote {filepath} ({len(events)} events, {file_size:,} bytes)")
    return file_size

What does this code do?

Two helpers split the work cleanly:

  • create_consumer() connects to your local Kafka broker with consumer group order-json-writer. The auto.offset.reset: earliest setting means the consumer starts reading from the very first message if no prior offset exists.
  • write_json_batch() takes a list of events and writes them as JSON Lines (one JSON object per line) into the output/json/ directory. It also prints the file size so you can track how large each batch is.
  • Confirm the file was created by running:
ls consumer_json.py

You should see consumer_json.py listed in the output.

Add the main consumer loop

The main function ties everything together. It polls Kafka for new messages, buffers them, and flushes a file every 50 events.

  • Add the following main() function below write_json_batch() in consumer_json.py:
def main():
    consumer = create_consumer()
    consumer.subscribe(["orders"])

    print("Starting JSON consumer...")
    print(f"Writing batches of {BATCH_SIZE} events to {OUTPUT_DIR}/")
    print("Press Ctrl+C to stop.\n")

    batch = []
    batch_number = 0
    total_events = 0
    total_bytes = 0

    try:
        while True:
            msg = consumer.poll(1.0)

            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                print(f"Consumer error: {msg.error()}")
                continue

            event = json.loads(msg.value().decode("utf-8"))
            batch.append(event)
  • Directly below batch.append(event), still inside the while True loop, add the batch-flush logic and shutdown handling:
            if len(batch) >= BATCH_SIZE:
                file_size = write_json_batch(batch, batch_number)
                total_events += len(batch)
                total_bytes += file_size
                batch_number += 1
                batch = []

    except KeyboardInterrupt:
        if batch:
            file_size = write_json_batch(batch, batch_number)
            total_events += len(batch)
            total_bytes += file_size

        print(f"\nStopping consumer.")
        print(f"Total: {total_events} events in {batch_number + 1} files ({total_bytes:,} bytes)")
    finally:
        consumer.close()


if __name__ == "__main__":
    main()

What does the main loop do?

The consumer polls Kafka once per second. Each message is deserialized from bytes back to a Python dictionary and added to an in-memory buffer.

When the buffer hits 50 events, it flushes a .jsonl file and resets. On Ctrl+C, any remaining events are flushed before the consumer disconnects from the consumer group gracefully.

  • Save consumer_json.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file, then move on to the next substep where you run the pipeline.

ⓧ I'd like to double check the full code

Your consumer_json.py should contain these sections in order:

  • Imports (json, os, Consumer, KafkaError).
  • Constants: OUTPUT_DIR = "output/json" and BATCH_SIZE = 50.
  • create_consumer() function returning a configured Consumer.
  • write_json_batch() function writing events to a .jsonl file.
  • main() function with the poll loop, batch flush, and KeyboardInterrupt handler.
  • if __name__ == "__main__": main() entry point at the bottom.

Scroll up and confirm each section is present and matches the code blocks above.

Seeing a syntax error when you save?

Make sure the if len(batch) >= BATCH_SIZE: block is indented at the same level as batch.append(event) (three levels of indentation inside while True). The except KeyboardInterrupt: should align with the try: keyword.

help me fix indentation in my consumer_json.py

Run the pipeline and inspect the output

Now you will run the producer and consumer side by side. The consumer will read events from Kafka and flush them as JSON files to disk.

  • In your current terminal, start the producer by running:
python producer.py
  • Open a second terminal tab or window.
  • Activate your virtual environment in the new terminal and navigate to your project folder.
  • Start the JSON consumer by running:
python consumer_json.py

You should see the consumer print messages like Wrote output/json/orders_batch_0000.jsonl (50 events, X bytes) each time it flushes a batch.

  • Let both run for about 30 seconds until 2-3 batch files are written.
  • Stop the consumer first with Ctrl+C, then stop the producer with Ctrl+C.

Before you inspect the output, predict: how large do you think each 50-event JSON file will be?

  • Check the output directory by running:
ls -lh output/json/

You should see files like orders_batch_0000.jsonl and orders_batch_0001.jsonl, each around 15-20 KB for 50 events.

What are the limitations you just saw?

The JSON consumer works, but look at what is missing:

  • No schema metadata. Open one of those .jsonl files. Nothing in the file tells you the data types of each field. A downstream system has to guess whether quantity is an integer or a string.
  • Full-file parsing required. If you only need the total_amount column, you still have to parse every single line of the file to extract it.
  • No compression. Each event stores all field names as plain text, repeated on every line. This bloat adds up fast at scale.

Consumer exits immediately without writing files?

Make sure your Kafka container is still running. Run docker ps and confirm you see the kafka container listed. If it is not running, start it again with docker compose up -d.

Also confirm the producer has published events. If you start the consumer before any messages exist, it will sit idle until the producer sends data.

help me debug why my JSON consumer is not receiving messages from Kafka

Your JSON consumer works. But those 15-20 KB files full of repeated field names, with no schema and no way to read a single column efficiently, are exactly the problem that a better file format can solve. Next up, you will switch to Parquet and see the difference firsthand.

Switch to Parquet for Analytics-Ready Output

In the previous step, your JSON consumer successfully landed Kafka events to disk. But you also saw the downsides: bloated file sizes, no embedded schema, and no way to read a single field without parsing every line.

Parquet solves all three problems at once. It stores data in columns instead of rows, which means compression shrinks files 3-5x. The schema is baked into the file header, so readers never have to guess field types. And you can load just the columns you need without touching the rest.

In this step, get ready to:
  • Write consumer_parquet.py that outputs schema-enforced Parquet files.
  • Run the Parquet consumer alongside the producer to accumulate output files.
  • Compare JSON and Parquet output side by side with compare_formats.py.
Write the Parquet consumer

The Parquet consumer follows the same Kafka polling pattern as your JSON consumer. The key difference is the output layer: instead of dumping raw JSON strings, you define an explicit schema with PyArrow and write typed, compressed columnar files.

  • Create a new file called consumer_parquet.py in your project folder.
  • Add the imports, configuration, and schema definition by pasting this code:
import json
import os

from confluent_kafka import Consumer, KafkaError
import pyarrow as pa
import pyarrow.parquet as pq

OUTPUT_DIR = "output/parquet"
BATCH_SIZE = 50

ORDER_SCHEMA = pa.schema([
    ("order_id", pa.string()),
    ("customer_id", pa.string()),
    ("customer_name", pa.string()),
    ("product_id", pa.string()),
    ("product_name", pa.string()),
    ("category", pa.string()),
    ("quantity", pa.int32()),
    ("unit_price", pa.float64()),
    ("total_amount", pa.float64()),
    ("order_timestamp", pa.string()),
    ("shipping_address", pa.string()),
])


def create_consumer():
    conf = {
        "bootstrap.servers": "localhost:9092",
        "group.id": "order-parquet-writer",
        "auto.offset.reset": "earliest",
        "enable.auto.commit": True,
    }
    return Consumer(conf)

What does this code do?

ORDER_SCHEMA declares the exact shape of your order data using PyArrow types. When you write a Parquet file with this schema, the types are embedded in the file header. Any downstream tool (Spark, DuckDB, Pandas) can read the schema without parsing a single row.

The consumer group is order-parquet-writer instead of order-json-writer. This means both consumers can run simultaneously against the same topic without competing for messages.

  • Below create_consumer(), add the function that converts a batch of events into a Parquet file:
def write_parquet_batch(events, batch_number):
    os.makedirs(OUTPUT_DIR, exist_ok=True)

    # Reshape row-oriented events into column-oriented dict
    columns = {field.name: [] for field in ORDER_SCHEMA}
    for event in events:
        for field in ORDER_SCHEMA:
            columns[field.name].append(event.get(field.name))

    # Build a PyArrow table and write to Parquet
    table = pa.table(columns, schema=ORDER_SCHEMA)
    filename = f"orders_batch_{batch_number:04d}.parquet"
    filepath = os.path.join(OUTPUT_DIR, filename)
    pq.write_table(table, filepath)

    file_size = os.path.getsize(filepath)
    print(f"Wrote {filepath} ({len(events)} events, {file_size:,} bytes)")
    return file_size

Why reshape rows into columns?

Kafka messages arrive as rows (one dict per event). Parquet stores data in columns (all values for one field together). The loop transposes the row-oriented list into a column-oriented dictionary that PyArrow can convert directly into a columnar table.

  • Save consumer_parquet.py and verify the imports and schema are valid by running:
python -c "import consumer_parquet; print('Imports OK, schema has', len(consumer_parquet.ORDER_SCHEMA), 'fields')"

You should see Imports OK, schema has 11 fields printed to your terminal.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated. Check that you installed pyarrow in Step 1. Run pip list | grep pyarrow to confirm it appears.

help me fix a ModuleNotFoundError for pyarrow in my consumer_parquet.py

Now add the main consumer loop. This is the same polling pattern you wrote in consumer_json.py with one difference: it calls write_parquet_batch instead of write_json_batch.

  • Add the main() function below write_parquet_batch:
def main():
    consumer = create_consumer()
    consumer.subscribe(["orders"])

    print("Starting Parquet consumer...")
    print(f"Writing batches of {BATCH_SIZE} events to {OUTPUT_DIR}/")
    print("Press Ctrl+C to stop.\n")

    batch = []
    batch_number = 0
    total_events = 0
    total_bytes = 0

    try:
        while True:
            msg = consumer.poll(1.0)

            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                print(f"Consumer error: {msg.error()}")
                continue

            event = json.loads(msg.value().decode("utf-8"))
            batch.append(event)

            if len(batch) >= BATCH_SIZE:
                file_size = write_parquet_batch(batch, batch_number)
                total_events += len(batch)
                total_bytes += file_size
                batch_number += 1
                batch = []
  • Add the shutdown handler and entry point directly below the poll loop:
    except KeyboardInterrupt:
        if batch:
            file_size = write_parquet_batch(batch, batch_number)
            total_events += len(batch)
            total_bytes += file_size

        print(f"\nStopping consumer.")
        print(f"Total: {total_events} events in {batch_number + 1} files ({total_bytes:,} bytes)")
    finally:
        consumer.close()


if __name__ == "__main__":
    main()

What does the shutdown handler do?

When you press Ctrl+C, the KeyboardInterrupt block flushes any remaining events in the batch to a final Parquet file so you never lose in-flight data. The finally block ensures the Kafka consumer is cleanly closed, releasing its partition assignments back to the consumer group.

  • Save consumer_parquet.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm your consumer_parquet.py contains these sections in order:

  • Imports (json, os, confluent_kafka, pyarrow) plus OUTPUT_DIR, BATCH_SIZE, and ORDER_SCHEMA with 11 fields.
  • create_consumer() returning a Consumer with group order-parquet-writer.
  • write_parquet_batch() that reshapes rows into columns and calls pq.write_table().
  • main() with the polling loop, KeyboardInterrupt handler, and if __name__ == "__main__" entry point.
Run the Parquet consumer

You need at least 100 events in Parquet format so the format comparison script has meaningful data to work with.

  • In your first terminal, start the producer by running:
python producer.py
  • Open a second terminal tab (make sure your virtual environment is activated in it too).
  • Start the Parquet consumer by running:
python consumer_parquet.py

You should see messages like Wrote output/parquet/orders_batch_0000.parquet (50 events, X bytes) appearing every time 50 events accumulate.

  • Let both run until you see at least 2 batch files written (100+ events).
  • Press Ctrl+C in the consumer terminal first, then Ctrl+C in the producer terminal.
  • Confirm the Parquet files exist by running:
ls -lh output/parquet/

You should see .parquet files listed, each noticeably smaller than the .jsonl files from the previous step.

Consumer hangs with no output?

If the consumer subscribes but never prints a batch, check that the producer is running in the other terminal and producing events. Also verify your Kafka container is still up with docker ps.

If you see a KafkaError about broker unavailability, restart the container with docker compose up -d.

help me debug why my Parquet consumer is not receiving messages from Kafka

Compare JSON and Parquet side by side

You have both formats on disk now. A comparison script makes the difference concrete: file sizes, schema visibility, and column-level access.

  • Create a new file called compare_formats.py in your project folder.
  • Add the imports and helper function by pasting this code:
import os

import pyarrow.parquet as pq

JSON_DIR = "output/json"
PARQUET_DIR = "output/parquet"


def get_dir_size(directory):
    # Walk through files and sum their sizes
    total = 0
    count = 0
    for f in os.listdir(directory):
        filepath = os.path.join(directory, f)
        if os.path.isfile(filepath):
            total += os.path.getsize(filepath)
            count += 1
    return total, count

What does get_dir_size do?

It scans a directory and returns the total byte count plus number of files. You will use it to compare the raw size of your JSON output against Parquet output.

  • Save compare_formats.py and test the helper against your JSON output by running:
python -c "from compare_formats import get_dir_size; print(get_dir_size('output/json'))"

You should see a tuple like (23456, 2) showing total bytes and file count for your JSON directory.

  • Now add the main() function that prints the size comparison and compression ratio:
def main():
    print("=" * 60)
    print("FILE FORMAT COMPARISON: JSON vs Parquet")
    print("=" * 60)

    if not os.path.exists(JSON_DIR):
        print(f"\nNo JSON output found at {JSON_DIR}")
        return
    if not os.path.exists(PARQUET_DIR):
        print(f"\nNo Parquet output found at {PARQUET_DIR}")
        return

    json_size, json_files = get_dir_size(JSON_DIR)
    print(f"\nJSON Lines:")
    print(f"  Files: {json_files}")
    print(f"  Total size: {json_size:,} bytes")

    parquet_size, parquet_files = get_dir_size(PARQUET_DIR)
    print(f"\nParquet:")
    print(f"  Files: {parquet_files}")
    print(f"  Total size: {parquet_size:,} bytes")

    if parquet_size > 0:
        ratio = json_size / parquet_size
        print(f"\n  Compression ratio: {ratio:.1f}x smaller with Parquet")
  • Add the schema inspection and column-level access demonstration below the size comparison, then close with the entry point:
    print(f"\n  Embedded schema (no guessing!):")
    first_file = os.path.join(PARQUET_DIR, sorted(os.listdir(PARQUET_DIR))[0])
    schema = pq.read_schema(first_file)
    for field in schema:
        print(f"    {field.name}: {field.type}")

    print(f"\n  Column-level access (read only 'total_amount'):")
    table = pq.read_table(first_file, columns=["total_amount"])
    print(f"    Read {len(table)} rows, only 1 column loaded into memory")
    print(f"    Sum of total_amount: ${table.column('total_amount').to_pylist()[-1]}")


if __name__ == "__main__":
    main()

What are pq.read_schema and pq.read_table doing?

pq.read_schema() reads ONLY the file header metadata. It returns the full schema (field names and types) without loading any data rows. This is impossible with JSON, where you must read the entire file to discover what fields exist.

pq.read_table(path, columns=["total_amount"]) loads a single column into memory. Because Parquet stores each column contiguously on disk, the reader skips all other columns entirely. With JSON, you would have to parse every field of every row just to access one.

  • Save compare_formats.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before running it.

ⓧ I'd like to double check the full code

Scroll up and confirm your compare_formats.py contains these sections in order:

  • Imports (os, pyarrow.parquet) plus JSON_DIR and PARQUET_DIR constants.
  • get_dir_size() helper that returns a (total, count) tuple.
  • main() printing size comparison, compression ratio, embedded schema via pq.read_schema(), and column-level read via pq.read_table().
  • if __name__ == "__main__" entry point calling main().

Before you run the comparison, think about this: based on what you know about columnar compression and JSON repetition, how much smaller do you expect the Parquet files to be?

  • Run the comparison script:
python compare_formats.py

You should see output showing Parquet is 3-5x smaller than JSON, a full schema listing with typed fields, and a single-column read that loaded only total_amount without touching other columns.

Why does this matter for real pipelines?

At scale, that 3-5x compression means terabytes become hundreds of gigabytes, cutting storage costs proportionally. The embedded schema means downstream jobs (Spark, Athena, BigQuery) never fail because a field was renamed or a type changed silently. Column-level access means analytical queries scan only the columns they need, reducing I/O by orders of magnitude.

Seeing 'No Parquet output found'?

This means the output/parquet/ directory is empty or missing. Go back and re-run the producer and consumer_parquet.py together for at least 30 seconds to accumulate batch files.

help me figure out why compare_formats.py cannot find my Parquet output files

Your pipeline now produces analytics-ready output. The Parquet files are compact, self-describing, and column-accessible. Next up in the Secret Mission, you will add a second event topic with partition-key routing and observe how Kafka distributes load across consumer instances.

Secret mission

Clickstream Topic with Partition Routing

Real pipelines handle multiple event types. Can you add a second Kafka topic with deliberate partition-key routing and observe how consumer groups coordinate load across multiple instances?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. However, the Kafka Docker container uses memory and disk while running.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker container running Apache Kafka (uses memory and disk while active).
  • Output files in output/json/, output/parquet/, and output/clickstream/.
  • Python virtual environment and project files (producer.py, consumers, docker-compose.yml, etc.).

Keep everything running

No action needed. Choose this if you plan to continue into Project 2 (batch processing with Spark), which connects directly to this Kafka broker and the Parquet files you produced.

  • Your Kafka container stays running and retains all topic data.
  • Your output files remain on disk for inspection or downstream processing.
  • Your virtual environment stays ready for the next session.

Pause - I'll come back to this later

Shut down running processes to free up memory while keeping all files and container state intact. You can resume later without reconfiguring anything.

  • Stop any running producer or consumer scripts with Ctrl+C in each terminal.
  • Stop the Kafka container by running:
docker compose stop
  • When you are ready to resume, run docker compose start from the same project folder to bring Kafka back.

Delete - I don't want to use this again

Remove all project resources and start fresh. This removes the Kafka container, its stored data, and all output files.

  • Stop any running producer or consumer scripts with Ctrl+C in each terminal.
  • Stop and remove the Kafka container and its volumes by running:
docker compose down -v
  • Delete all output data by running:
rm -rf output/
  • Deactivate and remove the Python virtual environment by running:
deactivate
cd ~/Desktop
rm -rf kafka-streaming-pipeline

What does docker compose down -v do?

The docker compose down command stops and removes the containers and networks defined in your Compose file. The -v flag also removes any named volumes, wiping Kafka's stored topic data completely.

That's a Wrap!

That's a Wrap!

Nice work. You just built a complete streaming ingestion pipeline that takes raw retail events from a Kafka broker and lands them as analytics-ready Parquet files.

You've learned how to:

  • Build a Python event producer using Faker to generate realistic retail order data and publish it to Apache Kafka in real time.
  • Run Kafka 4.3.1 in KRaft mode via Docker with zero ZooKeeper dependency.
  • Prove why Parquet beats JSON for analytics by building consumers for both formats and comparing file size, embedded schema, and column-level access.
  • Secret Mission: handled multiple event types across topics with partition-key routing and consumer group coordination.

Ready to quiz yourself?