Build a Real-Time Crypto Streaming Pipeline

Stream live crypto prices through Apache Kafka into PostgreSQL.

Introduction

30 Second Summary

Every time you check a stock price, see a live sports score, or get a fraud alert on your phone, a streaming data pipeline is working behind the scenes. These systems process millions of events per second without losing a single one.

In this project, you will build a real-time streaming pipeline that ingests live cryptocurrency prices, streams them through Apache Kafka, and stores processed results in PostgreSQL for analytics.

What You'll Build

A complete local data pipeline using Docker, Python, and PostgreSQL that continuously streams live crypto market data through Apache Kafka.

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

  • A Kafka broker running in Docker using modern KRaft mode with no Zookeeper dependency.
  • A Python producer script that polls a free crypto API and publishes price events to a Kafka topic.
  • A Python consumer script that reads events, processes batches, and writes analytics to PostgreSQL.
  • Secret Mission: A real-time price alert system that detects when coins cross configurable thresholds.

Are there any prerequisites?

You should be familiar with Python basics and simple SQL queries. You also need Docker Desktop and PostgreSQL installed locally on your machine.

Set Up the Project and Start Kafka

Every streaming pipeline needs a message broker at its center. The broker receives data from producers, stores it temporarily, and delivers it to consumers on demand.

In this step, you will set up Apache Kafka as your broker using Docker. You will configure Kafka in KRaft mode, which removes the need for a separate Zookeeper service. Then you will create a topic where your crypto price data will flow.

In this step, get ready to:
  • Create a Docker Compose file that runs a Kafka broker.
  • Start the Kafka broker container and verify it is running.
  • Create a Kafka topic for streaming cryptocurrency prices.
Create the Docker Compose file

Docker Compose lets you define and run multi-container applications with a single configuration file. For this project, you need one container running a Kafka broker.

  • In VS Code, create a new file by pressing Cmd+N (macOS) or Ctrl+N (Windows).
  • Copy the following Docker Compose configuration into the file:
services:
  broker:
    image: apache/kafka:4.3.0
    container_name: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://localhost:9092,CONTROLLER://localhost:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_NUM_PARTITIONS: 3

What does this configuration do?

This file defines a single Kafka broker using the official apache/kafka:4.3.0 image. Here is what the key settings control:

  • KAFKA_PROCESS_ROLES: broker,controller runs Kafka in KRaft mode, combining both roles in a single node. No separate Zookeeper container needed.
  • KAFKA_LISTENERS exposes port 9092 for your Python scripts to connect, and port 9093 for internal controller communication.
  • KAFKA_NUM_PARTITIONS: 3 splits incoming data across 3 partitions by default, enabling parallel processing.
  • Save the file as docker-compose.yml by pressing Cmd+S (macOS) or Ctrl+S (Windows) and typing docker-compose.yml as the filename.
  • Confirm the file exists by running this in your terminal:
ls

You should see docker-compose.yml in the output.

Don't see docker-compose.yml?

Make sure you saved the file in your project folder, not in a different directory. Check the VS Code Explorer sidebar to confirm the file appears at the top level of your workspace.

help me find where my docker-compose.yml was saved

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

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

services:
  broker:
    image: apache/kafka:4.3.0
    container_name: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://localhost:9092,CONTROLLER://localhost:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_NUM_PARTITIONS: 3
Start the Kafka broker

With the configuration in place, you can now start the Kafka broker as a Docker container. Docker Desktop must be running first.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Docker Desktop and press Enter to open it.
  • Wait until Docker Desktop shows the engine is running (the whale icon in your system tray stops animating).
  • Start the Kafka broker by running this command in your VS Code terminal:
docker compose up -d

Docker will pull the Kafka image (this may take a minute the first time) and start the broker container in the background.

  • Verify the broker is running by running:
docker compose ps

You should see the broker container listed with a status of running.

Container not running or exited immediately?

  • Make sure Docker Desktop is fully started before running docker compose up -d. If the engine is still starting, the command will fail.
  • Check if port 9092 is already in use by another application. Stop any other Kafka or service using that port.
  • Run docker compose logs broker to see the container's startup logs for specific error messages.

help me debug why my Kafka Docker container is not starting

Create the crypto-prices topic

A Kafka topic is a named channel where messages are published and consumed. You need to create one for your crypto price data before the producer can send messages to it.

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

