Real-Time Kafka Streaming Pipeline

Build a multi-source streaming pipeline with Kafka, Python, and PostgreSQL.

Introduction

30 Second Summary

Every online store pulls data from multiple systems at once. Customer profiles live in one database, product catalogs in another, and order records in a third.

In this project, you will build a real-time streaming pipeline that ingests data from a CSV file and a REST API, routes it through Apache Kafka, and loads it into a PostgreSQL database ready for analytics.

What You'll Build

A containerized streaming data pipeline that pulls from multiple data sources through Kafka topics and loads clean, queryable data into PostgreSQL.

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

  • A Docker Compose stack running Kafka, PostgreSQL, and a Kafka UI dashboard all connected on a shared network.
  • Two Python producers that ingest data from different sources (a CSV file and a REST API) into separate Kafka topics.
  • A Python consumer that subscribes to both topics, transforms the data, and inserts it into PostgreSQL tables.
  • Secret Mission: A dead letter queue that catches malformed records and validates data quality before loading.

Are there any prerequisites?

You need Docker Desktop installed with at least 4 GB of memory allocated, Python 3.8 or higher, and basic familiarity with Python and SQL.

Set Up the Infrastructure with Docker Compose

Before you can stream any data, you need infrastructure to carry it. A streaming pipeline requires a message broker to queue events, a database to store the processed results, and a way to observe what is flowing through the system.

In this step, you will spin up three services using Docker Compose: Apache Kafka as your message broker, PostgreSQL as your destination database, and Kafka UI as a visual dashboard to inspect topics and messages.

In this step, get ready to:
  • Create the project folder and write a Docker Compose file with Kafka, PostgreSQL, and Kafka UI.
  • Start the stack and verify all services are healthy.
  • Install the Python packages you need for producing and consuming messages.
Create the project directory and Docker Compose file

Docker Compose lets you define multiple containers in a single YAML file and start them all with one command. Your pipeline needs three services talking to each other over a shared network.

  • Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or Command Prompt (Windows) 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 kafka-pipeline
cd kafka-pipeline

Your terminal prompt should now show kafka-pipeline in the path, confirming the folder was created successfully.

Now you need to create the docker-compose.yml file that defines your three services. Open your preferred code editor and create a new file called docker-compose.yml inside your kafka-pipeline folder.

  • Paste the following configuration into docker-compose.yml:
services:
  kafka:
    image: apache/kafka:latest
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:19092,CONTROLLER://0.0.0.0:9093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:19092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      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
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 5

  postgres:
    image: postgres:16
    container_name: postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: pipeline
      POSTGRES_PASSWORD: pipeline123
      POSTGRES_DB: ecommerce
    volumes:
      - pgdata:/var/lib/postgresql/data

  kafka-ui:
    image: kafbat/kafka-ui:latest
    container_name: kafka-ui
    ports:
      - "8080:8080"
    environment:
      DYNAMIC_CONFIG_ENABLED: "true"
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:19092
    depends_on:
      kafka:
        condition: service_healthy

volumes:
  pgdata:

What does this configuration do?

This file defines three services that Docker will run as containers on a shared network:

  • kafka runs in KRaft combined mode (broker + controller in one container). The PLAINTEXT listener on port 19092 is for inter-container communication, while PLAINTEXT_HOST on port 9092 lets your Python scripts connect from the host machine.
  • postgres stores the processed data. It creates a database called ecommerce with the user pipeline on startup.
  • kafka-ui gives you a visual dashboard to inspect topics, messages, and consumer groups. It waits for Kafka to be healthy before starting, thanks to the depends_on condition.

Why KRaft mode instead of Zookeeper?

Older Kafka setups required a separate Zookeeper container to manage cluster metadata. KRaft (Kafka Raft) mode removes this dependency by letting Kafka handle its own consensus internally.

For a single-node development setup, combined mode (broker + controller in one container) is the simplest and recommended approach.

  • Save docker-compose.yml.

Having trouble creating the file?

Make sure you saved the file as docker-compose.yml (not docker-compose.yml.txt). If you are using Notepad on Windows, change the "Save as type" dropdown to "All Files" before saving.

Help me troubleshoot creating docker-compose.yml

✔️ Awesome, I've got everything!

Great. Double check you have saved docker-compose.yml inside your kafka-pipeline folder.

ⓧ I'd like to double check the full code

Here is the complete docker-compose.yml file for comparison:

