Real-Time Data Pipeline with Kafka

Build a streaming pipeline that ingests events through Kafka into MinIO and PostgreSQL.

Introduction

30 Second Summary

Every time you order something online, that click triggers a chain of events. Your order flows through systems that track inventory, process payments, and update delivery status in real time.

In this project, you will build a complete real-time data pipeline that processes e-commerce order events using Apache Kafka, stores raw data in a MinIO data lake, and loads transformed results into PostgreSQL for analytics.

What You'll Build

A real-time e-commerce order processing pipeline running entirely on your local machine using Docker containers.

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

  • A Docker Compose stack running Kafka, MinIO, and PostgreSQL as containerized infrastructure.
  • A Python event producer that generates realistic e-commerce order data and publishes it to a Kafka topic.
  • A Python stream consumer that reads events from Kafka, stores raw JSON in a data lake, and loads transformed data into a warehouse for querying.
  • Secret Mission: Add a dead-letter queue for failed events and implement retry logic in the consumer.

Are there any prerequisites?

This project assumes you have basic Python programming knowledge and are comfortable using the command line. You also need Docker and Docker Compose installed on your machine.

All infrastructure (Kafka, MinIO, PostgreSQL) runs inside Docker containers, so you do not need to install any of these services directly.

Before We Start

Real-time data pipelines power every modern platform you interact with daily. Every time you place an online order, that event flows through a message broker, lands in a data lake for durability, and gets transformed into an analytics-ready table.

In this step, you'll create your project workspace and install the Python packages that connect to Kafka, MinIO, and PostgreSQL.

In this step, get ready to:
  • Create your project directory and dependencies file.
  • Set up a Python virtual environment and install all required packages.

How does the pipeline work?

Your pipeline has three core components, each running in its own Docker container.

  • Apache Kafka is the message broker. It receives order events from the producer and queues them for consumers to process.
  • MinIO is your data lake. It stores raw JSON event files partitioned by date, just like AWS S3.
  • PostgreSQL is your analytics warehouse. It holds transformed order summaries ready for SQL queries.
Create your project directory
  • Open your terminal.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir kafka-pipeline
cd kafka-pipeline
  • Confirm the folder was created by running:
pwd

You should see /home/<your-username>/Desktop/kafka-pipeline in the output.

Create the requirements file

A requirements.txt file lists all the Python packages your project needs, pinned to exact versions so every install is reproducible.

  • Open the kafka-pipeline folder in your code editor.
  • Create a new file called requirements.txt and paste the following:
confluent-kafka==2.14.2
minio==7.2.20
psycopg2-binary==2.9.10

What are these packages?

  • confluent-kafka is a high-performance Python client for Apache Kafka, built on top of the C library librdkafka. It handles producing and consuming messages.
  • minio is the Python SDK for MinIO. Since MinIO is S3-compatible, this SDK works identically with AWS S3.
  • psycopg2-binary is the most popular PostgreSQL adapter for Python. The binary variant ships pre-compiled so you don't need system-level build tools.
  • Save requirements.txt.
  • Confirm the file exists by running this in your terminal:
ls

You should see requirements.txt listed in the output.

Don't see requirements.txt?

Make sure you saved the file inside the kafka-pipeline folder, not in a parent directory. Run pwd to check your current location.

Help me find where my requirements.txt was saved

✔️ Awesome, I've got everything!

Great. Make sure you've saved your file before moving on.

ⓧ I'd like to double check the full code

confluent-kafka==2.14.2
minio==7.2.20
psycopg2-binary==2.9.10
Set up the Python virtual environment

A virtual environment isolates your project's packages from the rest of your system. This prevents version conflicts between different projects on your machine.

  • Create a virtual environment by running this command:
python3 -m venv venv
  • Activate the virtual environment by running:
source venv/bin/activate

You should see (venv) appear at the start of your terminal prompt. This confirms you're working inside the virtual environment.

  • Install all project dependencies by running:
pip install -r requirements.txt

You should see output ending with Successfully installed followed by the names of all three packages.

Seeing an installation error?

  • If you see a permissions error, make sure your virtual environment is activated. Check for (venv) in your prompt.
  • If psycopg2-binary fails to install, check that you're running Python 3.10 or higher with python3 --version.

Help me troubleshoot my pip install failure

Your workspace is ready with all the Python libraries needed to talk to Kafka, MinIO, and PostgreSQL. Next up, you'll spin up the actual infrastructure these libraries connect to.

Set Up the Infrastructure Stack

Your pipeline needs three pieces of infrastructure to function. A message broker handles the event stream. A data lake stores raw event files. A database holds transformed data ready for analytics queries.

You'll define all three services in a single Docker Compose file. This lets you start Apache Kafka, MinIO, and PostgreSQL with one command and tear them all down just as easily.

In this step, get ready to:
  • Create the Docker Compose configuration defining Kafka, MinIO, and PostgreSQL.
  • Start the infrastructure stack and verify all three services are healthy.
  • Create the MinIO bucket and verify the database table the pipeline writes to.
Create the Docker Compose configuration

Docker Compose reads a YAML file that declares your services, their images, ports, environment variables, and health checks. You'll build this file service by service.

  • Create a file called docker-compose.yml in your project directory and add the Kafka service by pasting the following code:
services:
  kafka:
    image: apache/kafka:4.3.0
    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@kafka: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
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 5

What is KRaft mode?

Kafka 4.x uses KRaft mode by default. This means Kafka manages its own metadata internally without needing a separate Zookeeper cluster. The KAFKA_PROCESS_ROLES: broker,controller setting tells this single node to handle both message brokering and cluster coordination.