What does this command do?

  • docker exec -it broker runs a command inside your running Kafka container.
  • kafka-topics.sh --create tells Kafka to create a new topic.
  • --partitions 3 splits the topic into 3 partitions. This lets multiple consumers read in parallel later.
  • --replication-factor 1 keeps one copy of the data (appropriate for a single-node local setup).

You should see a confirmation message that the topic was created successfully.

  • Verify the topic exists by listing all topics:
docker exec -it broker /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

You should see crypto-prices in the output.

Topic creation failed?

  • If you see a connection error, the broker may still be starting up. Wait 10-15 seconds after docker compose up -d and try again.
  • If the error says the topic already exists, that means it was created on a previous attempt. Run the --list command to confirm it is there.

help me fix the Kafka topic creation error

Why KRaft instead of Zookeeper?

Older Kafka versions required a separate Zookeeper service to manage cluster metadata. Starting with Kafka 3.3, KRaft mode handles metadata internally. Kafka 4.x runs exclusively in KRaft mode.

This means your Docker setup only needs one container instead of two. It is also what you will encounter in production environments going forward.

Your Kafka broker is live and your topic is ready to receive messages. Next up, you will build a Python producer that fetches live cryptocurrency prices and publishes them to this topic.

Build the Producer

Your Kafka broker is running and the crypto-prices topic is ready to receive messages. A streaming system without data flowing through it is just an empty pipe.

In this step, you will prepare PostgreSQL as the destination for your stream, then build a Python producer that fetches live cryptocurrency prices and publishes them into Kafka.

In this step, get ready to:
  • Create a PostgreSQL database and table to store crypto price history.
  • Build a Python producer that fetches live prices from the CoinGecko API.
  • Publish price events to the crypto-prices Kafka topic every 30 seconds.
Set up the PostgreSQL database

Before your consumer can write streaming data anywhere, you need a database ready to receive it. You will create a setup script that builds the database and table schema in one go.

  • In VS Code, create a new file called setup_db.py in your project folder.
  • Paste the following code into setup_db.py:
import psycopg2
from psycopg2 import sql