services:
  kafka:
    image: apache/kafka:latest
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:19092,CONTROLLER://0.0.0.0:9093,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:19092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      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
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 5

  postgres:
    image: postgres:16
    container_name: postgres
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: pipeline
      POSTGRES_PASSWORD: pipeline123
      POSTGRES_DB: ecommerce
    volumes:
      - pgdata:/var/lib/postgresql/data

  kafka-ui:
    image: kafbat/kafka-ui:latest
    container_name: kafka-ui
    ports:
      - "8080:8080"
    environment:
      DYNAMIC_CONFIG_ENABLED: "true"
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:19092
    depends_on:
      kafka:
        condition: service_healthy

volumes:
  pgdata:
Start the Docker Compose stack

With your configuration in place, you can now start all three services with a single command. Docker will pull the images, create containers, and wire them together on a shared network.

  • Make sure Docker Desktop is running. You should see it in your taskbar or menu bar.
  • Start the entire stack in detached mode by running this command from your kafka-pipeline folder:
docker compose up -d

What does this command do?

The -d flag runs the containers in the background (detached mode) so they do not block your terminal. Docker Compose pulls any images it does not have locally, then starts all three services in dependency order.

Kafka starts first. PostgreSQL starts independently. Kafka UI waits for Kafka's healthcheck to pass before starting.

The first run takes a minute or two as Docker downloads the container images. Subsequent starts are much faster.

  • Verify all three containers are running by checking their status:
docker compose ps

You should see three containers listed with STATUS showing "Up" for each. The kafka container may also show "(healthy)" once its healthcheck passes.

  • Open your web browser and navigate to http://localhost:8080 to confirm Kafka UI is running.

You should see the Kafka UI dashboard with your cluster named local listed in the navigation.

Containers not starting?

  • If you see a port conflict error, another service may be using port 9092, 5432, or 8080. Stop the conflicting service or change the host port in your docker-compose.yml.
  • If Docker reports insufficient memory, open Docker Desktop settings and ensure at least 4 GB of memory is allocated.
  • If the kafka-ui container keeps restarting, Kafka's healthcheck may still be failing. Wait 30 seconds and run docker compose ps again.

Help me troubleshoot my Docker containers

Install the Python dependencies

Your Python producers and consumer will use three libraries to interact with Kafka, PostgreSQL, and an external REST API. You need to install them on your host machine since the scripts run outside Docker.

  • Install the required Python packages by running:
pip install confluent-kafka==2.14.2 psycopg2-binary==2.9.10 requests==2.32.3

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.
  • psycopg2-binary is the PostgreSQL adapter for Python. The -binary variant bundles the C library so you do not need to compile anything.
  • requests is a simple HTTP library for fetching data from REST APIs.
  • Verify the installation succeeded by running:
pip show confluent-kafka

You should see package metadata including the version 2.14.2 in the output.

Install command failing?

  • If you see a permissions error, try running the install with pip install --user instead.
  • If pip is not recognized, try pip3 instead. Some systems use pip3 for Python 3 packages.
  • On Windows, if psycopg2-binary fails, ensure you have Python 3.8 or higher installed.

Help me debug the Python package installation

Your infrastructure is up and running. Next up, you will create Kafka topics and set up the PostgreSQL tables that will receive your streaming data.

Create Kafka Topics and the PostgreSQL Schema

Your infrastructure is running and healthy. Now you need to prepare the data channels and storage that your producers and consumer will use.

In a streaming pipeline, data flows through named channels called topics. Each topic holds messages from a specific source. You also need database tables ready to receive the processed data. This step sets up both sides of the pipeline.

In this step, get ready to:
  • Create two Kafka topics for customer and product data.
  • Build the PostgreSQL schema with tables matching your expected data shapes.
  • Verify both topics and tables exist using Kafka UI and psql.
Create the Kafka topics

Kafka organizes messages into topics. Your pipeline needs two topics: one for customer data from CSV files, and one for product data from the API.

  • Create the customers topic by running this command:
docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic customers --partitions 1 --replication-factor 1

What does this command do?

This command runs the Kafka topic management script inside the running kafka container.

  • docker exec kafka runs a command inside the container named kafka.
  • --partitions 1 creates a single partition (enough for a development environment).
  • --replication-factor 1 means one copy of the data (since you only have one broker).
  • You should see a confirmation message like Created topic customers. in your terminal.
  • Create the products topic by running this command:
docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic products --partitions 1 --replication-factor 1
  • You should see Created topic products. in the output.

Seeing a connection error?

  • Make sure the Kafka container is running and healthy. Run docker ps and check that the kafka container shows a status of healthy.
  • If the container is still starting up, wait 10-15 seconds and try again. The healthcheck needs time to pass.
  • help me troubleshoot why my docker exec kafka command is failing
Create the PostgreSQL tables

Your consumer will write processed messages into PostgreSQL tables. You need two tables: one for customer records and one for product records, each with columns that match the data your producers will send.

  • Connect to the PostgreSQL database inside the container by running this command:
docker exec -it postgres psql -U pipeline -d ecommerce
  • You should see a psql prompt that looks like ecommerce=# indicating you are connected to the ecommerce database.
  • Create the raw_customers table by running this SQL:
CREATE TABLE IF NOT EXISTS raw_customers (
    id VARCHAR(10) PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100),
    city VARCHAR(50),
    signup_date VARCHAR(20),
    ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

What does this table store?

  • id is the customer identifier from the CSV file, stored as text.
  • name, email, and city hold the customer profile fields.
  • signup_date stores the date as text (matching the CSV format).
  • ingested_at automatically records when each record was inserted.
  • You should see CREATE TABLE printed in the terminal confirming the table was created.
  • Create the raw_products table by running this SQL:
CREATE TABLE IF NOT EXISTS raw_products (
    id INTEGER PRIMARY KEY,
    title VARCHAR(200),
    price NUMERIC(10,2),
    category VARCHAR(100),
    stock INTEGER,
    ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

What does this table store?

  • id is an integer matching the product ID from the DummyJSON API.
  • price uses NUMERIC(10,2) to store exact decimal values (important for financial data).
  • stock is an integer representing the current inventory level.
  • You should see CREATE TABLE printed again confirming the second table was created.

Seeing a syntax error?

  • Make sure you included the semicolon at the end of the SQL statement. PostgreSQL waits for ; before executing.
  • If you see ecommerce-# (with a hyphen) instead of ecommerce=# (with equals), the statement is incomplete. Type ; and press Enter to finish it.
  • help me fix a SQL syntax error when creating my PostgreSQL tables
Verify the topics and tables

Before writing any producer or consumer code, confirm that both your Kafka topics and PostgreSQL tables are ready to receive data.

  • While still in the psql prompt, list the tables by running:
\dt
  • You should see both raw_customers and raw_products listed in the output.
  • Exit the psql prompt by running:
\q
  • Switch to your browser and return to Kafka UI at http://localhost:8080.
  • Click Topics in the left sidebar.
  • You should see both customers and products listed as topics.

Topics not showing in Kafka UI?

  • Try refreshing the page. Kafka UI may take a moment to pick up newly created topics.
  • Verify the topics exist from the terminal by running docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list to see all topics.
  • help me figure out why my Kafka topics are not appearing in Kafka UI

Your data channels and storage are ready. Next up, you will build the Python producers that push real data into these topics.

Build the Multi-Source Producers

Your infrastructure is running and your Kafka topics and PostgreSQL tables are waiting for data. Now you need something to actually push data into those topics.

In a real e-commerce company, data arrives from multiple systems in different formats. A legacy CRM might export customer profiles as CSV files, while a product catalog streams as JSON from a REST API. In this step, you will write two separate Kafka producers that simulate exactly this scenario.

In this step, get ready to:
  • Create a sample CSV file and a producer that publishes customer data to Kafka.
  • Write an API producer that fetches product data from a REST endpoint and publishes it to Kafka.
  • Run both producers and verify messages appear in Kafka UI.
Create the customer CSV data file

The CSV producer needs a data file to read from. This simulates a batch export from a legacy CRM system.

  • Create a data folder inside your project directory by running this command:
mkdir data
  • Create a new file called customers.csv inside the data folder and paste in this content:
id,name,email,city,signup_date
1,Alice Johnson,alice@example.com,New York,2024-01-15
2,Bob Smith,bob.smith@email.com,Los Angeles,2024-02-20
3,Carol White,carol.white@company.org,Chicago,2024-03-10
4,David Brown,david.b@mail.com,Houston,2024-04-05
5,Eva Martinez,eva.martinez@web.net,Phoenix,2024-05-12
6,Frank Wilson,,Dallas,2024-06-01
7,Grace Lee,grace.lee@inbox.com,San Diego,2024-07-22
8,Henry Taylor,henry@domain.com,San Jose,2024-08-30
9,Iris Davis,iris.davis@provider.com,Austin,2024-09-14
10,Jack Anderson,jack.a@service.io,Denver,2024-10-08

What is this data?

This is a 10-row customer dataset with columns for ID, name, email, city, and signup date. Each row represents a customer profile exported from a fictional CRM.

Notice row 6 (Frank Wilson) has a missing email field. This is intentional. Real-world data is messy, and this empty value will come into play later in the Secret Mission.

  • Save data/customers.csv.
  • Confirm the file exists by running:
ls data

You should see customers.csv in the output.

Write the CSV producer script

A Kafka producer is a program that writes messages into a topic. Your CSV producer will read each row from the customer file and publish it as a JSON message.

  • Create a new file called csv_producer.py in your project root (the same folder as docker-compose.yml) and paste in this code:
import csv
import json
import time
from confluent_kafka import Producer


# Callback function that runs after each message is delivered (or fails)
def delivery_report(err, msg):
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}]")