The healthcheck runs kafka-topics.sh --list every 10 seconds. If Kafka can list its topics, the broker is ready to accept connections.

  • You won't see a change yet. The next section adds MinIO and PostgreSQL to the same file.
  • Add the MinIO and PostgreSQL services below the Kafka service, plus the shared volumes at the bottom. Paste this code directly after the Kafka healthcheck block:
  minio:
    image: minio/minio:RELEASE.2025-09-07T16-13-09Z
    container_name: minio
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin123
    command: server /data --console-address ":9001"
    volumes:
      - minio_data:/data
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 10s
      timeout: 5s
      retries: 3

  postgres:
    image: postgres:16.14-alpine
    container_name: postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_DB: pipeline_db
      POSTGRES_USER: pipeline_user
      POSTGRES_PASSWORD: pipeline_pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U pipeline_user -d pipeline_db"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  minio_data:
  postgres_data:

What do these services do?

MinIO is an S3-compatible object storage server. Port 9000 serves the S3 API and port 9001 serves the web console. The command field starts the server with the console on a fixed port.

PostgreSQL mounts ./init.sql into /docker-entrypoint-initdb.d/. Any SQL file in that directory runs automatically the first time the container starts. This is how the orders table gets created without manual intervention.

The volumes section at the bottom declares named volumes so data persists across container restarts.

  • Save docker-compose.yml.

Seeing a YAML syntax error?

YAML is sensitive to indentation. Each service (kafka, minio, postgres) must be indented exactly 2 spaces under services:. Properties under each service are indented 4 spaces total. Make sure you have no tabs in the file.

help me fix YAML indentation errors in my docker-compose.yml

✔️ Awesome, I've got everything!

Great. Make sure you've saved docker-compose.yml before continuing.

ⓧ I'd like to double check the full code

services:
  kafka:
    image: apache/kafka:4.3.0
    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@kafka: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
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 5

  minio:
    image: minio/minio:RELEASE.2025-09-07T16-13-09Z
    container_name: minio
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin123
    command: server /data --console-address ":9001"
    volumes:
      - minio_data:/data
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 10s
      timeout: 5s
      retries: 3

  postgres:
    image: postgres:16.14-alpine
    container_name: postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_DB: pipeline_db
      POSTGRES_USER: pipeline_user
      POSTGRES_PASSWORD: pipeline_pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U pipeline_user -d pipeline_db"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  minio_data:
  postgres_data:
  • Create a file called init.sql in the same directory as your docker-compose.yml and paste the following SQL:
CREATE TABLE IF NOT EXISTS orders (
    id SERIAL PRIMARY KEY,
    order_id VARCHAR(36) NOT NULL UNIQUE,
    customer_id VARCHAR(36) NOT NULL,
    product VARCHAR(255) NOT NULL,
    quantity INTEGER NOT NULL,
    price NUMERIC(10, 2) NOT NULL,
    total_amount NUMERIC(10, 2) NOT NULL,
    order_timestamp TIMESTAMP NOT NULL,
    ingested_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_orders_timestamp ON orders(order_timestamp);
CREATE INDEX idx_orders_product ON orders(product);

What does this SQL do?

This creates the orders table that the consumer will write to later. Each row stores one processed order event with a computed total_amount and an ingested_at timestamp recording when it arrived.

The two indexes speed up analytical queries that filter by time range or product name.

  • Save init.sql.
  • Confirm both files exist by running:
ls
  • You should see docker-compose.yml, init.sql, and requirements.txt in the output.

✔️ Awesome, I've got everything!

Great. Make sure you've saved init.sql before moving on.

ⓧ I'd like to double check the full code

CREATE TABLE IF NOT EXISTS orders (
    id SERIAL PRIMARY KEY,
    order_id VARCHAR(36) NOT NULL UNIQUE,
    customer_id VARCHAR(36) NOT NULL,
    product VARCHAR(255) NOT NULL,
    quantity INTEGER NOT NULL,
    price NUMERIC(10, 2) NOT NULL,
    total_amount NUMERIC(10, 2) NOT NULL,
    order_timestamp TIMESTAMP NOT NULL,
    ingested_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_orders_timestamp ON orders(order_timestamp);
CREATE INDEX idx_orders_product ON orders(product);
Start the infrastructure stack

With both files in place, you can bring up all three services in one command. Docker will pull the images the first time, which may take a minute or two.

  • Start the stack in detached mode by running:
docker compose up -d

What does the -d flag do?

The -d flag runs containers in detached mode (in the background). Without it, your terminal would be locked showing container logs. Detached mode lets you keep using the same terminal.

  • Wait about 30 seconds for the health checks to pass, then verify all services are healthy by running:
docker compose ps
  • You should see three containers listed (kafka, minio, postgres) all showing a running state with healthy in the status.

Containers not showing as healthy?

If a container shows starting instead of healthy, wait another 30 seconds and run docker compose ps again. Kafka can take up to 50 seconds to pass its first health check.

If a container shows exited, check its logs with docker compose logs kafka (replace kafka with the failing service name). A common cause is another process already using port 9092, 9000, or 5432.

help me debug why my Docker Compose containers aren't starting

Create the MinIO bucket and verify the database

The PostgreSQL table was created automatically by init.sql when the container first started. The MinIO bucket needs to be created manually through the web console.

  • Open your browser and navigate to http://localhost:9001.
  • Log in with username minioadmin and password minioadmin123.
  • In the left sidebar, click Buckets.
  • Click Create Bucket.
  • Enter raw-events as the bucket name.
  • Click Create Bucket to confirm.

You should see raw-events listed in the Buckets view.

Why MinIO instead of AWS S3?

MinIO implements the exact same S3 API that AWS uses. Code you write against MinIO transfers directly to AWS S3 with only an endpoint and credential change. Running it locally means zero cloud costs during development and no internet dependency.

  • Now verify the PostgreSQL table exists. Connect to the database by running:
docker exec -it postgres psql -U pipeline_user -d pipeline_db
  • Once you see the pipeline_db=# prompt, list the tables by running:
\dt
  • You should see the orders table listed in the output.
  • Exit the PostgreSQL prompt by running:
\q

Don't see the orders table?

The init.sql script only runs the first time the container starts. If you edited init.sql after the container was already created, you need to destroy the volume and recreate. Run docker compose down -v followed by docker compose up -d to start fresh.

help me fix my PostgreSQL init script not running in Docker

Your infrastructure is running and ready to receive data. Next up, you'll build the Python producer that generates order events and publishes them into Kafka.

Build the Event Producer

Your infrastructure stack is running and ready to receive data. Now you need something to actually send events into Kafka.

In a real-world pipeline, a producer is the entry point for data. It could be a web server logging user activity, a payment service recording transactions, or an IoT device streaming sensor readings. You will build a Python producer that simulates an e-commerce order stream.

In this step, get ready to:
  • Create a producer script with realistic e-commerce order data.
  • Add delivery confirmation callbacks so you know each message landed safely.
  • Run the producer and verify 100 events arrive in Kafka.
Create the producer configuration and product catalog

Every producer needs to know two things: where to send messages and what data to generate. You will define a connection to your local Kafka broker and a product catalog that the producer picks from randomly.

  • Create a new file called producer.py in your project directory (the same folder that contains docker-compose.yml).
  • Add the imports, Kafka connection settings, and product catalog by pasting this code:
import json
import uuid
import random
import time
from datetime import datetime, timezone
from confluent_kafka import Producer


# Kafka broker connection
KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
TOPIC = "ecommerce.orders"

# Product catalog that the producer picks from randomly
PRODUCTS = [
    {"name": "Wireless Headphones", "price": 79.99},
    {"name": "USB-C Cable", "price": 12.99},
    {"name": "Mechanical Keyboard", "price": 149.99},
    {"name": "Mouse Pad", "price": 24.99},
    {"name": "Webcam HD", "price": 59.99},
    {"name": "Monitor Stand", "price": 39.99},
    {"name": "Laptop Sleeve", "price": 29.99},
    {"name": "Phone Charger", "price": 19.99},
]

What does this code do?

The confluent_kafka library is a Python wrapper around librdkafka, a high-performance C library for Kafka. It is the production-standard Python client.

KAFKA_BOOTSTRAP_SERVERS points the producer at your local Kafka container. TOPIC defines the Kafka topic name where all order events will land. The PRODUCTS list acts as a simulated inventory that the producer samples from.

  • Save producer.py.

You will not see any output yet. The next chunk adds the order generation function.

Add the order generation function

Each order event needs a unique ID, a random product, a quantity, and a timestamp. This function produces a single realistic order dictionary that Kafka can serialize as JSON.

  • In producer.py, add the following function below the PRODUCTS list:
def generate_order():
    # Pick a random product and quantity
    product = random.choice(PRODUCTS)
    quantity = random.randint(1, 5)
    return {
        "order_id": str(uuid.uuid4()),
        "customer_id": str(uuid.uuid4()),
        "product": product["name"],
        "quantity": quantity,
        "price": product["price"],
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }

What does this code do?

generate_order() builds a dictionary representing one e-commerce order. uuid.uuid4() creates globally unique IDs for both the order and customer. random.choice() picks a product from the catalog, and random.randint(1, 5) gives a realistic quantity between 1 and 5.

The UTC timestamp ensures all events have a consistent time reference regardless of where the pipeline runs.

  • Save producer.py.

You won't see a change yet. The next chunk adds the delivery callback and main loop that actually publishes events.

Add delivery callbacks and the main producer loop

In production, you never assume a message was delivered. A delivery callback fires for every message and tells you whether it succeeded or failed. This is how you get delivery guarantees in Kafka.

  • In producer.py, add the following delivery_callback function above the generate_order function (between the PRODUCTS list and generate_order):
def delivery_callback(err, msg):
    # Called once per message to confirm delivery or report failure
    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?

Kafka calls delivery_callback asynchronously after each message is acknowledged by the broker. If err is not None, the message failed. Otherwise, you see exactly which topic, partition, and offset received the message.

This is the foundation of error handling in Kafka producers. In production, you would log failures or retry them.

  • Now add the main() function and entry point at the bottom of producer.py, after the generate_order function:
def main():
    # Create a Kafka producer instance
    producer = Producer({"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS})

    print(f"Producing 100 order events to topic '{TOPIC}'...")
    for i in range(100):
        order = generate_order()
        # Send the order as JSON with the order_id as the message key
        producer.produce(
            TOPIC,
            key=order["order_id"],
            value=json.dumps(order),
            callback=delivery_callback,
        )
        # Trigger delivery callbacks without blocking
        producer.poll(0)
        time.sleep(0.05)

    # Wait for all messages to be delivered
    remaining = producer.flush(timeout=10)
    print(f"\nDone. {100 - remaining} messages delivered, {remaining} still in queue.")


if __name__ == "__main__":
    main()

What does this code do?

producer.produce() queues each message for delivery. The key parameter uses the order ID so Kafka can partition messages deterministically. The callback parameter hooks in your delivery_callback function.

producer.poll(0) triggers any pending callbacks without blocking. time.sleep(0.05) adds a 50ms delay between messages to simulate a realistic order stream. Finally, producer.flush() blocks until all queued messages are delivered or the timeout expires.

  • Save producer.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved producer.py and move on to running it.

ⓧ I'd like to double check the full code

Here is the complete producer.py file. Make sure yours matches:

import json
import uuid
import random
import time
from datetime import datetime, timezone
from confluent_kafka import Producer


KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
TOPIC = "ecommerce.orders"

PRODUCTS = [
    {"name": "Wireless Headphones", "price": 79.99},
    {"name": "USB-C Cable", "price": 12.99},
    {"name": "Mechanical Keyboard", "price": 149.99},
    {"name": "Mouse Pad", "price": 24.99},
    {"name": "Webcam HD", "price": 59.99},
    {"name": "Monitor Stand", "price": 39.99},
    {"name": "Laptop Sleeve", "price": 29.99},
    {"name": "Phone Charger", "price": 19.99},
]


def delivery_callback(err, msg):
    if err is not None:
        print(f"DELIVERY FAILED: {err}")
    else:
        print(f"Delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}")


def generate_order():
    product = random.choice(PRODUCTS)
    quantity = random.randint(1, 5)
    return {
        "order_id": str(uuid.uuid4()),
        "customer_id": str(uuid.uuid4()),
        "product": product["name"],
        "quantity": quantity,
        "price": product["price"],
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }


def main():
    producer = Producer({"bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS})

    print(f"Producing 100 order events to topic '{TOPIC}'...")
    for i in range(100):
        order = generate_order()
        producer.produce(
            TOPIC,
            key=order["order_id"],
            value=json.dumps(order),
            callback=delivery_callback,
        )
        producer.poll(0)
        time.sleep(0.05)

    remaining = producer.flush(timeout=10)
    print(f"\nDone. {100 - remaining} messages delivered, {remaining} still in queue.")


if __name__ == "__main__":
    main()
Run the producer and verify delivery

Time to send 100 order events through your pipeline. The producer will connect to your local Kafka container and publish events to the ecommerce.orders topic.

  • Run the producer by executing this command in your terminal:
python producer.py

You should see 100 delivery confirmation lines printed, each showing the topic, partition, and offset. The final line reads something like:

Done. 100 messages delivered, 0 still in queue.

Seeing a connection error?

  • Make sure your Docker containers are still running. Run docker compose ps and confirm the kafka container shows as healthy.
  • Check that your virtual environment is activated. If you see a ModuleNotFoundError, the confluent-kafka package is not installed in your current Python environment.
  • Verify Kafka is listening on port 9092 by checking docker compose logs kafka for startup errors.

Help me troubleshoot my Kafka producer connection.

  • Verify the messages are in Kafka by reading a few events from the topic:
docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ecommerce.orders --from-beginning --max-messages 3

You should see three JSON objects printed, each containing fields like order_id, customer_id, product, quantity, price, and timestamp.

Why use the console consumer to verify?

The Kafka console consumer is a built-in CLI tool that reads messages from a topic. Using --from-beginning reads all messages from offset 0. The --max-messages 3 flag limits output so your terminal is not flooded with all 100 events.

This gives you independent confirmation that messages are stored in Kafka, separate from what the producer reported.

Console consumer hangs or shows no output?

  • Wait a few seconds. The consumer needs time to assign partitions and fetch messages.
  • If it hangs indefinitely, press Ctrl+C to exit, then re-run the command. Sometimes the initial partition assignment takes longer on a fresh topic.

Help me debug why the Kafka console consumer shows no messages.

Your event producer is working. You have 100 order events sitting in Kafka, ready to be consumed. Next up, you will build the stream consumer that reads these events, stores raw data in MinIO, and loads transformed records into PostgreSQL.

Build the Stream Consumer

In Step 2, you built a producer that pumps order events into Kafka. Those 100 messages are now sitting in the ecommerce.orders topic, waiting to be processed.

In this step, you'll build the consumer side of the pipeline. The consumer reads events from Kafka, stores raw JSON files in MinIO (your data lake), and loads transformed data into PostgreSQL (your data warehouse). This is the core ETL pattern: extract from the message broker, transform the data, and load it into storage systems.

In this step, get ready to:
  • Subscribe to the Kafka topic and read events in batches.
  • Store raw event data as partitioned JSON files in MinIO.
  • Transform events and load them into PostgreSQL.
Create the consumer with MinIO storage

The consumer needs to connect to three systems: Kafka (to read events), MinIO (to store raw data), and PostgreSQL (to load transformed data). You'll start by setting up the file with all the imports, configuration, and the functions that handle MinIO storage.

  • Create a new file called consumer.py in your project directory.
  • Add the imports and configuration constants by pasting this code:
import json
import io
from datetime import datetime, timezone
from confluent_kafka import Consumer, KafkaError
from minio import Minio
from minio.error import S3Error
import psycopg2


# Kafka connection settings
KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
TOPIC = "ecommerce.orders"
GROUP_ID = "order-pipeline-consumer"

# MinIO connection settings (S3-compatible object storage)
MINIO_ENDPOINT = "localhost:9000"
MINIO_ACCESS_KEY = "minioadmin"
MINIO_SECRET_KEY = "minioadmin123"
MINIO_BUCKET = "raw-events"

# PostgreSQL connection settings
PG_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "dbname": "pipeline_db",
    "user": "pipeline_user",
    "password": "pipeline_pass",
}

# Number of events to collect before writing a batch
BATCH_SIZE = 10

What does this code do?

The imports pull in libraries for each system the consumer talks to: confluent_kafka for reading from Kafka, minio for writing to MinIO, and psycopg2 for inserting into PostgreSQL.

GROUP_ID identifies this consumer within Kafka's consumer group system. Kafka tracks which messages this group has already processed, so if the consumer restarts, it picks up where it left off.

BATCH_SIZE controls how many events the consumer collects before writing them to storage. Batching reduces the number of write operations to MinIO and PostgreSQL.

  • Next, add the MinIO helper functions below the configuration. These handle connecting to MinIO and ensuring the bucket exists:

def get_minio_client():
    return Minio(
        MINIO_ENDPOINT,
        access_key=MINIO_ACCESS_KEY,
        secret_key=MINIO_SECRET_KEY,
        secure=False,
    )


def ensure_bucket(client, bucket_name):
    if not client.bucket_exists(bucket_name):
        client.make_bucket(bucket_name)
        print(f"Created bucket: {bucket_name}")

What does this code do?

get_minio_client() creates a connection to your local MinIO server. The secure=False flag tells the client to use HTTP instead of HTTPS, which is appropriate for local development.

ensure_bucket() is a safety check. It creates the bucket if it does not already exist. Since you already created raw-events in Step 1, this function will skip creation. It protects against the case where someone runs the consumer before setting up the bucket.

  • Now add the store_raw_events() function that writes batches of events to MinIO as partitioned JSON files:

def store_raw_events(minio_client, events):
    # Build a date-partitioned path (data lake convention)
    now = datetime.now(timezone.utc)
    path = f"year={now.year}/month={now.month:02d}/day={now.day:02d}"
    filename = f"batch_{now.strftime('%H%M%S_%f')}.json"
    object_name = f"{path}/{filename}"

    # Serialize events to JSON and upload to MinIO
    data = json.dumps(events, indent=2).encode("utf-8")
    minio_client.put_object(
        MINIO_BUCKET,
        object_name,
        io.BytesIO(data),
        length=len(data),
        content_type="application/json",
    )
    print(f"Stored {len(events)} raw events to MinIO: {object_name}")

Why partition by date?

The path format year=YYYY/month=MM/day=DD is a data lake convention called Hive-style partitioning. It organizes files by time so query engines can skip irrelevant date ranges without scanning every file.

Each batch gets a unique filename using the current timestamp down to microseconds. This prevents filename collisions when multiple batches arrive within the same second.

  • Save consumer.py.

You won't see a visible change yet. The next chunk adds PostgreSQL loading and the main event loop that ties everything together.

Add the transformation and PostgreSQL loading

Raw data in MinIO is great for archival, but you also need structured data in PostgreSQL for fast queries. The next function transforms each event by computing total_amount and inserts it into the orders table.

  • Add the load_to_postgres() function below store_raw_events() in consumer.py:

def load_to_postgres(conn, events):
    cursor = conn.cursor()
    insert_sql = """
        INSERT INTO orders (order_id, customer_id, product, quantity, price, total_amount, order_timestamp)
        VALUES (%s, %s, %s, %s, %s, %s, %s)
        ON CONFLICT (order_id) DO NOTHING
    """
    # Transform each event: compute total and insert
    for event in events:
        total_amount = event["quantity"] * event["price"]
        cursor.execute(
            insert_sql,
            (
                event["order_id"],
                event["customer_id"],
                event["product"],
                event["quantity"],
                event["price"],
                total_amount,
                event["timestamp"],
            ),
        )
    conn.commit()
    cursor.close()
    print(f"Loaded {len(events)} events to PostgreSQL")

What does this code do?

This is the "transform" and "load" in ETL. The transformation computes total_amount by multiplying quantity * price for each event. The load step inserts the result into PostgreSQL using parameterized queries (the %s placeholders) to prevent SQL injection.

ON CONFLICT (order_id) DO NOTHING makes the insert idempotent. If the consumer processes the same event twice (which can happen after a restart), it skips the duplicate instead of crashing.

  • Now add the main() function and the entry point at the bottom of consumer.py. This is the event loop that ties everything together:

def main():
    # Configure the Kafka consumer with manual offset commits
    consumer = Consumer(
        {
            "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS,
            "group.id": GROUP_ID,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,
        }
    )
    consumer.subscribe([TOPIC])

    # Connect to MinIO and PostgreSQL
    minio_client = get_minio_client()
    ensure_bucket(minio_client, MINIO_BUCKET)

    conn = psycopg2.connect(**PG_CONFIG)

    print(f"Consuming from topic '{TOPIC}'... (Ctrl+C to stop)")
    batch = []

    try:
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                # No message within timeout - flush any partial batch
                if batch:
                    store_raw_events(minio_client, batch)
                    load_to_postgres(conn, batch)
                    consumer.commit()
                    batch = []
                continue

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

            # Decode the message and add to batch
            event = json.loads(msg.value().decode("utf-8"))
            batch.append(event)

            # When batch is full, process it
            if len(batch) >= BATCH_SIZE:
                store_raw_events(minio_client, batch)
                load_to_postgres(conn, batch)
                consumer.commit()
                batch = []

    except KeyboardInterrupt:
        print("\nShutting down...")
        # Flush any remaining events before exiting
        if batch:
            store_raw_events(minio_client, batch)
            load_to_postgres(conn, batch)
            consumer.commit()
    finally:
        consumer.close()
        conn.close()


if __name__ == "__main__":
    main()

What does the main loop do?

consumer.poll(timeout=1.0) waits up to 1 second for a message. If no message arrives, it flushes any partial batch. This ensures events do not sit in memory indefinitely.

enable.auto.commit: False means the consumer manually calls consumer.commit() only after successfully writing data to both MinIO and PostgreSQL. This guarantees at-least-once delivery. If the consumer crashes mid-batch, Kafka re-delivers those messages on restart.

The KeyboardInterrupt handler ensures a clean shutdown. It flushes any remaining events and closes all connections before exiting.

  • Save consumer.py.

Seeing a syntax error when you save?

Check that the triple-quoted SQL string in load_to_postgres() uses three double quotes (""") on both sides. Also verify the f-strings are using the correct quote style and have matching brackets.

help me find the syntax error in my consumer.py

✔️ Awesome, I've got everything!

Great. Double check you have saved consumer.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete consumer.py file for comparison:

import json
import io
from datetime import datetime, timezone
from confluent_kafka import Consumer, KafkaError
from minio import Minio
from minio.error import S3Error
import psycopg2


KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
TOPIC = "ecommerce.orders"
GROUP_ID = "order-pipeline-consumer"

MINIO_ENDPOINT = "localhost:9000"
MINIO_ACCESS_KEY = "minioadmin"
MINIO_SECRET_KEY = "minioadmin123"
MINIO_BUCKET = "raw-events"

PG_CONFIG = {
    "host": "localhost",
    "port": 5432,
    "dbname": "pipeline_db",
    "user": "pipeline_user",
    "password": "pipeline_pass",
}

BATCH_SIZE = 10


def get_minio_client():
    return Minio(
        MINIO_ENDPOINT,
        access_key=MINIO_ACCESS_KEY,
        secret_key=MINIO_SECRET_KEY,
        secure=False,
    )


def ensure_bucket(client, bucket_name):
    if not client.bucket_exists(bucket_name):
        client.make_bucket(bucket_name)
        print(f"Created bucket: {bucket_name}")


def store_raw_events(minio_client, events):
    now = datetime.now(timezone.utc)
    path = f"year={now.year}/month={now.month:02d}/day={now.day:02d}"
    filename = f"batch_{now.strftime('%H%M%S_%f')}.json"
    object_name = f"{path}/{filename}"

    data = json.dumps(events, indent=2).encode("utf-8")
    minio_client.put_object(
        MINIO_BUCKET,
        object_name,
        io.BytesIO(data),
        length=len(data),
        content_type="application/json",
    )
    print(f"Stored {len(events)} raw events to MinIO: {object_name}")


def load_to_postgres(conn, events):
    cursor = conn.cursor()
    insert_sql = """
        INSERT INTO orders (order_id, customer_id, product, quantity, price, total_amount, order_timestamp)
        VALUES (%s, %s, %s, %s, %s, %s, %s)
        ON CONFLICT (order_id) DO NOTHING
    """
    for event in events:
        total_amount = event["quantity"] * event["price"]
        cursor.execute(
            insert_sql,
            (
                event["order_id"],
                event["customer_id"],
                event["product"],
                event["quantity"],
                event["price"],
                total_amount,
                event["timestamp"],
            ),
        )
    conn.commit()
    cursor.close()
    print(f"Loaded {len(events)} events to PostgreSQL")


def main():
    consumer = Consumer(
        {
            "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS,
            "group.id": GROUP_ID,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,
        }
    )
    consumer.subscribe([TOPIC])

    minio_client = get_minio_client()
    ensure_bucket(minio_client, MINIO_BUCKET)

    conn = psycopg2.connect(**PG_CONFIG)

    print(f"Consuming from topic '{TOPIC}'... (Ctrl+C to stop)")
    batch = []

    try:
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                if batch:
                    store_raw_events(minio_client, batch)
                    load_to_postgres(conn, batch)
                    consumer.commit()
                    batch = []
                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:
                store_raw_events(minio_client, batch)
                load_to_postgres(conn, batch)
                consumer.commit()
                batch = []

    except KeyboardInterrupt:
        print("\nShutting down...")
        if batch:
            store_raw_events(minio_client, batch)
            load_to_postgres(conn, batch)
            consumer.commit()
    finally:
        consumer.close()
        conn.close()


if __name__ == "__main__":
    main()
Run the full pipeline end-to-end

Your consumer is complete. Time to run it and watch events flow through the entire pipeline.

  • Start the consumer by running this command:
python consumer.py

You should see output like:

Consuming from topic 'ecommerce.orders'... (Ctrl+C to stop)
Stored 10 raw events to MinIO: year=2026/month=06/day=15/batch_143022_123456.json
Loaded 10 events to PostgreSQL
Stored 10 raw events to MinIO: year=2026/month=06/day=15/batch_143022_234567.json
Loaded 10 events to PostgreSQL
...

What is happening?

The consumer is reading the 100 events you produced in Step 2. Since auto.offset.reset is set to earliest, it starts from the beginning of the topic. It processes them in batches of 10, writing each batch to both MinIO and PostgreSQL.

Once all 100 events are processed, the consumer keeps running and waits for new events. You will see it idle after the last batch.

  • Once the consumer finishes processing the existing 100 events and idles, stop it by pressing Ctrl+C.
  • Now run the producer again to send fresh events through the full pipeline. Open a second terminal, activate your virtual environment, and run:
python producer.py
  • Then start the consumer again to process the new batch:
python consumer.py

You should see the consumer process another 100 events in batches of 10, then idle again waiting for more.

  • Stop the consumer with Ctrl+C once it finishes processing.

Why did it start from offset 100?

Kafka tracks which messages the consumer group has already processed. On the second run, it starts from offset 100 (where the first run left off) and only processes the new messages. This is Kafka's consumer group offset tracking in action.

  • Verify data landed in MinIO by opening the MinIO Console at http://localhost:9001 in your browser.
  • Log in with username minioadmin and password minioadmin123.
  • Click on the raw-events bucket. You should see a folder structure like year=2026/month=06/day=15/ containing JSON batch files.
  • Verify data landed in PostgreSQL by running this command:
docker exec -it postgres psql -U pipeline_user -d pipeline_db -c "SELECT COUNT(*) FROM orders;"

You should see a count of 200 (100 from the first producer run plus 100 from the second).

Consumer exits immediately with no output?

Make sure your Docker containers are still running. Run docker compose ps and check that all three services show as healthy. If Kafka restarted, the topic may be empty and the consumer will just idle.

If you see a connection error mentioning port 9092, Kafka may not be ready yet. Wait 10 seconds and try again.

help me troubleshoot why my Kafka consumer is not processing events

Your pipeline is processing events end-to-end. Next up, you'll query PostgreSQL to prove the data is correct and test the pipeline's resilience by restarting the consumer mid-stream.

Query and Verify the Pipeline

Your pipeline is processing events end-to-end. Data flows from the producer through Kafka, lands in MinIO, and arrives in PostgreSQL.

Now you need to prove it actually works. In production, data engineers run analytical queries and cross-check event counts across every system in the pipeline. If the numbers don't match, something dropped data along the way.

In this step, get ready to:
  • Run analytical SQL queries against PostgreSQL to calculate revenue, top products, and order frequency.
  • Verify data consistency across Kafka, MinIO, and PostgreSQL.
  • Test pipeline resilience by stopping and restarting the consumer.
Run analytical queries against PostgreSQL

Analytical queries prove that your pipeline delivers usable business data. You'll create a SQL file with queries that answer real questions: how much revenue came in, which products sell best, and how orders distribute over time.

  • Create a file called queries.sql in your project directory by adding the following SQL:
-- Total revenue
SELECT SUM(total_amount) AS total_revenue FROM orders;

-- Top 5 products by quantity sold
SELECT product, SUM(quantity) AS total_quantity, SUM(total_amount) AS revenue
FROM orders
GROUP BY product
ORDER BY total_quantity DESC
LIMIT 5;

-- Orders per hour
SELECT DATE_TRUNC('hour', order_timestamp) AS hour, COUNT(*) AS order_count
FROM orders
GROUP BY hour
ORDER BY hour;

-- Total event count (should match Kafka offset and MinIO file counts)
SELECT COUNT(*) AS total_orders FROM orders;

What do these queries do?

The first query sums the total_amount column across all orders to give you total revenue.

The second groups orders by product name, ranks them by total quantity sold, and limits to the top 5. The third uses DATE_TRUNC to bucket orders by hour so you can see order volume over time. The last query counts all rows, which you'll use in the next substep to verify data consistency.

  • Save queries.sql.
  • Run the queries against PostgreSQL by executing this command:
docker exec -i postgres psql -U pipeline_user -d pipeline_db < queries.sql

You should see four result sets printed to your terminal: a single revenue number, a table of products with quantities, hourly order counts, and a total row count matching the 100 events you produced earlier.

Seeing zero rows or an error?

Make sure you ran both the producer and consumer in Step 3 before running these queries. The orders table only has data after the consumer processes events.

If you see a connection error, confirm the postgres container is running with docker ps.

help me debug why my PostgreSQL queries return no data

Verify data consistency across all three systems

A reliable pipeline delivers the same event count at every stage. If Kafka shows 100 messages produced but PostgreSQL only has 95 rows, you have data loss somewhere.

  • Check the Kafka consumer group offsets to see how many messages were consumed by running:
docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group order-pipeline-consumer

What does this output mean?

The output shows columns including CURRENT-OFFSET, LOG-END-OFFSET, and LAG for each partition of the ecommerce.orders topic. CURRENT-OFFSET is how far your consumer has read. LOG-END-OFFSET is the total messages in that partition. LAG is the difference.

If LAG is 0 across all partitions, your consumer has processed every message. Sum the CURRENT-OFFSET values across partitions to get the total consumed count.

  • Verify the PostgreSQL row count matches by running:
docker exec postgres psql -U pipeline_user -d pipeline_db -c "SELECT COUNT(*) AS total_orders FROM orders;"
  • Check the MinIO bucket by opening http://localhost:9001 in your browser.
  • Sign in with username minioadmin and password minioadmin123.
  • Click the raw-events bucket, then navigate into the date-partitioned folders to see the JSON batch files.

The sum of events across all JSON batch files in MinIO should equal the PostgreSQL row count, which should equal the total Kafka offset. All three numbers should be 100.

Counts don't match?

If PostgreSQL has fewer rows than Kafka offsets, the consumer may have been interrupted before finishing. Run python consumer.py again to process any remaining events, then re-check the count.

help me figure out why my event counts don't match across Kafka, MinIO, and PostgreSQL

Test pipeline resilience with consumer restart

Production pipelines must survive restarts without losing or duplicating data. Kafka's consumer group offset tracking remembers where each consumer left off. When the consumer restarts, it picks up from the last committed offset instead of re-reading from the beginning.

What is consumer group offset tracking?

Every time your consumer calls consumer.commit(), it tells Kafka which messages it has successfully processed. Kafka stores this position (the offset) for the consumer group order-pipeline-consumer. On restart, the consumer asks Kafka for its last committed offset and resumes from there.

  • Make sure your consumer from Step 3 is stopped. If it's still running, press Ctrl+C to stop it.
  • Produce 100 more events by running the producer again:
python producer.py
  • Check the consumer group lag to confirm events are waiting:
docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group order-pipeline-consumer

You should see LAG values greater than 0 in the output. This means Kafka has messages the consumer hasn't processed yet.

  • Restart the consumer to process the waiting events:
python consumer.py
  • Wait for the consumer to process all events, then press Ctrl+C to stop it.
  • Verify the new total count in PostgreSQL:
docker exec postgres psql -U pipeline_user -d pipeline_db -c "SELECT COUNT(*) AS total_orders FROM orders;"

You should see 200 total rows. The consumer picked up exactly where it left off and processed only the new 100 events without re-processing the original 100.

Seeing duplicate rows or fewer than 200?

If you see exactly 100, the consumer may have re-read from the beginning due to an expired offset. Make sure your consumer uses enable.auto.commit: False and calls consumer.commit() after each batch, as shown in consumer.py.

Duplicates won't happen here because the orders table has a UNIQUE constraint on order_id and the INSERT uses ON CONFLICT DO NOTHING.

help me debug why my consumer isn't resuming from the correct Kafka offset

✔️ Awesome, I've got everything!

Great. Double check you've saved queries.sql and confirmed all your counts match.

ⓧ I'd like to double check the full code

Here is the complete queries.sql file:

-- Total revenue
SELECT SUM(total_amount) AS total_revenue FROM orders;

-- Top 5 products by quantity sold
SELECT product, SUM(quantity) AS total_quantity, SUM(total_amount) AS revenue
FROM orders
GROUP BY product
ORDER BY total_quantity DESC
LIMIT 5;

-- Orders per hour
SELECT DATE_TRUNC('hour', order_timestamp) AS hour, COUNT(*) AS order_count
FROM orders
GROUP BY hour
ORDER BY hour;

-- Total event count (should match Kafka offset and MinIO file counts)
SELECT COUNT(*) AS total_orders FROM orders;

Secret mission

Add a Dead-Letter Queue and Retry Logic

In production, events fail. Maybe the JSON is malformed, or maybe PostgreSQL is temporarily unavailable. Production pipelines never silently drop data. Can you build a dead-letter queue that catches failures and routes them for investigation?

Clean Up Your Resources

Clean Up Your Resources

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

This project runs entirely locally using Docker containers, so there are no ongoing cloud costs. However, the running containers do consume memory and CPU on your machine.

Resources you used:

  • Docker containers: kafka, minio, postgres.
  • Docker volumes: minio_data and postgres_data.
  • Project files on your Desktop (docker-compose.yml, init.sql, producer.py, consumer.py, consumer_with_dlq.py, dlq_monitor.py, queries.sql, requirements.txt).
  • Python virtual environment inside your project folder.

Keep everything running

No action needed. Choose this if you're still actively building or plan to continue experimenting with the pipeline right away.

  • Your Docker containers will keep running and consuming memory until you manually stop them or restart your machine.
  • You can re-run producer.py and consumer.py at any time to push more events through the pipeline.

Pause - I'll come back to this later

Stop the running containers to free up memory and CPU. Your data stays intact in the Docker volumes, so you can pick up where you left off.

  • Navigate to your project folder (the one on your Desktop containing docker-compose.yml) in your terminal.
  • Stop the containers by running:
docker compose stop
  • When you want to resume, run docker compose start from the same folder to bring everything back up.

Delete - I don't want to use this again

Remove all project containers, volumes, and files to start completely fresh.

  • Navigate to your project folder (the one on your Desktop containing docker-compose.yml) in your terminal.
  • Stop and remove all containers, networks, and volumes by running:
docker compose down -v
  • This removes the kafka, minio, and postgres containers along with the minio_data and postgres_data volumes. All stored events and database rows will be permanently deleted.
  • Move up one level and delete the project folder by running:
cd ~/Desktop
rm -rf kafka-pipeline

Folder name

Replace kafka-pipeline with whatever you named your project folder when you created it in Step 1.

  • Verify the folder is gone by running:
ls ~/Desktop
  • Your project folder should no longer appear in the output.

Nice Work!

Nice Work!

You just built a complete real-time data pipeline using Apache Kafka, MinIO, and PostgreSQL. This is the same architecture pattern that powers streaming platforms at companies like Uber, LinkedIn, and Spotify.

You've learned how to:

  • Operated containerized infrastructure with Docker Compose, running Kafka, MinIO, and PostgreSQL as a local stack that mirrors production cloud environments.
  • Built a Kafka producer with delivery guarantees and a stream consumer that implements the data lake + data warehouse pattern, storing raw events in MinIO and transformed data in PostgreSQL.
  • Verified end-to-end data consistency across three systems and tested pipeline resilience with consumer restart and offset tracking.
  • Secret Mission: Implemented production-grade error handling with dead-letter queues and retry logic.

Ready to quiz yourself?