def setup_database():
    conn = psycopg2.connect(
        host="localhost",
        database="postgres",
        user="postgres",
        password="your-password-here",
    )
    conn.autocommit = True

    with conn.cursor() as cur:
        cur.execute("SELECT 1 FROM pg_database WHERE datname = 'crypto_stream'")
        if not cur.fetchone():
            cur.execute(sql.SQL("CREATE DATABASE {}").format(sql.Identifier("crypto_stream")))
            print("Created database: crypto_stream")
        else:
            print("Database crypto_stream already exists")

    conn.close()

    conn = psycopg2.connect(
        host="localhost",
        database="crypto_stream",
        user="postgres",
        password="your-password-here",
    )
    conn.autocommit = True

    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS crypto_prices (
                id SERIAL PRIMARY KEY,
                asset_id VARCHAR(50) NOT NULL,
                symbol VARCHAR(20) NOT NULL,
                name VARCHAR(100) NOT NULL,
                price_usd DECIMAL(20, 8),
                change_percent_24hr DECIMAL(10, 4),
                volume_usd_24hr DECIMAL(20, 2),
                event_timestamp BIGINT NOT NULL,
                ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );
        """)
        print("Created table: crypto_prices")

        cur.execute("""
            CREATE INDEX IF NOT EXISTS idx_crypto_prices_symbol
            ON crypto_prices (symbol);
        """)
        cur.execute("""
            CREATE INDEX IF NOT EXISTS idx_crypto_prices_timestamp
            ON crypto_prices (event_timestamp);
        """)
        print("Created indexes")

    conn.close()
    print("\nDatabase setup complete!")


if __name__ == "__main__":
    setup_database()

What does this code do?

The script connects to PostgreSQL twice. The first connection targets the default postgres database to create the crypto_stream database if it does not exist yet.

The second connection targets crypto_stream directly, creating the crypto_prices table with columns for price data, volume, and timestamps. Two indexes on symbol and event_timestamp speed up queries you will run later.

Replace the password placeholder

Replace both instances of your-password-here with your actual PostgreSQL password before running the script.

  • Save setup_db.py.
  • Run the setup script by executing this command in your terminal:
python setup_db.py
  • You should see output confirming the database, table, and indexes were created:
Created database: crypto_stream
Created table: crypto_prices
Created indexes

Database setup complete!

Seeing a connection refused error?

  • Make sure PostgreSQL is running on your machine. Check with pg_isready in your terminal.
  • Confirm your PostgreSQL password matches what you entered in the script.
  • If PostgreSQL is running on a non-default port, add port=5433 (or your port) to the psycopg2.connect() call.

Why can't my script connect to PostgreSQL?

Create the Kafka producer script

A Kafka producer is a client that publishes messages to a topic. Your producer will poll a free cryptocurrency API every 30 seconds and push each coin's price data into the crypto-prices topic.

  • Create a new file called producer.py in your project folder.
  • Start by adding the imports and the delivery callback function:
import json
import time

import requests
from confluent_kafka import Producer


def delivery_report(err, msg):
    # Callback that fires once Kafka acknowledges (or rejects) a message
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}")

What does this code do?

The imports bring in JSON serialization, time delays, HTTP requests, and the confluent-kafka Producer class.

The delivery_report function is a callback that Kafka calls once it confirms a message landed on the broker. It prints the topic, partition, and offset so you can verify delivery in real time.

  • Below the delivery_report function, add the fetch_crypto_prices function:
def fetch_crypto_prices():
    # Hit the CoinGecko API for the top 10 coins by market cap
    url = "https://api.coingecko.com/api/v3/coins/markets"
    params = {
        "vs_currency": "usd",
        "order": "market_cap_desc",
        "per_page": 10,
        "page": 1,
    }
    response = requests.get(url, params=params)
    response.raise_for_status()
    return response.json()

What does this code do?

This function calls the CoinGecko /coins/markets endpoint, which returns bulk market data for coins. The parameters request the top 10 coins sorted by market cap, priced in USD.

The API is free to use without an API key. Each response includes fields like current_price, price_change_percentage_24h, and total_volume for every coin.

  • Below the fetch_crypto_prices function, add the main function and the entry point:
def main():
    # Configure the producer to connect to the local Kafka broker
    conf = {
        "bootstrap.servers": "localhost:9092",
    }

    producer = Producer(conf)
    topic = "crypto-prices"

    print("Starting crypto price producer...")
    print("Press Ctrl+C to stop.\n")

    try:
        while True:
            coins = fetch_crypto_prices()

            for coin in coins:
                # Build a standardized message from the API response
                message = {
                    "id": coin["id"],
                    "symbol": coin["symbol"],
                    "name": coin["name"],
                    "priceUsd": str(coin["current_price"]),
                    "changePercent24Hr": str(coin.get("price_change_percentage_24h", 0)),
                    "volumeUsd24Hr": str(coin.get("total_volume", 0)),
                    "timestamp": int(time.time()),
                }

                producer.produce(
                    topic,
                    key=coin["id"],
                    value=json.dumps(message),
                    callback=delivery_report,
                )

            # Flush ensures all queued messages are sent before sleeping
            producer.flush()
            print(f"Published {len(coins)} prices. Waiting 30 seconds...")
            time.sleep(30)

    except KeyboardInterrupt:
        print("\nShutting down producer...")
    finally:
        producer.flush()


if __name__ == "__main__":
    main()

What does this code do?

  • The Producer connects to your local Kafka broker at localhost:9092.
  • Each API response is transformed into a standardized JSON message. Using key=coin["id"] ensures all messages for the same coin go to the same partition for ordered processing.
  • The producer.flush() call blocks until every queued message is acknowledged by the broker, guaranteeing no messages are lost.
  • The loop repeats every 30 seconds to stay within the CoinGecko free API rate limit.
  • Save producer.py.

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

Here is the complete producer.py file for comparison:

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()}] @ offset {msg.offset()}")


def fetch_crypto_prices():
    url = "https://api.coingecko.com/api/v3/coins/markets"
    params = {
        "vs_currency": "usd",
        "order": "market_cap_desc",
        "per_page": 10,
        "page": 1,
    }
    response = requests.get(url, params=params)
    response.raise_for_status()
    return response.json()


def main():
    conf = {
        "bootstrap.servers": "localhost:9092",
    }

    producer = Producer(conf)
    topic = "crypto-prices"

    print("Starting crypto price producer...")
    print("Press Ctrl+C to stop.\n")

    try:
        while True:
            coins = fetch_crypto_prices()

            for coin in coins:
                message = {
                    "id": coin["id"],
                    "symbol": coin["symbol"],
                    "name": coin["name"],
                    "priceUsd": str(coin["current_price"]),
                    "changePercent24Hr": str(coin.get("price_change_percentage_24h", 0)),
                    "volumeUsd24Hr": str(coin.get("total_volume", 0)),
                    "timestamp": int(time.time()),
                }

                producer.produce(
                    topic,
                    key=coin["id"],
                    value=json.dumps(message),
                    callback=delivery_report,
                )

            producer.flush()
            print(f"Published {len(coins)} prices. Waiting 30 seconds...")
            time.sleep(30)

    except KeyboardInterrupt:
        print("\nShutting down producer...")
    finally:
        producer.flush()


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

Time to see your streaming pipeline in action. The producer will fetch prices and push them into Kafka continuously until you stop it.

  • Start the producer by running:
python producer.py
  • Watch the terminal output. You should see 10 delivery confirmations per cycle, followed by a 30-second wait:
Starting crypto price producer...
Press Ctrl+C to stop.

Message delivered to crypto-prices [0] @ offset 0
Message delivered to crypto-prices [2] @ offset 0
Message delivered to crypto-prices [1] @ offset 0
...
Published 10 prices. Waiting 30 seconds...

What do the partition numbers mean?

Each message is routed to one of the 3 partitions you created earlier. Kafka hashes the message key (the coin ID) to determine which partition receives it. The same coin always lands on the same partition, preserving order per coin.

  • Let the producer run for at least two cycles (about 60 seconds) to confirm it repeats successfully.
  • Press Ctrl+C to stop the producer when you are ready.

Seeing a connection error or no messages?

  • Make sure the Kafka broker container is still running. Check with docker compose ps and confirm the broker shows a healthy status.
  • If you see a rate limit error from the API, wait 60 seconds and try again. The CoinGecko free tier allows about 50 calls per minute.
  • If confluent_kafka cannot be imported, confirm your virtual environment is activated and confluent-kafka is installed.

Why is my producer failing to connect or deliver messages?

Your producer is live. In the next step, you will build a consumer that reads these messages from Kafka and writes them into PostgreSQL for analytics.

Build the Consumer

Your producer is now steadily publishing cryptocurrency prices into Kafka. Messages are accumulating in the crypto-prices topic, waiting for something to read them.

In this step, you will build the other side of the pipeline: a consumer that subscribes to the topic, reads messages in batches, and writes them to PostgreSQL. This is the stream-to-database sink pattern used in production data pipelines.

In this step, get ready to:
  • Create a consumer script that subscribes to the Kafka topic.
  • Add batch processing logic that writes records to PostgreSQL.
  • Run the consumer and verify data flowing into the database.
Create the consumer script

The consumer needs two connections: one to Kafka (to read messages) and one to PostgreSQL (to write records). You will set up both connection helpers first.

  • Create a new file called consumer.py in your crypto-kafka-pipeline folder.
  • Add the imports and connection setup functions by pasting this code:
import json

import psycopg2
from psycopg2.extras import execute_values
from confluent_kafka import Consumer, KafkaException


def create_consumer():
    # Configure the Kafka consumer with group ID and offset settings
    conf = {
        "bootstrap.servers": "localhost:9092",
        "group.id": "crypto-pipeline",
        "auto.offset.reset": "earliest",
        "enable.auto.offset.store": False,
    }
    return Consumer(conf)


def connect_db():
    # Connect to the crypto_stream database
    conn = psycopg2.connect(
        host="localhost",
        database="crypto_stream",
        user="postgres",
        password="your-password-here",
    )
    return conn

What does this code do?

The create_consumer() function configures a Kafka consumer with four key settings:

  • bootstrap.servers tells the consumer where to find the Kafka broker.
  • group.id assigns this consumer to the crypto-pipeline consumer group, which Kafka uses to track what messages have been read.
  • auto.offset.reset set to earliest means the consumer starts reading from the beginning of the topic if no prior offset exists.
  • enable.auto.offset.store set to False gives you manual control over when offsets are committed, which prevents data loss.

The connect_db() function opens a connection to the crypto_stream database you created in Step 2.

Why manual offset management?

By default, Kafka auto-commits offsets on a timer. If your consumer crashes after committing but before writing to the database, those messages are lost forever.

With manual offset storage, you only mark a message as "done" after it is safely written to PostgreSQL. This gives you at-least-once delivery semantics. The same pattern is used in production streaming systems.

  • Save consumer.py.
  • Confirm the file exists by running:
ls consumer.py

You should see consumer.py listed in the output.

Don't see consumer.py?

Make sure you saved the file in the same crypto-kafka-pipeline folder where your other project files live. Run ls to check you can also see producer.py and docker-compose.yml in the same directory.

If not, use cd to navigate to the correct folder and try saving again.

help me find where my consumer.py was saved

Add batch processing and database writes

Instead of writing every single message to PostgreSQL individually, you will accumulate messages into batches of 10. Batch inserts reduce the number of database round-trips and dramatically improve throughput.

  • In consumer.py, add the write_to_db() function below the connect_db() function:
def write_to_db(conn, records):
    # SQL template for bulk inserting price records
    query = """
        INSERT INTO crypto_prices
        (asset_id, symbol, name, price_usd, change_percent_24hr, volume_usd_24hr, event_timestamp)
        VALUES %s
    """
    # Transform each record dict into a tuple matching the column order
    values = [
        (
            r["id"],
            r["symbol"],
            r["name"],
            float(r["priceUsd"]) if r["priceUsd"] else None,
            float(r["changePercent24Hr"]) if r["changePercent24Hr"] else None,
            float(r["volumeUsd24Hr"]) if r["volumeUsd24Hr"] else None,
            r["timestamp"],
        )
        for r in records
    ]

    # Execute the bulk insert and commit the transaction
    with conn.cursor() as cur:
        execute_values(cur, query, values)
    conn.commit()

What does this code do?

The write_to_db() function takes a list of message records and bulk-inserts them into the crypto_prices table.

  • execute_values() from psycopg2 generates a single SQL statement with all rows, which is much faster than inserting one row at a time.
  • The list comprehension transforms each JSON record into a tuple matching the table's column order.
  • conn.commit() finalizes the transaction so the data is permanently stored.
  • Save consumer.py.
  • You won't see a change yet. The next function will use write_to_db() to flush batches to the database.

Now add the main loop that polls Kafka, accumulates messages, and triggers batch writes.

  • Below write_to_db(), add the main() function and the entry point:
def main():
    consumer = create_consumer()
    conn = connect_db()
    topic = "crypto-prices"

    # Subscribe to the topic and start polling
    consumer.subscribe([topic])
    print("Starting crypto price consumer...")
    print("Press Ctrl+C to stop.\n")

    batch = []
    batch_size = 10

    try:
        while True:
            # Poll for a single message with 1-second timeout
            msg = consumer.poll(timeout=1.0)

            if msg is None:
                # No message available - flush any partial batch
                if batch:
                    write_to_db(conn, batch)
                    print(f"Flushed {len(batch)} records to PostgreSQL")
                    batch = []
                continue

            if msg.error():
                raise KafkaException(msg.error())

            # Deserialize the message and add to the batch
            record = json.loads(msg.value().decode("utf-8"))
            batch.append(record)
            print(f"  Consumed: {record['symbol']} @ ${float(record['priceUsd']):.2f}")

            # Write to DB when batch is full
            if len(batch) >= batch_size:
                write_to_db(conn, batch)
                print(f"Wrote batch of {len(batch)} records to PostgreSQL\n")
                batch = []

            # Store the offset after processing for at-least-once delivery
            consumer.store_offsets(msg)

    except KeyboardInterrupt:
        print("\nShutting down consumer...")
        if batch:
            write_to_db(conn, batch)
            print(f"Flushed final {len(batch)} records")
    finally:
        consumer.close()
        conn.close()


if __name__ == "__main__":
    main()

What does this code do?

The main() function runs an infinite loop that polls Kafka for new messages:

  • consumer.poll(timeout=1.0) waits up to 1 second for a new message. If none arrives, it flushes any partial batch to avoid stale data sitting in memory.
  • Each message is deserialized from JSON and appended to a batch list.
  • When the batch reaches 10 records, write_to_db() bulk-inserts them into PostgreSQL.
  • consumer.store_offsets(msg) marks the message as processed only after it has been safely handled. This is the at-least-once guarantee.
  • The KeyboardInterrupt handler ensures any remaining records in the batch are flushed before shutdown.
  • Save 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

import json

import psycopg2
from psycopg2.extras import execute_values
from confluent_kafka import Consumer, KafkaException


def create_consumer():
    conf = {
        "bootstrap.servers": "localhost:9092",
        "group.id": "crypto-pipeline",
        "auto.offset.reset": "earliest",
        "enable.auto.offset.store": False,
    }
    return Consumer(conf)


def connect_db():
    conn = psycopg2.connect(
        host="localhost",
        database="crypto_stream",
        user="postgres",
        password="your-password-here",
    )
    return conn


def write_to_db(conn, records):
    query = """
        INSERT INTO crypto_prices
        (asset_id, symbol, name, price_usd, change_percent_24hr, volume_usd_24hr, event_timestamp)
        VALUES %s
    """
    values = [
        (
            r["id"],
            r["symbol"],
            r["name"],
            float(r["priceUsd"]) if r["priceUsd"] else None,
            float(r["changePercent24Hr"]) if r["changePercent24Hr"] else None,
            float(r["volumeUsd24Hr"]) if r["volumeUsd24Hr"] else None,
            r["timestamp"],
        )
        for r in records
    ]

    with conn.cursor() as cur:
        execute_values(cur, query, values)
    conn.commit()


def main():
    consumer = create_consumer()
    conn = connect_db()
    topic = "crypto-prices"

    consumer.subscribe([topic])
    print("Starting crypto price consumer...")
    print("Press Ctrl+C to stop.\n")

    batch = []
    batch_size = 10

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

            if msg is None:
                if batch:
                    write_to_db(conn, batch)
                    print(f"Flushed {len(batch)} records to PostgreSQL")
                    batch = []
                continue

            if msg.error():
                raise KafkaException(msg.error())

            record = json.loads(msg.value().decode("utf-8"))
            batch.append(record)
            print(f"  Consumed: {record['symbol']} @ ${float(record['priceUsd']):.2f}")

            if len(batch) >= batch_size:
                write_to_db(conn, batch)
                print(f"Wrote batch of {len(batch)} records to PostgreSQL\n")
                batch = []

            consumer.store_offsets(msg)

    except KeyboardInterrupt:
        print("\nShutting down consumer...")
        if batch:
            write_to_db(conn, batch)
            print(f"Flushed final {len(batch)} records")
    finally:
        consumer.close()
        conn.close()


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

Your producer should still be running in its terminal from earlier. You will start the consumer in a separate terminal so both scripts run side by side.

  • Open a new terminal window.
  • Navigate to your crypto-kafka-pipeline folder and activate your virtual environment.
  • Start the consumer by running:
python consumer.py

You should see output like this as the consumer reads messages from Kafka:

Starting crypto price consumer...
Press Ctrl+C to stop.

  Consumed: btc @ $104532.00
  Consumed: eth @ $3845.21
  Consumed: sol @ $187.43
  ...
Wrote batch of 10 records to PostgreSQL

What should I see?

The consumer prints each message as it arrives, showing the coin symbol and current price. Once 10 messages accumulate, it writes the batch to PostgreSQL and prints a confirmation.

If no new messages arrive for 1 second (the poll timeout), the consumer flushes any partial batch so data does not sit in memory indefinitely.

Now verify the data actually landed in PostgreSQL.

  • Open a third terminal and connect to your database by running:
psql -U postgres -d crypto_stream
  • Check how many records exist by running this query:
SELECT COUNT(*) FROM crypto_prices;

You should see a count that grows each time the consumer writes a batch.

  • View the most recent records by running:
SELECT symbol, price_usd, event_timestamp FROM crypto_prices ORDER BY ingested_at DESC LIMIT 5;

You should see the latest cryptocurrency prices with their symbols, USD values, and timestamps.

Consumer starts but no messages appear?

  • Make sure your producer is still running in the other terminal. The consumer can only read messages that exist in the topic.
  • Check that the Kafka broker is still running with docker compose ps. If the container stopped, restart it with docker compose up -d.
  • Verify the topic exists by running: docker exec -it broker /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092.

help me debug why my Kafka consumer is not receiving messages

Seeing a database connection error?

  • Make sure PostgreSQL is running locally.
  • Check that the password in consumer.py matches the password you set when you ran setup_db.py.
  • Verify the crypto_stream database exists by running psql -U postgres -l and looking for it in the list.

help me fix my PostgreSQL connection in the consumer

Your streaming pipeline is now fully connected. Data flows from the CoinGecko API through Kafka and into PostgreSQL in real time. Next, you will run the full pipeline end-to-end and test its resilience by stopping and restarting the consumer.

Run the Full Pipeline End-to-End

You have all three components of a streaming pipeline built: a Kafka broker, a producer, and a consumer. Now it's time to run them together and watch real-time data flow from a public API, through Kafka, and into your database.

The real test of any streaming system is resilience. What happens when a consumer crashes? Does the data disappear, or can the system recover without losing a single message? You'll prove that your pipeline handles this gracefully.

In this step, get ready to:
  • Run all three pipeline components simultaneously across separate terminals.
  • Query PostgreSQL to see analytics across your collected price data.
  • Test pipeline resilience by stopping and restarting the consumer with zero data loss.
Run all three components simultaneously

A streaming pipeline has multiple moving parts that all need to run at the same time. You'll arrange three terminal windows so you can observe data flowing through each stage in real time.

  • Open three separate terminal windows and arrange them so you can see all three at once.
  • In Terminal 1, confirm your Kafka broker is still running by running this command:
docker compose ps

You should see the broker container with a status of Up.

  • In Terminal 2, navigate to your project folder, activate your virtual environment, and start the producer by running:
python producer.py

You should see delivery confirmations printing every 30 seconds, showing 10 messages delivered to the crypto-prices topic.

  • In Terminal 3, navigate to your project folder, activate your virtual environment, and start the consumer by running:
python consumer.py

You should see consumed prices printing in Terminal 3, followed by batch write confirmations as records flow into PostgreSQL.

What's happening across the three terminals?

You're watching the classic streaming pattern in action: source to broker to sink. Terminal 1 hosts the Kafka broker that decouples the producer from the consumer. Terminal 2 shows the producer fetching prices and publishing them. Terminal 3 shows the consumer reading those messages and writing them to the database.

This decoupling is the key advantage of streaming. The producer and consumer don't know about each other. They only know about Kafka.

Consumer not printing anything?

If the consumer already processed all existing messages during Step 3, it will wait silently for new ones. Wait 30 seconds for the producer to publish a fresh batch, then watch the consumer pick them up.

If you still see nothing after a full cycle, make sure both scripts are using the same topic name crypto-prices and connecting to localhost:9092.

Help me debug why my Kafka consumer isn't receiving messages from the producer.

Run an analytics query on your collected data

Your pipeline has been streaming data into PostgreSQL. Now you can query it to see aggregated analytics across all the price readings you've collected.

  • Let the pipeline run for at least 2-3 cycles (about 60-90 seconds) so you have enough data to see meaningful averages.
  • Open a new terminal window (a fourth one) and connect to your crypto_stream database by running:
psql -U postgres -d crypto_stream
  • Run this analytics query to see the average price per coin across all readings:
SELECT symbol, ROUND(AVG(price_usd)::numeric, 2) as avg_price, COUNT(*) as readings FROM crypto_prices GROUP BY symbol ORDER BY avg_price DESC;

What does this query do?

  • GROUP BY symbol groups all price records by their cryptocurrency symbol.
  • AVG(price_usd) calculates the average price across all readings for each coin.
  • COUNT(*) as readings shows how many data points you've collected per coin.
  • ORDER BY avg_price DESC sorts by most expensive coins first.

You should see a table with 10 rows showing each cryptocurrency's symbol, average price, and number of readings. The more cycles the pipeline has run, the higher the readings count.

  • Verify the total record count is growing by running:
SELECT COUNT(*) FROM crypto_prices;
  • Check the most recent records to confirm live data is flowing by running:
SELECT symbol, price_usd, event_timestamp FROM crypto_prices ORDER BY ingested_at DESC LIMIT 5;

You should see the 5 most recent price records with recent timestamps, proving your pipeline is actively streaming data into the database.

Query returns zero rows?

Make sure your consumer is running in Terminal 3 and has printed at least one batch write confirmation. The producer publishes data to Kafka, but only the consumer writes it to PostgreSQL.

Help me figure out why my crypto_prices table is empty even though my pipeline is running.

Test pipeline resilience with a stop/restart

In production, consumers crash. Networks blip. Containers restart. A reliable streaming pipeline must handle these disruptions without losing data. Kafka's offset management makes this possible.

What is offset management?

Every message in a Kafka topic has a numbered position called an offset. When your consumer reads a message and processes it successfully, it stores that offset. If the consumer crashes and restarts, it asks Kafka for the last stored offset and resumes from there.

This is why your consumer uses store_offsets explicitly after processing each message. It guarantees at-least-once delivery: no message is ever skipped.

  • In Terminal 3 (where the consumer is running), press Ctrl+C to stop the consumer.
  • Watch Terminal 2. The producer continues publishing prices every 30 seconds. These messages accumulate in Kafka with no consumer reading them.
  • Wait at least 30 seconds so that one or two batches of messages pile up in the topic.
  • Restart the consumer in Terminal 3 by running:
python consumer.py

Watch Terminal 3 closely. The consumer immediately processes the backlog of messages that accumulated while it was stopped. You should see a burst of consumed messages followed by batch writes to PostgreSQL.

  • Switch back to your psql session and verify no data was lost by running the count query again:
SELECT COUNT(*) FROM crypto_prices;

The count should be higher than before you stopped the consumer. Every message the producer published during the downtime has been recovered and written to the database.

Why is this important for production systems?

You just demonstrated at-least-once delivery semantics. In production, this means you can deploy a new version of your consumer, restart it after a crash, or scale it across multiple instances without worrying about gaps in your data.

Kafka retains messages for a configurable period (7 days by default). As long as your consumer reconnects within that window, it picks up exactly where it left off.

Consumer isn't processing the backlog?

Make sure your consumer uses the same group.id value (crypto-pipeline) every time you restart it. If the group ID changes, Kafka treats it as a brand new consumer and the offset history is lost.

Help me understand why my Kafka consumer isn't picking up messages from when it was offline.

You now have a proven, resilient streaming pipeline running on your machine. You've watched data flow from a live API through Kafka into PostgreSQL, queried real analytics, and confirmed that your system handles failures gracefully. Next up, you'll take it further with a Secret Mission that adds real-time alerting to your pipeline.

Secret mission

Add a Real-Time Price Alert System

Your pipeline streams and stores data. But what if you could make it react in real time? In this secret mission, you'll add stateful processing to your consumer that detects when coin prices cross configurable thresholds and logs alerts to PostgreSQL.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on your local machine, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker container running the Kafka broker.
  • PostgreSQL database crypto_stream with tables crypto_prices and alerts.
  • Project files on your Desktop (docker-compose.yml, producer.py, consumer.py, setup_db.py, and the Python virtual environment).

Keep everything running

No action needed. Choose this if you are still actively building or want to keep accumulating historical price data for analysis.

  • Your Docker container continues running in the background.
  • You can restart the producer and consumer scripts at any time to resume streaming.
  • Your PostgreSQL data persists between sessions.

Pause - I'll come back to this later

Shut down running processes to free up memory and CPU, but keep your files and data intact for later.

  • Stop the producer and consumer scripts by pressing Ctrl+C in each terminal window.
  • Pause the Kafka broker by running:
docker compose stop
  • When you are ready to resume, restart the broker by running:
docker compose start
  • Your topic data, consumer offsets, and PostgreSQL records remain intact.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

  • Stop the producer and consumer scripts by pressing Ctrl+C in each terminal window.
  • Remove the Kafka container and its volumes by running:
docker compose down -v
  • Drop the PostgreSQL database by opening psql and running:
DROP DATABASE crypto_stream;
  • Deactivate your Python virtual environment by running:
deactivate
  • Delete your project folder from the Desktop by running:
cd ~/Desktop
rm -rf kafka-crypto-pipeline

Folder name

Replace kafka-crypto-pipeline with whatever you named your project folder in Step 1.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade real-time streaming data pipeline that ingests live cryptocurrency prices, streams them through Apache Kafka, and sinks processed results into PostgreSQL.

You've learned how to:

  • Set up Apache Kafka in Docker using modern KRaft mode and built a Python producer that ingests real-time data from a public API and publishes to Kafka topics.
  • Built a Python consumer with batch processing and at-least-once delivery semantics, implementing a stream-to-database sink pattern writing to PostgreSQL.
  • Tested pipeline resilience by stopping and restarting the consumer with zero data loss, proving Kafka's offset management in action.
  • Secret Mission: Added stateful stream processing with configurable price alerts that detect threshold crossings and log them to PostgreSQL.

Ready to quiz yourself?