def main():
    # Connect to the Kafka broker running on localhost
    producer = Producer({
        "bootstrap.servers": "localhost:9092"
    })

    # Open the CSV file and read each row as a dictionary
    with open("data/customers.csv", "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            # Convert the row dictionary to a JSON string
            message = json.dumps(row)
            # Publish the message to the 'customers' topic
            producer.produce(
                topic="customers",
                key=row["id"],
                value=message,
                callback=delivery_report
            )
            # Trigger delivery report callbacks without blocking
            producer.poll(0)
            # Small delay to simulate real-time streaming
            time.sleep(0.5)

    # Wait for any remaining messages to be delivered
    producer.flush()
    print("All customer messages produced.")


if __name__ == "__main__":
    main()

What does this code do?

  • delivery_report() is a callback that Kafka triggers after each message is successfully delivered or fails. It gives you real-time feedback in the terminal.
  • Producer({"bootstrap.servers": "localhost:9092"}) creates a connection to your local Kafka broker through the host-exposed port.
  • csv.DictReader(f) turns each CSV row into a Python dictionary using the header row as keys.
  • producer.produce() publishes each row as a JSON-serialized message. The key is the customer ID, which Kafka uses for partition routing.
  • producer.flush() blocks until all buffered messages are confirmed delivered. Without it, the script could exit before messages reach Kafka.
  • Save csv_producer.py.

You won't run this script yet. You will run both producers together in the final substep to verify them side by side.

Seeing a red underline on the import?

If your editor shows an error on from confluent_kafka import Producer, make sure you installed the Python packages in Step 1. Re-run pip install confluent-kafka==2.14.2 and check that you are using the same Python environment your editor is configured for.

Help me troubleshoot my confluent-kafka import error

✔️ Awesome, I've got everything!

Great. Make sure you have saved both files before moving on.

ⓧ I'd like to double check the full code

import csv
import json
import time
from confluent_kafka import Producer


def delivery_report(err, msg):
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}]")


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

    with open("data/customers.csv", "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            message = json.dumps(row)
            producer.produce(
                topic="customers",
                key=row["id"],
                value=message,
                callback=delivery_report
            )
            producer.poll(0)
            time.sleep(0.5)

    producer.flush()
    print("All customer messages produced.")


if __name__ == "__main__":
    main()
Write the API producer script

Your second data source is a REST API. The DummyJSON API provides free sample product data that simulates a real product catalog service.

  • Create a new file called api_producer.py in your project root and paste in this code:
import json
import time
import requests
from confluent_kafka import Producer


# Callback function that runs after each message is delivered (or fails)
def delivery_report(err, msg):
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}]")


def main():
    # Connect to the Kafka broker running on localhost
    producer = Producer({
        "bootstrap.servers": "localhost:9092"
    })

    # Fetch 20 products from the DummyJSON API
    response = requests.get("https://dummyjson.com/products?limit=20")
    response.raise_for_status()
    data = response.json()

    # Extract only the fields we need from each product
    for product in data["products"]:
        message = json.dumps({
            "id": product["id"],
            "title": product["title"],
            "price": product["price"],
            "category": product["category"],
            "stock": product["stock"]
        })
        # Publish to the 'products' topic with product ID as the key
        producer.produce(
            topic="products",
            key=str(product["id"]),
            value=message,
            callback=delivery_report
        )
        # Trigger delivery report callbacks without blocking
        producer.poll(0)
        # Small delay to simulate real-time streaming
        time.sleep(0.5)

    # Wait for any remaining messages to be delivered
    producer.flush()
    print("All product messages produced.")


if __name__ == "__main__":
    main()

What does this code do?

  • requests.get("https://dummyjson.com/products?limit=20") fetches 20 product records from the DummyJSON API as JSON.
  • response.raise_for_status() throws an error if the API returns a non-200 status code, so you catch failures immediately.
  • The loop extracts only five fields (id, title, price, category, stock) from the raw API response, which contains many more fields. This is a basic form of data transformation at the ingestion layer.
  • The key is converted with str(product["id"]) because Kafka message keys must be strings, but the API returns numeric IDs.

Why two separate producers instead of one?

In production data pipelines, each source system typically has its own ingestion process. This keeps producers independent so one can fail without affecting the other. It also lets each run on its own schedule (batch CSV nightly, API polling every 5 minutes).

  • Save api_producer.py.

Seeing an error on the requests import?

Make sure you installed the requests library in Step 1. Run pip install requests==2.32.3 to install it.

Help me fix my requests import issue

✔️ 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

import json
import time
import requests
from confluent_kafka import Producer


def delivery_report(err, msg):
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}]")


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

    response = requests.get("https://dummyjson.com/products?limit=20")
    response.raise_for_status()
    data = response.json()

    for product in data["products"]:
        message = json.dumps({
            "id": product["id"],
            "title": product["title"],
            "price": product["price"],
            "category": product["category"],
            "stock": product["stock"]
        })
        producer.produce(
            topic="products",
            key=str(product["id"]),
            value=message,
            callback=delivery_report
        )
        producer.poll(0)
        time.sleep(0.5)

    producer.flush()
    print("All product messages produced.")


if __name__ == "__main__":
    main()
Run both producers and verify in Kafka UI

Both producer scripts are ready. Time to run them and confirm that messages flow into your Kafka topics.

  • Run the CSV producer by executing this command:
python csv_producer.py

You should see 10 delivery confirmation messages printed in your terminal, one every half second. The final line reads All customer messages produced..

Seeing a connection refused error?

This means the producer cannot reach Kafka on localhost:9092. Make sure your Docker Compose stack is still running. Check with docker compose ps and confirm the kafka container status is healthy.

Help me troubleshoot my Kafka connection error

  • Run the API producer by executing this command:
python api_producer.py

You should see 20 delivery confirmation messages printed in your terminal. The final line reads All product messages produced..

Getting a network or timeout error?

The API producer needs internet access to reach https://dummyjson.com/products?limit=20. Check your internet connection. If you are behind a corporate proxy, you may need to configure your proxy settings for Python.

Help me troubleshoot my DummyJSON API request failure

  • Open Kafka UI in your browser at http://localhost:8080.
  • Click **Topics** in the left sidebar.
  • Click the customers topic. You should see 10 messages listed.
  • Go back to **Topics** and click the products topic. You should see 20 messages listed.

What should I see in Kafka UI?

Each topic page shows the messages stored in that topic. You can click on individual messages to see their JSON content. Customer messages contain the CSV fields (id, name, email, city, signup_date). Product messages contain the five extracted API fields (id, title, price, category, stock).

Both producers are publishing data into Kafka successfully. Next up, you will build a consumer that reads from both topics and loads the data into PostgreSQL.

Build the Consumer and Load into PostgreSQL

Your producers are publishing data from two different sources into Kafka. Now you need something on the other side to pick up those messages and put them somewhere useful.

In this step, you will write a consumer that subscribes to both Kafka topics, deserializes each message back into structured data, and inserts it into the correct PostgreSQL table. This completes the streaming pipeline from source to database.

In this step, get ready to:
  • Write a Kafka consumer that reads from both topics and routes data to PostgreSQL.
  • Run the consumer and verify data lands in the database with SELECT queries.
  • Execute the full pipeline end-to-end and inspect consumer group lag in Kafka UI.
Write the consumer script

The consumer connects to both Kafka and PostgreSQL. It reads messages from the customers and products topics, figures out which table each message belongs to based on the topic name, and inserts the data.

  • Create a new file called consumer.py in your kafka-streaming-pipeline project folder.
  • Add the imports and the create_tables() function by pasting this code:
import json
import psycopg2
from confluent_kafka import Consumer, KafkaError


# Ensure the target tables exist before inserting
def create_tables(conn):
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS raw_customers (
                id VARCHAR(10) PRIMARY KEY,
                name VARCHAR(100),
                email VARCHAR(100),
                city VARCHAR(50),
                signup_date VARCHAR(20),
                ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS raw_products (
                id INTEGER PRIMARY KEY,
                title VARCHAR(200),
                price NUMERIC(10,2),
                category VARCHAR(100),
                stock INTEGER,
                ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );
        """)
        conn.commit()

What does this code do?

The imports bring in psycopg2 for PostgreSQL connections, Consumer for reading from Kafka, and KafkaError for handling partition-level events.

The create_tables() function uses CREATE TABLE IF NOT EXISTS so the consumer is self-sufficient. It can guarantee its target tables exist every time it starts, without depending on a separate setup script.

  • Below the create_tables() function, add the two insert helper functions:
# Insert or update a single customer record
def insert_customer(conn, record):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO raw_customers (id, name, email, city, signup_date)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (id) DO UPDATE SET
                name = EXCLUDED.name,
                email = EXCLUDED.email,
                city = EXCLUDED.city,
                signup_date = EXCLUDED.signup_date;
        """, (record["id"], record["name"], record["email"],
              record["city"], record["signup_date"]))
        conn.commit()


# Insert or update a single product record
def insert_product(conn, record):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO raw_products (id, title, price, category, stock)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (id) DO UPDATE SET
                title = EXCLUDED.title,
                price = EXCLUDED.price,
                category = EXCLUDED.category,
                stock = EXCLUDED.stock;
        """, (record["id"], record["title"], record["price"],
              record["category"], record["stock"]))
        conn.commit()

Why ON CONFLICT DO UPDATE?

This is an upsert pattern. If a record with the same primary key already exists, PostgreSQL updates the row instead of throwing a duplicate-key error. This makes the consumer safe to re-run without duplicating data.

The EXCLUDED keyword refers to the incoming row that would have been inserted. It lets you reference the new values in the update clause.

  • Below the insert helpers, add the main() function and the entry point:
def main():
    # Connect to PostgreSQL
    conn = psycopg2.connect(
        host="localhost",
        port=5432,
        dbname="ecommerce",
        user="pipeline",
        password="pipeline123"
    )
    create_tables(conn)

    # Configure the Kafka consumer
    consumer = Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "ecommerce-pipeline",
        "auto.offset.reset": "earliest"
    })
    consumer.subscribe(["customers", "products"])

    print("Consumer started. Waiting for messages...")
    try:
        while True:
            msg = consumer.poll(timeout=5.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                print(f"Consumer error: {msg.error()}")
                break

            # Route message to the correct table based on topic
            topic = msg.topic()
            value = json.loads(msg.value().decode("utf-8"))

            if topic == "customers":
                insert_customer(conn, value)
                print(f"Inserted customer: {value['name']}")
            elif topic == "products":
                insert_product(conn, value)
                print(f"Inserted product: {value['title']}")

    except KeyboardInterrupt:
        print("Shutting down consumer...")
    finally:
        consumer.close()
        conn.close()


if __name__ == "__main__":
    main()

What does the main function do?

First, it opens a connection to PostgreSQL and ensures the tables exist. Then it creates a Kafka consumer with group.id set to ecommerce-pipeline. Kafka uses this group ID to track which messages have already been read.

The auto.offset.reset setting of earliest tells the consumer to start from the very first message if no prior offset exists for this group. This means the consumer picks up all existing messages on its first run.

Inside the loop, consumer.poll() waits up to 5 seconds for a message. If one arrives, the code checks the topic name and routes the deserialized JSON to the matching insert function. Pressing Ctrl+C triggers a clean shutdown.

  • Save consumer.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

import json
import psycopg2
from confluent_kafka import Consumer, KafkaError


def create_tables(conn):
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS raw_customers (
                id VARCHAR(10) PRIMARY KEY,
                name VARCHAR(100),
                email VARCHAR(100),
                city VARCHAR(50),
                signup_date VARCHAR(20),
                ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );
        """)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS raw_products (
                id INTEGER PRIMARY KEY,
                title VARCHAR(200),
                price NUMERIC(10,2),
                category VARCHAR(100),
                stock INTEGER,
                ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );
        """)
        conn.commit()


def insert_customer(conn, record):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO raw_customers (id, name, email, city, signup_date)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (id) DO UPDATE SET
                name = EXCLUDED.name,
                email = EXCLUDED.email,
                city = EXCLUDED.city,
                signup_date = EXCLUDED.signup_date;
        """, (record["id"], record["name"], record["email"],
              record["city"], record["signup_date"]))
        conn.commit()


def insert_product(conn, record):
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO raw_products (id, title, price, category, stock)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (id) DO UPDATE SET
                title = EXCLUDED.title,
                price = EXCLUDED.price,
                category = EXCLUDED.category,
                stock = EXCLUDED.stock;
        """, (record["id"], record["title"], record["price"],
              record["category"], record["stock"]))
        conn.commit()


def main():
    conn = psycopg2.connect(
        host="localhost",
        port=5432,
        dbname="ecommerce",
        user="pipeline",
        password="pipeline123"
    )
    create_tables(conn)

    consumer = Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "ecommerce-pipeline",
        "auto.offset.reset": "earliest"
    })
    consumer.subscribe(["customers", "products"])

    print("Consumer started. Waiting for messages...")
    try:
        while True:
            msg = consumer.poll(timeout=5.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                print(f"Consumer error: {msg.error()}")
                break

            topic = msg.topic()
            value = json.loads(msg.value().decode("utf-8"))

            if topic == "customers":
                insert_customer(conn, value)
                print(f"Inserted customer: {value['name']}")
            elif topic == "products":
                insert_product(conn, value)
                print(f"Inserted product: {value['title']}")

    except KeyboardInterrupt:
        print("Shutting down consumer...")
    finally:
        consumer.close()
        conn.close()


if __name__ == "__main__":
    main()
Run the consumer and verify the data

Your topics already have messages from the producers you ran in the previous step. The consumer will read all of them on its first run because auto.offset.reset is set to earliest.

  • Run the consumer by executing this command:
python consumer.py

You should see output like this as the consumer processes each message:

Consumer started. Waiting for messages...
Inserted customer: Alice Johnson
Inserted customer: Bob Smith
...
Inserted product: Essence Mascara Lash Princess
Inserted product: Eyeshadow Palette with Mirror
...
  • Wait until the output stops (the consumer is still running but has processed all existing messages).
  • Press Ctrl+C to stop the consumer.

You should see the shutdown message confirming it closed cleanly.

  • Verify the customer data landed in PostgreSQL by running this query:
docker exec -it postgres psql -U pipeline -d ecommerce -c "SELECT id, name, city FROM raw_customers ORDER BY id;"
  • Verify the product data by running:
docker exec -it postgres psql -U pipeline -d ecommerce -c "SELECT id, title, price FROM raw_products ORDER BY id LIMIT 5;"

You should see 10 customer rows and at least 5 product rows with titles, prices, and IDs matching the data your producers published.

Not seeing any rows?

  • Make sure the Docker Compose stack is still running. Run docker compose ps to check that the postgres and kafka containers show as healthy.
  • Check the consumer output for error messages. A Connection refused error means PostgreSQL is not reachable on port 5432.
  • Confirm you ran both producers in the previous step. If the topics are empty, the consumer has nothing to read.

Why isn't my consumer inserting data into PostgreSQL?

Run the full pipeline end-to-end

Now you will run the entire flow in real time. The consumer listens for new messages while the producers publish them. This proves the pipeline works as a continuous stream, not just a one-time batch load.

  • Start the consumer again by running:
python consumer.py

The consumer starts and waits. It has already read the earlier messages, so it sits idle until new ones arrive.

  • Open a second terminal window.
  • Navigate to your project folder in the new terminal.
  • Run both producers one after the other:
python csv_producer.py
python api_producer.py

Switch back to the terminal running the consumer. You should see it printing insert confirmations in real time as each message arrives from the producers.

  • After the consumer finishes processing all messages, verify the row counts by running:
docker exec -it postgres psql -U pipeline -d ecommerce -c "SELECT 'customers' AS table_name, COUNT(*) FROM raw_customers UNION ALL SELECT 'products', COUNT(*) FROM raw_products;"

You should see 10 rows for customers and 20 rows for products. The upsert logic means running the producers multiple times does not create duplicates.

  • Open Kafka UI at http://localhost:8080 in your browser.
  • Click on Consumer Groups in the left sidebar.
  • Find the ecommerce-pipeline group. The lag column should show 0 for both topics, meaning the consumer has fully caught up.

What is consumer lag?

Consumer lag is the difference between the latest message offset in a topic and the offset the consumer group has processed up to. A lag of 0 means the consumer has read every available message.

In production pipelines, monitoring lag is critical. Growing lag means your consumer is falling behind the rate of incoming data, which can signal performance problems or downstream bottlenecks.

  • Press Ctrl+C in the consumer terminal to stop it.

Consumer lag is not zero?

  • Make sure you waited for the consumer to finish processing before checking Kafka UI. If the producers just finished, give the consumer a few seconds to catch up.
  • If you restarted the consumer with a different group.id, Kafka UI may show a separate group. Confirm the group name is exactly ecommerce-pipeline.

Why is my Kafka consumer group showing lag in Kafka UI?

Your streaming pipeline is fully operational. Data flows from CSV files and REST APIs, through Kafka topics, and into PostgreSQL tables ready for analytics. Next up, you will extend this pipeline with production-grade error handling in the Secret Mission.

Secret mission

Add a Dead Letter Queue for Data Quality

In production pipelines, bad data doesn't disappear. It needs somewhere to go. Can you build a dead letter queue that catches malformed records and preserves them for investigation, keeping your main PostgreSQL tables clean?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally using Docker containers, so there are no ongoing cloud costs. However, the containers do use system memory while running. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker containers (kafka, postgres, kafka-ui) managed by Docker Compose.
  • Docker volume (pgdata) storing PostgreSQL data.
  • Kafka topics (customers, products, dead-letter-queue) inside the Kafka container.
  • PostgreSQL tables (raw_customers, raw_products) inside the PostgreSQL container.
  • Local project files (docker-compose.yml, Python scripts, data/customers.csv).
  • Python packages (confluent-kafka, psycopg2-binary, requests).

Keep everything running

No action needed. Choose this if you are still actively building on this pipeline or plan to experiment more with Kafka topics and consumers.

  • Your Docker containers will continue running in the background and use system memory.
  • Kafka UI remains accessible at http://localhost:8080.
  • PostgreSQL remains accessible on port 5432.
  • The Python packages (confluent-kafka, psycopg2-binary, requests) remain installed in your Python environment.

Pause - I'll come back to this later

Stop the Docker containers to free up system memory. Your project files and Docker volume stay intact so you can resume later.

  • Open a terminal and navigate to the folder containing your docker-compose.yml file.
  • Stop all containers by running:
docker compose stop
  • When you are ready to resume, start the containers again by running:
docker compose start

What does stop vs down do?

The docker compose stop command stops containers without removing them. Your data in the pgdata volume and Kafka topics are preserved. Use docker compose start to bring everything back up exactly as you left it.

Delete - I don't want to use this again

Remove all Docker containers, volumes, project files, and Python packages for a clean slate.

  • Open a terminal and navigate to the folder containing your docker-compose.yml file.
  • Stop and remove all containers, networks, and volumes by running:
docker compose down -v

What does the -v flag do?

The -v flag removes named volumes declared in your Compose file (like pgdata). This permanently deletes all PostgreSQL data. Without -v, the volume persists even after containers are removed.

  • Delete the project folder from your Desktop. On macOS, open Finder, navigate to your Desktop, right-click the project folder, and select Move to Trash. On Windows, open File Explorer, navigate to your Desktop, right-click the project folder, and select Delete.
  • Uninstall the Python packages by running:
pip uninstall confluent-kafka psycopg2-binary requests -y
  • Verify the containers are gone by running:
docker ps -a

You should no longer see the kafka, postgres, or kafka-ui containers in the list.

Nice Work!

Nice Work!

Nice work! You just built a real-time streaming data pipeline that ingests data from multiple sources, flows it through Apache Kafka topics, and lands it in PostgreSQL as clean, queryable tables.

You've learned how to:

  • Built a multi-service streaming infrastructure with Kafka, PostgreSQL, and Kafka UI using Docker Compose.
  • Wrote Python producers that ingest data from two different sources (CSV files and REST API) into separate Kafka topics.
  • Built a Kafka consumer that normalizes and loads multi-source data into PostgreSQL tables ready for analytics.
  • Implemented a dead letter queue pattern for production-grade error handling and data quality validation.

Ready to quiz yourself?