Build a Real-Time Analytics Pipeline

Stream 500K+ e-commerce records from PostgreSQL to Grafana via Kafka and ClickHouse.

Introduction

30 Second Summary

Every time you place an online order, that transaction needs to travel from the app's database all the way to a dashboard where analysts can spot trends. Modern companies build dedicated pipelines to move this data in real time.

In this project, you will build a complete data engineering pipeline that captures database changes, streams them through a message broker, transforms them for analytics, and visualizes the results in a live dashboard.

What You'll Build

A fully Dockerized modern data stack with seven interconnected services running on your local machine, processing 500K+ rows of e-commerce data from source to dashboard.

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

  • A CDC pipeline that automatically captures PostgreSQL changes and streams them through Apache Kafka into ClickHouse for analytics.
  • A dbt project with staging, intermediate, and mart models transforming raw data into business intelligence tables.
  • A Dagster orchestration layer managing dbt runs, paired with a Grafana dashboard visualizing revenue, order trends, and customer analytics.
  • Secret Mission: Add real-time streaming simulation to watch new orders flow through the entire pipeline and appear in Grafana within seconds.

Are there any prerequisites?

You should be familiar with SQL and basic Python. You also need Docker Desktop already installed on your machine, since the entire stack runs in containers.

Set Up the Project Structure and Python Environment

Your data pipeline needs a home. Before you can stream data from PostgreSQL through Kafka into ClickHouse, you need a well-organized project folder, a Python environment with the right packages, and a custom Docker image.

In this step, you will create the project directory structure, install the Python libraries your pipeline depends on, and build a custom Kafka Connect image that bundles both the Debezium source connector and the ClickHouse sink connector.

In this step, get ready to:
  • Create the project folder and directory structure.
  • Set up a Python virtual environment and install required packages.
  • Build a custom Kafka Connect Docker image with Debezium and ClickHouse connectors.
Create the project folder and directories

Every service in your pipeline will live inside a single project folder. Organizing it now saves confusion later.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir data-pipeline
cd data-pipeline
  • Create the subdirectories for Docker configuration, scripts, dbt models, and Dagster orchestration by running:
mkdir docker scripts dbt_project dagster_project
  • Confirm the directories exist by running:
ls

You should see docker, scripts, dbt_project, and dagster_project in the output.

What are these directories for?

  • docker/ holds custom Dockerfiles and configuration for your services.
  • scripts/ holds Python scripts for data generation and connector setup.
  • dbt_project/ holds your dbt models that transform raw data into analytics tables.
  • dagster_project/ holds your Dagster orchestration code that schedules and monitors dbt runs.
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type VS Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the data-pipeline folder, then click Open.

Don't see the folder on your Desktop?

Make sure you ran the mkdir commands from ~/Desktop. If you skipped the cd ~/Desktop step, the folder was created in your home directory instead.

help me find my data-pipeline folder

Set up the Python virtual environment

A virtual environment isolates the Python packages for this project from your system-wide installations. This prevents version conflicts between projects.

  • In VS Code, open the integrated terminal by pressing Ctrl+` (backtick key, located above Tab on most keyboards).
  • Create a virtual environment by running this command:
python -m venv .venv
  • Activate the virtual environment by running the command for your operating system:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

You should see (.venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

  • Install all required Python packages by running:
pip install dbt-core dbt-clickhouse dagster dagster-webserver dagster-dbt psycopg2-binary clickhouse-connect faker

What are these packages?

  • dbt-core and dbt-clickhouse let you write SQL-based data transformations that run against ClickHouse.
  • dagster, dagster-webserver, and dagster-dbt provide pipeline orchestration with a web UI.
  • psycopg2-binary and clickhouse-connect are database drivers for PostgreSQL and ClickHouse.
  • faker generates realistic fake data for your e-commerce dataset.

The installation may take 2-3 minutes. When it finishes, you should see a success message with no errors.

Seeing a compilation error for psycopg2-binary?

On some systems, psycopg2-binary needs a C compiler. If you see errors about missing pg_config or build tools, try upgrading pip first with pip install --upgrade pip and then re-run the install command.

help me fix my pip install error

Create the custom Kafka Connect Dockerfile

Your pipeline needs two Kafka Connect connectors: Debezium to read changes from PostgreSQL, and the ClickHouse sink to write those changes into ClickHouse. You will build a single Docker image that bundles both.

  • In the VS Code Explorer sidebar, right-click the docker folder.
  • Click New File.
  • Name the file Dockerfile.connect and press Enter.
  • Paste the following code into Dockerfile.connect:
# Start from the official Debezium Kafka Connect image
FROM quay.io/debezium/connect:3.5
# Download the ClickHouse Kafka Connect Sink connector JAR
RUN curl -fsSL -o /kafka/connect/clickhouse-kafka-connect.jar https://github.com/ClickHouse/clickhouse-kafka-connect/releases/download/v1.3.7/clickhouse-kafka-connect-v1.3.7.jar

What does this Dockerfile do?

The FROM line uses the official Debezium image, which already includes Debezium's PostgreSQL source connector. This gives you Change Data Capture capabilities out of the box.

The RUN line downloads the ClickHouse sink connector JAR from GitHub and places it in the Kafka Connect plugins directory. This lets the same Kafka Connect worker both read from PostgreSQL and write to ClickHouse.

  • Save docker/Dockerfile.connect by pressing Cmd+S (macOS) or Ctrl+S (Windows).

VS Code warning about Dockerfile extensions?

VS Code might suggest installing a Docker extension. This is optional for this project. The file will work regardless of whether the extension is installed.

help me with my Dockerfile.connect setup

✔️ Awesome, I've got everything!

Great. Double check you have saved docker/Dockerfile.connect and your virtual environment is still active (you see (.venv) in your terminal prompt).

ⓧ I'd like to double check the full code

Here is the complete content of docker/Dockerfile.connect:

FROM quay.io/debezium/connect:3.5
RUN curl -fsSL -o /kafka/connect/clickhouse-kafka-connect.jar https://github.com/ClickHouse/clickhouse-kafka-connect/releases/download/v1.3.7/clickhouse-kafka-connect-v1.3.7.jar

Your project structure is ready and your Python environment has all the tools it needs. Next up, you will create the Docker Compose stack that brings all seven services to life.

Create the Docker Compose Stack

Your project structure and custom Kafka Connect image are ready. Now you need the infrastructure to run everything on.

In this step, you will define a Docker Compose file that spins up 7 interconnected services on your machine. This single file replaces what would normally be weeks of infrastructure provisioning in the cloud.

In this step, get ready to:
  • Define all 7 services (PostgreSQL, Kafka, Kafka Connect, Kafka UI, ClickHouse, Dagster, Grafana) in one Docker Compose file.
  • Configure networking so every service can communicate over a shared Docker network.
  • Launch the stack and verify each service is healthy.
Define the data infrastructure services

Every service in the pipeline needs to talk to every other service. Docker Compose lets you define all of them in a single YAML file and connect them through a shared network.

  • In VS Code, create a new file called docker-compose.yml in the root of your data-pipeline folder (not inside any subfolder).
  • Paste the following code into docker-compose.yml:
version: '3.8'
services:
  # Transactional database - source of truth for e-commerce data
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: ecommerce
    # Enable logical replication for CDC
    command: postgres -c wal_level=logical
    ports:
      - '5432:5432'
    networks:
      - pipeline

  # Event streaming platform in KRaft mode (no ZooKeeper needed)
  kafka:
    image: apache/kafka:3.9.2
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    ports:
      - '9092:9092'
    networks:
      - pipeline

What does this code do?

This defines the two foundational services your pipeline depends on.

  • postgres runs PostgreSQL 16 with wal_level=logical enabled. This setting turns on logical replication, which is what allows Change Data Capture to read every insert, update, and delete from the database.
  • kafka runs Apache Kafka 3.9.2 in KRaft mode. KRaft eliminates the need for a separate ZooKeeper service by letting Kafka manage its own metadata.
  • You will not see a change yet. The next chunk completes the file so you can launch the full stack.
Add the streaming, analytics, and visualization services

The remaining five services handle connector management, stream monitoring, analytical storage, orchestration, and dashboarding.

  • In docker-compose.yml, add the following code directly below the kafka service (after its networks line):
  # Runs Debezium source + ClickHouse sink connectors
  kafka-connect:
    build:
      context: ./docker
      dockerfile: Dockerfile.connect
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_PLUGIN_PATH: /kafka/connect
    ports:
      - '8083:8083'
    depends_on:
      - kafka
      - postgres
    networks:
      - pipeline

  # Web UI for browsing Kafka topics and messages
  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
    ports:
      - '8080:8080'
    depends_on:
      - kafka
    networks:
      - pipeline

  # Columnar OLAP database for fast analytical queries
  clickhouse:
    image: clickhouse/clickhouse-server:24
    environment:
      CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1
    ports:
      - '8123:8123'
      - '9000:9000'
    networks:
      - pipeline

  # Orchestrator for dbt transformations
  dagster:
    image: python:3.10-slim
    working_dir: /app
    volumes:
      - .:/app
    command: dagster-webserver -h 0.0.0.0 -p 3000 -m dagster_project
    ports:
      - '3000:3000'
    networks:
      - pipeline

  # Dashboard and visualization layer
  grafana:
    image: grafana/grafana:11.0.0
    environment:
      GF_INSTALL_PLUGINS: grafana-clickhouse-datasource
    ports:
      - '3001:3000'
    networks:
      - pipeline

# Shared network for all services
networks:
  pipeline:
    driver: bridge

What does this code do?

These five services complete your data platform.

  • kafka-connect builds from your custom Dockerfile (with Debezium + ClickHouse sink). It uses KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS to find Kafka over the internal Docker network, not localhost.
  • kafka-ui gives you a visual browser for topics and messages so you can watch CDC events flow in real time.
  • clickhouse is the columnar OLAP database where all analytical queries will run.
  • dagster runs the Dagster web server for orchestrating dbt transformations.
  • grafana runs Grafana with the ClickHouse data source plugin pre-installed via GF_INSTALL_PLUGINS.

The networks block at the bottom creates a shared bridge network called pipeline. Every service joins this network, so they can reach each other by service name (e.g. kafka:9092).

  • Save docker-compose.yml.

✔️ Awesome, I've got everything!

Great. Double check you have saved docker-compose.yml before moving on.

ⓧ I'd like to double check the full code

version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: ecommerce
    command: postgres -c wal_level=logical
    ports:
      - '5432:5432'
    networks:
      - pipeline

  kafka:
    image: apache/kafka:3.9.2
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    ports:
      - '9092:9092'
    networks:
      - pipeline

  kafka-connect:
    build:
      context: ./docker
      dockerfile: Dockerfile.connect
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_PLUGIN_PATH: /kafka/connect
    ports:
      - '8083:8083'
    depends_on:
      - kafka
      - postgres
    networks:
      - pipeline

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
    ports:
      - '8080:8080'
    depends_on:
      - kafka
    networks:
      - pipeline

  clickhouse:
    image: clickhouse/clickhouse-server:24
    environment:
      CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1
    ports:
      - '8123:8123'
      - '9000:9000'
    networks:
      - pipeline

  dagster:
    image: python:3.10-slim
    working_dir: /app
    volumes:
      - .:/app
    command: dagster-webserver -h 0.0.0.0 -p 3000 -m dagster_project
    ports:
      - '3000:3000'
    networks:
      - pipeline

  grafana:
    image: grafana/grafana:11.0.0
    environment:
      GF_INSTALL_PLUGINS: grafana-clickhouse-datasource
    ports:
      - '3001:3000'
    networks:
      - pipeline

networks:
  pipeline:
    driver: bridge
Launch the stack and verify all services

With the Compose file saved, you can bring up all 7 services in one command. Docker will build the custom Kafka Connect image on the first run, which may take a minute.

  • In your terminal, make sure you are in the data-pipeline folder, then start the stack by running:
docker compose up -d

What does this command do?

docker compose up reads your docker-compose.yml and starts every service defined in it. The -d flag runs containers in detached mode so they stay running in the background while you continue working.

Wait for all containers to finish starting. This may take 1-3 minutes on the first run as Docker pulls images and builds the Kafka Connect image.

  • Verify the ClickHouse HTTP interface is responding by running:
curl http://localhost:8123
  • You should see Ok. printed in your terminal.
  • Open your browser and go to http://localhost:8080 to confirm Kafka UI is running. You should see the Kafka UI dashboard with a cluster named local.
  • Open http://localhost:3001 to confirm Grafana is running. You should see the Grafana login page (default credentials: admin / admin).

Services not starting?

  • Run docker compose ps to check which containers are unhealthy or exited.
  • Port conflicts are the most common issue. If another service is using port 5432, 9092, 8080, or 3001, stop it before retrying.
  • This stack needs 4-6 GB of RAM allocated to Docker. Open Docker Desktop, go to Settings then Resources, and increase the memory limit if containers keep restarting.
  • help me debug my Docker Compose services

Your entire data platform is now running locally. Next up, you will generate a massive e-commerce dataset and load it into PostgreSQL.

Generate and Load the E-Commerce Dataset

Your Docker services are running, but the PostgreSQL database is empty. For the CDC pipeline to capture real changes, it needs a substantial dataset to work with.

In this step, you will create a Python script that generates a realistic e-commerce dataset with 500,000+ rows. The script uses the Faker library to produce realistic names, emails, and dates, then bulk-loads everything into PostgreSQL using execute_values for performance.

In this step, get ready to:
  • Create a Python script that generates 500K+ rows of e-commerce data.
  • Load customers, products, and orders into PostgreSQL with bulk inserts.
  • Verify the data loaded correctly with SQL queries.
Create the data generation script

The data generator needs to define three related tables: customers, products, and orders. It connects to PostgreSQL, creates the table schemas with primary keys and foreign keys, then prepares to fill them with data.

  • Create a new file called generate_data.py inside your scripts/ directory.
  • Add the imports, database connection, and table creation code by pasting the following:
import psycopg2
from psycopg2.extras import execute_values
from faker import Faker
import random
from datetime import datetime, timedelta

fake = Faker()

conn = psycopg2.connect(
    host='localhost', port=5432,
    user='postgres', password='postgres', dbname='ecommerce'
)
cur = conn.cursor()

# Create tables
cur.execute('''
    CREATE TABLE IF NOT EXISTS customers (
        id SERIAL PRIMARY KEY,
        name VARCHAR(255),
        email VARCHAR(255) UNIQUE,
        country VARCHAR(100),
        created_at TIMESTAMP DEFAULT NOW()
    );
    CREATE TABLE IF NOT EXISTS products (
        id SERIAL PRIMARY KEY,
        name VARCHAR(255),
        category VARCHAR(100),
        price NUMERIC(10,2),
        created_at TIMESTAMP DEFAULT NOW()
    );
    CREATE TABLE IF NOT EXISTS orders (
        id SERIAL PRIMARY KEY,
        customer_id INTEGER REFERENCES customers(id),
        product_id INTEGER REFERENCES products(id),
        quantity INTEGER,
        total_amount NUMERIC(10,2),
        status VARCHAR(50),
        order_date TIMESTAMP
    );
''')
conn.commit()

What does this code do?

This first section sets up the foundation for the data generator.

  • psycopg2 is the PostgreSQL adapter for Python. execute_values is a fast bulk-insert helper that sends many rows in a single query instead of one at a time.
  • The connection uses the same credentials you configured in docker-compose.yml (user postgres, database ecommerce).
  • The three tables are linked by foreign keys: each order references a customer_id and a product_id. This mirrors a real e-commerce schema.

You will not see any output yet. The data generation logic in the next chunk completes this script.

Add the data generation logic

With the table schemas defined, the script needs code to generate realistic data and insert it. The orders table is the largest (500,000 rows), so it uses a batched approach to avoid memory issues.

  • Below the conn.commit() line you just added, paste the customer and product generation code:
# Generate customers
print('Inserting 10,000 customers...')
customer_data = [
    (fake.name(), fake.unique.email(), fake.country(), fake.date_time_this_decade())
    for _ in range(10000)
]
execute_values(cur, 'INSERT INTO customers (name, email, country, created_at) VALUES %s', customer_data)
conn.commit()

# Generate products
print('Inserting 1,000 products...')
categories = ['Electronics', 'Clothing', 'Books', 'Home', 'Sports', 'Toys', 'Food']
product_data = [
    (fake.word().capitalize() + ' ' + fake.word().capitalize(),
     random.choice(categories),
     round(random.uniform(5, 500), 2),
     fake.date_time_this_decade())
    for _ in range(1000)
]
execute_values(cur, 'INSERT INTO products (name, category, price, created_at) VALUES %s', product_data)
conn.commit()

What does this code do?

  • Customers: generates 10,000 rows with realistic names, unique emails, random countries, and timestamps from this decade.
  • Products: generates 1,000 rows with two-word product names, a random category from the categories list, a random price between $5 and $500, and a timestamp.
  • Both use list comprehensions to build all rows in memory, then execute_values sends them in one efficient batch.
  • Now add the order generation code below the products section. This is the largest table, so it inserts in batches of 10,000:
# Generate orders in batches
print('Inserting 500,000 orders...')
statuses = ['pending', 'processing', 'shipped', 'delivered', 'cancelled']
batch_size = 10000
for batch_num in range(50):
    order_data = []
    for _ in range(batch_size):
        qty = random.randint(1, 10)
        price = round(random.uniform(5, 500), 2)
        order_data.append((
            random.randint(1, 10000),
            random.randint(1, 1000),
            qty,
            round(qty * price, 2),
            random.choice(statuses),
            fake.date_time_between(start_date='-2y', end_date='now')
        ))
    execute_values(cur,
        'INSERT INTO orders (customer_id, product_id, quantity, total_amount, status, order_date) VALUES %s',
        order_data)
    conn.commit()
    print(f'  Inserted {(batch_num+1)*batch_size:,} / 500,000 orders')

cur.close()
conn.close()
print('Done!')

Why insert in batches?

Building 500,000 rows in memory at once would consume a lot of RAM. Instead, the loop generates 10,000 rows at a time across 50 iterations.

Each batch commits separately, so you get a progress indicator showing how far along the insert is. The random.randint(1, 10000) and random.randint(1, 1000) calls pick valid customer and product IDs that already exist in the database.

  • Save scripts/generate_data.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file before running it in the next substep.

ⓧ I'd like to double check the full code

import psycopg2
from psycopg2.extras import execute_values
from faker import Faker
import random
from datetime import datetime, timedelta

fake = Faker()

conn = psycopg2.connect(
    host='localhost', port=5432,
    user='postgres', password='postgres', dbname='ecommerce'
)
cur = conn.cursor()

# Create tables
cur.execute('''
    CREATE TABLE IF NOT EXISTS customers (
        id SERIAL PRIMARY KEY,
        name VARCHAR(255),
        email VARCHAR(255) UNIQUE,
        country VARCHAR(100),
        created_at TIMESTAMP DEFAULT NOW()
    );
    CREATE TABLE IF NOT EXISTS products (
        id SERIAL PRIMARY KEY,
        name VARCHAR(255),
        category VARCHAR(100),
        price NUMERIC(10,2),
        created_at TIMESTAMP DEFAULT NOW()
    );
    CREATE TABLE IF NOT EXISTS orders (
        id SERIAL PRIMARY KEY,
        customer_id INTEGER REFERENCES customers(id),
        product_id INTEGER REFERENCES products(id),
        quantity INTEGER,
        total_amount NUMERIC(10,2),
        status VARCHAR(50),
        order_date TIMESTAMP
    );
''')
conn.commit()

# Generate customers
print('Inserting 10,000 customers...')
customer_data = [
    (fake.name(), fake.unique.email(), fake.country(), fake.date_time_this_decade())
    for _ in range(10000)
]
execute_values(cur, 'INSERT INTO customers (name, email, country, created_at) VALUES %s', customer_data)
conn.commit()

# Generate products
print('Inserting 1,000 products...')
categories = ['Electronics', 'Clothing', 'Books', 'Home', 'Sports', 'Toys', 'Food']
product_data = [
    (fake.word().capitalize() + ' ' + fake.word().capitalize(),
     random.choice(categories),
     round(random.uniform(5, 500), 2),
     fake.date_time_this_decade())
    for _ in range(1000)
]
execute_values(cur, 'INSERT INTO products (name, category, price, created_at) VALUES %s', product_data)
conn.commit()

# Generate orders in batches
print('Inserting 500,000 orders...')
statuses = ['pending', 'processing', 'shipped', 'delivered', 'cancelled']
batch_size = 10000
for batch_num in range(50):
    order_data = []
    for _ in range(batch_size):
        qty = random.randint(1, 10)
        price = round(random.uniform(5, 500), 2)
        order_data.append((
            random.randint(1, 10000),
            random.randint(1, 1000),
            qty,
            round(qty * price, 2),
            random.choice(statuses),
            fake.date_time_between(start_date='-2y', end_date='now')
        ))
    execute_values(cur,
        'INSERT INTO orders (customer_id, product_id, quantity, total_amount, status, order_date) VALUES %s',
        order_data)
    conn.commit()
    print(f'  Inserted {(batch_num+1)*batch_size:,} / 500,000 orders')

cur.close()
conn.close()
print('Done!')
Run the script and verify the data

The script is ready. Running it connects to the PostgreSQL container, creates the tables, and fills them with generated data. This will take a minute or two depending on your machine.

  • Make sure your virtual environment is activated and your Docker containers are running.
  • Run the data generation script by executing this command:
python scripts/generate_data.py

What should I see?

The script prints progress as it inserts data. You should see output like this:

Inserting 10,000 customers...
Inserting 1,000 products...
Inserting 500,000 orders...
  Inserted 10,000 / 500,000 orders
  Inserted 20,000 / 500,000 orders
  ...
  Inserted 500,000 / 500,000 orders
Done!
  • Verify the data loaded correctly by running this command to check the orders table row count:
docker exec -it postgres psql -U postgres -d ecommerce -c "SELECT COUNT(*) FROM orders;"
  • You should see 500000 in the output.

Check all three tables

You can also verify customers and products by changing orders to customers or products in the query above. You should see 10,000 and 1,000 respectively.

Seeing a connection error?

  • Make sure the PostgreSQL container is running. Check with docker ps and look for the postgres container.
  • Confirm port 5432 is not already in use by another PostgreSQL instance on your machine.
  • Verify your virtual environment is activated. You should see (.venv) at the start of your terminal prompt.

Help me debug my PostgreSQL connection error in the data generation script.

Your database is loaded with over half a million rows of realistic e-commerce data. Next up, you will configure Debezium to capture every change in this database and stream it through Kafka in real time.

Configure the Debezium Source Connector

Your PostgreSQL database is loaded with 500K+ rows of e-commerce data. Now you need a way to capture every change in that database and stream it into Kafka automatically.

This is where Change Data Capture (CDC) comes in. Debezium is a CDC tool that reads PostgreSQL's write-ahead log (WAL) and publishes every insert, update, and delete as an event to Kafka. In this step, you will configure and register a Debezium source connector so your data starts flowing.

In this step, get ready to:
  • Create a Debezium connector configuration file.
  • Register the connector with Kafka Connect's REST API.
  • Verify CDC topics and messages are flowing in Kafka UI.
Create the connector configuration file

Kafka Connect uses JSON configuration files to define how a connector should behave. Each file specifies which connector class to use, where to find the source database, and how to name the output topics.

  • Create a new file called source-connector.json inside your scripts/ directory by pasting the following configuration:
{
  "name": "ecommerce-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname": "ecommerce",
    "topic.prefix": "ecommerce",
    "plugin.name": "pgoutput",
    "schema.include.list": "public",
    "table.include.list": "public.customers,public.products,public.orders",
    "snapshot.mode": "initial"
  }
}

What does this configuration do?

  • connector.class tells Kafka Connect to use the Debezium PostgreSQL connector.
  • database.* fields point to the PostgreSQL container running in your Docker network.
  • topic.prefix sets ecommerce as the prefix for all topic names. Debezium creates topics using the pattern prefix.schema.table.
  • plugin.name uses pgoutput, the built-in logical replication plugin for PostgreSQL 10+.
  • snapshot.mode set to initial means Debezium will take a full snapshot of existing data before switching to streaming mode.
  • Save scripts/source-connector.json.

Why pgoutput instead of a plugin like decoderbufs?

PostgreSQL 10 and later includes the pgoutput logical replication plugin natively. You do not need to install any extensions. Older approaches required external plugins like decoderbufs or wal2json, but pgoutput is now the recommended default.

Register the connector with Kafka Connect

Kafka Connect exposes a REST API on port 8083 for managing connectors. You send your JSON configuration as a POST request, and Kafka Connect starts the connector immediately.

  • Register your Debezium source connector by running this command:
curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d @scripts/source-connector.json

Windows users

If you are using PowerShell, curl is an alias for Invoke-WebRequest. Use curl.exe instead to call the real curl binary, or run the command from Git Bash.

  • Verify the connector is running by checking its status:
curl http://localhost:8083/connectors/ecommerce-source/status

You should see a JSON response with "state": "RUNNING" for both the connector and its task.

Connector showing FAILED state?

  • Make sure all your Docker containers are still running. Run docker compose ps and confirm the kafka-connect service is healthy.
  • Kafka Connect takes 30-60 seconds after startup to become fully ready. If you registered the connector too soon, delete it and re-register.
  • Check the database.hostname in your JSON is postgres (the Docker service name), not localhost.

Help me troubleshoot my Debezium connector showing a FAILED state

Verify topics and CDC events in Kafka UI

Once the connector is running, Debezium takes a snapshot of all existing data in your tables. Each row becomes a CDC event published to a Kafka topic.

  • Open Kafka UI in your browser at http://localhost:8080.
  • Click Topics in the left sidebar.
  • Confirm you see three data topics: ecommerce.public.customers, ecommerce.public.products, and ecommerce.public.orders.
  • Click on ecommerce.public.orders and then select the Messages tab to browse individual CDC events.

What do CDC events look like?

Each message in the topic represents a single database row change. The event payload has a before field (the old state, null for inserts) and an after field (the new state of the row). This structure lets downstream consumers know exactly what changed.

During the initial snapshot, all events are inserts so the before field will be null and the after field contains the full row data.

Topics not appearing in Kafka UI?

  • The initial snapshot of 500K+ rows can take 2-5 minutes. Wait and refresh the page.
  • Check that kafka-ui is connected to the right broker. In your docker-compose.yml, KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS should be kafka:9092.
  • Verify the connector status is RUNNING using the curl command from the previous substep.

Help me figure out why my Kafka topics are not showing up

Your CDC pipeline is live. Every change to your PostgreSQL tables is now being captured and streamed into Kafka automatically. Next up, you will land that data into ClickHouse so you can run fast analytical queries on it.

Create ClickHouse Tables and Configure the Sink Connector

Your CDC events are flowing through Kafka, but they need a destination. ClickHouse is a columnar OLAP database built for fast analytical queries over massive datasets.

In this step, you will create landing tables in ClickHouse to receive the CDC data, then configure the ClickHouse Kafka Connect Sink connector to automatically write Kafka messages into those tables.

In this step, get ready to:
  • Create target tables in ClickHouse for the CDC data.
  • Configure and register the ClickHouse sink connector.
  • Verify data is flowing from Kafka into ClickHouse.
Create the ClickHouse target tables

The ClickHouse sink connector writes into existing tables. You need to create these tables before registering the connector.

  • Create the raw_customers table by sending this SQL to the ClickHouse HTTP interface:
curl http://localhost:8123 --data "CREATE TABLE IF NOT EXISTS raw_customers (id Int64, name String, email String, country String, created_at DateTime, __deleted UInt8, __op String) ENGINE = MergeTree ORDER BY id"
  • Create the raw_products table:
curl http://localhost:8123 --data "CREATE TABLE IF NOT EXISTS raw_products (id Int64, name String, category String, price Decimal(10,2), created_at DateTime, __deleted UInt8, __op String) ENGINE = MergeTree ORDER BY id"
  • Create the raw_orders table:
curl http://localhost:8123 --data "CREATE TABLE IF NOT EXISTS raw_orders (id Int64, customer_id Int64, product_id Int64, quantity Int32, total_amount Decimal(10,2), status String, order_date DateTime, __deleted UInt8, __op String) ENGINE = MergeTree ORDER BY id"

What do these tables do?

Each table uses the MergeTree engine, which is ClickHouse's default storage engine optimized for fast inserts and analytical queries. The ORDER BY id clause defines how data is sorted on disk for efficient lookups.

The __deleted and __op columns are CDC metadata. They track whether a row was deleted and what operation produced it (create, update, or delete).

  • Verify the tables were created by running:
curl http://localhost:8123 --data "SHOW TABLES"

You should see raw_customers, raw_products, and raw_orders in the output.

Seeing 'Connection refused'?

Make sure your Docker containers are still running with docker compose ps. The ClickHouse container must be healthy before it accepts HTTP requests on port 8123.

help me debug why curl to localhost:8123 is being refused

Configure and register the sink connector

The ClickHouse sink connector reads CDC events from Kafka topics and writes them into the tables you just created. Since the Debezium topic names (like ecommerce.public.orders) do not match the ClickHouse table names (like raw_orders), you need a topic2TableMap to bridge them.

  • Create the file scripts/sink-connector.json with the following connector configuration:
{
  "name": "ecommerce-sink",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "hostname": "clickhouse",
    "port": "8123",
    "database": "default",
    "topics": "ecommerce.public.customers,ecommerce.public.products,ecommerce.public.orders",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "topic2TableMap": "ecommerce.public.customers=raw_customers,ecommerce.public.products=raw_products,ecommerce.public.orders=raw_orders"
  }
}

What does this configuration do?

  • connector.class specifies the official ClickHouse Kafka Connect Sink connector.
  • hostname is set to clickhouse because that is the Docker service name, reachable from within the Docker network.
  • port is 8123, the ClickHouse HTTP interface.
  • topic2TableMap maps each Debezium topic to its corresponding ClickHouse table name.
  • value.converter.schemas.enable is false because the worker-level converter already handles schema extraction.
  • Register the sink connector by posting it to the Kafka Connect REST API:
curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d @scripts/sink-connector.json
  • Verify the connector is running:
curl http://localhost:8083/connectors/ecommerce-sink/status

You should see a JSON response with "state":"RUNNING" for both the connector and its task.

Connector showing FAILED?

Check that all three ClickHouse tables exist. The sink connector will fail if it cannot find the target table. Re-run the SHOW TABLES query to confirm.

Also verify that the topic names in your JSON match exactly what Debezium created. Open Kafka UI at http://localhost:8080 and compare topic names character by character.

help me debug why my ClickHouse sink connector is failing

Verify data is flowing into ClickHouse

The sink connector will replay the initial snapshot that Debezium captured. This means all 500,000+ orders will flow from Kafka into ClickHouse over the next few minutes.

  • Check the row count in raw_orders by running:
curl http://localhost:8123 --data "SELECT count() FROM raw_orders"

The count will start at 0 and climb as the connector processes messages. Wait 2-5 minutes for the full snapshot to complete.

  • Run the query again after a few minutes. You should see approximately 500,000 rows.
  • Also check customers and products:
curl http://localhost:8123 --data "SELECT count() FROM raw_customers"
curl http://localhost:8123 --data "SELECT count() FROM raw_products"

You should see approximately 10,000 customers and 1,000 products.

How can I monitor progress?

Open Kafka UI at http://localhost:8080 and look at consumer lag for the ecommerce-sink consumer group. When lag reaches 0, the connector has caught up with all available messages.

You can also check Kafka Connect status periodically with the /connectors/ecommerce-sink/status endpoint to confirm the task is still running without errors.

Count stuck at 0 after several minutes?

Verify the connector is still RUNNING (not FAILED) with the status endpoint. If it has failed, check the Kafka Connect logs with docker compose logs kafka-connect for error details.

Also confirm that Kafka topics have messages by checking the topic message count in Kafka UI. If topics are empty, the source connector may have stopped.

help me debug why no data is flowing into my ClickHouse raw_orders table

✔️ Awesome, I've got everything!

Great. Double check you have saved scripts/sink-connector.json and that your row counts are climbing in ClickHouse.

ⓧ I'd like to double check the full code

Here is the complete scripts/sink-connector.json file:

{
  "name": "ecommerce-sink",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "hostname": "clickhouse",
    "port": "8123",
    "database": "default",
    "topics": "ecommerce.public.customers,ecommerce.public.products,ecommerce.public.orders",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "topic2TableMap": "ecommerce.public.customers=raw_customers,ecommerce.public.products=raw_products,ecommerce.public.orders=raw_orders"
  }
}

Your CDC pipeline is now end-to-end. Data flows from PostgreSQL through Kafka and lands in ClickHouse automatically. Next up, you will build dbt models to transform this raw data into analytical tables ready for dashboards.

Build dbt Models for Analytics

Your raw data is now sitting in ClickHouse, streamed directly from PostgreSQL through Kafka.

Raw CDC data is messy. It includes metadata columns, deleted records, and no business logic. In this step, you will build a layered dbt project that transforms this raw data into clean, aggregated tables ready for dashboards.

In this step, get ready to:
  • Initialize a dbt project with ClickHouse connection settings.
  • Create staging, intermediate, and mart SQL models.
  • Run dbt to materialize all models and verify the results in ClickHouse.
Initialize the dbt project configuration

A dbt project needs two configuration files to know what to build and where to connect. The project file defines metadata and model paths. The profiles file tells dbt how to reach your ClickHouse database.

  • In your terminal, make sure you are inside the dbt_project/ directory by running:
cd dbt_project
  • Create the model subdirectories by running:
mkdir -p models/staging models/intermediate models/marts
  • Confirm the directories exist by running:
ls models
  • You should see intermediate, marts, and staging listed.
  • Create the project configuration file by pasting this into dbt_project/dbt_project.yml:
name: ecommerce_analytics
version: '1.0.0'
config-version: 2
profile: ecommerce_analytics
model-paths: ['models']
models:
  ecommerce_analytics:
    staging:
      +schema: analytics
    intermediate:
      +schema: analytics
    marts:
      +schema: analytics

What does this configuration do?

The project file tells dbt your project is called ecommerce_analytics and that all SQL models live in the models/ directory.

The +schema: analytics setting tells dbt to create all tables in a ClickHouse database called analytics, keeping them separate from your raw data in the default database.

  • Create the connection profile by pasting this into dbt_project/profiles.yml:
ecommerce_analytics:
  target: dev
  outputs:
    dev:
      type: clickhouse
      schema: analytics
      host: localhost
      port: 8123
      user: default
      password: ''
      secure: False

What does the profiles file do?

This file tells dbt how to connect to ClickHouse. It uses the HTTP interface on port 8123 with the default user and no password (matching your local Docker setup).

  • Save both dbt_project.yml and profiles.yml.

Not sure where to create these files?

Both files should be at the top level of your dbt_project/ directory. In VS Code, right-click the dbt_project folder in the Explorer sidebar, select New File, and type the filename.

help me create dbt configuration files in the right location

Create the staging, intermediate, and mart models

dbt models follow a layered architecture. Staging models clean up the raw data. Intermediate models join tables together. Mart models aggregate results for dashboards.

Why three layers?

Each layer has a single job. Staging handles data cleaning (filtering deleted rows, casting types). Intermediate handles enrichment (joining orders with customer and product details). Marts handle business questions (revenue by day, top customers).

This separation makes your pipeline easier to debug. If a dashboard shows wrong numbers, you can check each layer independently to find where things went wrong.

  • Create the first staging model by pasting this into dbt_project/models/staging/stg_customers.sql:
SELECT
    id,
    name,
    email,
    country,
    created_at
FROM default.raw_customers
WHERE __deleted = 0

What does this model do?

This query reads from the raw_customers table that the CDC pipeline populated. The WHERE __deleted = 0 filter removes any rows that Debezium marked as deleted. This gives you only the currently active customer records.

  • Create the products staging model by pasting this into dbt_project/models/staging/stg_products.sql:
SELECT
    id,
    name,
    category,
    price,
    created_at
FROM default.raw_products
WHERE __deleted = 0
  • Create the orders staging model by pasting this into dbt_project/models/staging/stg_orders.sql:
SELECT
    id,
    customer_id,
    product_id,
    quantity,
    total_amount,
    status,
    order_date
FROM default.raw_orders
WHERE __deleted = 0
  • Save all three staging model files.
  • Now create the intermediate model that joins all three staging tables together. Paste this into dbt_project/models/intermediate/int_orders_enriched.sql:
SELECT
    o.id AS order_id,
    o.customer_id,
    c.name AS customer_name,
    c.country AS customer_country,
    o.product_id,
    p.name AS product_name,
    p.category AS product_category,
    p.price AS unit_price,
    o.quantity,
    o.total_amount,
    o.status,
    o.order_date
FROM {{ ref('stg_orders') }} o
LEFT JOIN {{ ref('stg_customers') }} c ON o.customer_id = c.id
LEFT JOIN {{ ref('stg_products') }} p ON o.product_id = p.id

What does this model do?

This model joins every order with the customer who placed it and the product they bought. The {{ ref('stg_orders') }} syntax is dbt's way of referencing other models. dbt uses these references to figure out the correct execution order.

The result is a single denormalized table with all the context you need for analytics: customer names, countries, product categories, and order amounts in one place.

  • Save int_orders_enriched.sql.
  • Create the daily revenue mart by pasting this into dbt_project/models/marts/mart_revenue_by_day.sql:
SELECT
    toDate(order_date) AS order_day,
    count() AS total_orders,
    sum(total_amount) AS total_revenue,
    avg(total_amount) AS avg_order_value
FROM {{ ref('int_orders_enriched') }}
WHERE status != 'cancelled'
GROUP BY order_day
ORDER BY order_day

What does this mart do?

This aggregates revenue by day, filtering out cancelled orders. The toDate() function is a ClickHouse function that truncates timestamps to date-level precision. This table will power the time-series chart in Grafana.

  • Create the top customers mart by pasting this into dbt_project/models/marts/mart_top_customers.sql:
SELECT
    customer_id,
    customer_name,
    customer_country,
    count() AS total_orders,
    sum(total_amount) AS total_spend
FROM {{ ref('int_orders_enriched') }}
WHERE status != 'cancelled'
GROUP BY customer_id, customer_name, customer_country
ORDER BY total_spend DESC
LIMIT 100
  • Create the category performance mart by pasting this into dbt_project/models/marts/mart_category_performance.sql:
SELECT
    product_category,
    count() AS total_orders,
    sum(total_amount) AS total_revenue,
    avg(total_amount) AS avg_order_value
FROM {{ ref('int_orders_enriched') }}
WHERE status != 'cancelled'
GROUP BY product_category
ORDER BY total_revenue DESC

What do these two marts do?

The top customers mart ranks the top 100 customers by total spend. The category performance mart shows revenue broken down by product category. Both filter out cancelled orders so the numbers reflect real business activity.

  • Save all three mart files.

File structure looking wrong?

Your dbt_project/ folder should now contain: dbt_project.yml, profiles.yml, and a models/ folder with three subfolders (staging/, intermediate/, marts/). Each subfolder contains the .sql files you just created.

help me check my dbt project file structure

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files before moving on.

ⓧ I'd like to double check the full code

Here is the complete file content for each model. Compare your files against these:

name: ecommerce_analytics
version: '1.0.0'
config-version: 2
profile: ecommerce_analytics
model-paths: ['models']
models:
  ecommerce_analytics:
    staging:
      +schema: analytics
    intermediate:
      +schema: analytics
    marts:
      +schema: analytics
ecommerce_analytics:
  target: dev
  outputs:
    dev:
      type: clickhouse
      schema: analytics
      host: localhost
      port: 8123
      user: default
      password: ''
      secure: False
SELECT
    id,
    name,
    email,
    country,
    created_at
FROM default.raw_customers
WHERE __deleted = 0
SELECT
    id,
    name,
    category,
    price,
    created_at
FROM default.raw_products
WHERE __deleted = 0
SELECT
    id,
    customer_id,
    product_id,
    quantity,
    total_amount,
    status,
    order_date
FROM default.raw_orders
WHERE __deleted = 0
SELECT
    o.id AS order_id,
    o.customer_id,
    c.name AS customer_name,
    c.country AS customer_country,
    o.product_id,
    p.name AS product_name,
    p.category AS product_category,
    p.price AS unit_price,
    o.quantity,
    o.total_amount,
    o.status,
    o.order_date
FROM {{ ref('stg_orders') }} o
LEFT JOIN {{ ref('stg_customers') }} c ON o.customer_id = c.id
LEFT JOIN {{ ref('stg_products') }} p ON o.product_id = p.id
SELECT
    toDate(order_date) AS order_day,
    count() AS total_orders,
    sum(total_amount) AS total_revenue,
    avg(total_amount) AS avg_order_value
FROM {{ ref('int_orders_enriched') }}
WHERE status != 'cancelled'
GROUP BY order_day
ORDER BY order_day
SELECT
    customer_id,
    customer_name,
    customer_country,
    count() AS total_orders,
    sum(total_amount) AS total_spend
FROM {{ ref('int_orders_enriched') }}
WHERE status != 'cancelled'
GROUP BY customer_id, customer_name, customer_country
ORDER BY total_spend DESC
LIMIT 100
SELECT
    product_category,
    count() AS total_orders,
    sum(total_amount) AS total_revenue,
    avg(total_amount) AS avg_order_value
FROM {{ ref('int_orders_enriched') }}
WHERE status != 'cancelled'
GROUP BY product_category
ORDER BY total_revenue DESC
Run dbt to build and verify the models

With all your SQL files in place, dbt can now compile and execute them against ClickHouse. You will test the connection first, then run all models.

  • Make sure your terminal is inside the dbt_project/ directory and your virtual environment is active.
  • Verify your connection to ClickHouse by running:
dbt debug

What should I see?

You should see a series of checks ending with "All checks passed!" at the bottom. This confirms dbt can find your project files and connect to ClickHouse on port 8123.

Seeing a connection error?

Make sure your Docker containers are still running. Run docker compose ps from the project root to check. The ClickHouse container must be healthy.

Also confirm that profiles.yml uses host: localhost and port: 8123. The dbt-clickhouse adapter connects via the HTTP interface, not the native port 9000.

help me debug my dbt connection to ClickHouse

  • Execute all models by running:
dbt run

What should I see?

dbt will process your models in dependency order: staging first, then intermediate, then marts. Each model should show a green "OK" status. At the end, you should see all 7 models completed successfully.

  • Verify the mart tables exist in ClickHouse by running this query:
curl http://localhost:8123 --data "SELECT name FROM system.tables WHERE database = 'analytics'"
  • You should see your 7 tables listed: the three staging models, the intermediate model, and the three marts.
  • Optionally, run dbt tests to validate the data:
dbt test

Models failing with table not found?

If your staging models fail because raw_customers, raw_products, or raw_orders don't exist, the ClickHouse sink connector may not have finished loading data. Go back to Kafka UI at http://localhost:8080 and check consumer lag for the sink connector.

If models fail because of the __deleted column, make sure your ClickHouse tables were created with that column. The sink connector adds it automatically when writing CDC data.

help me fix dbt run failures against ClickHouse

Your raw CDC data is now transformed into clean, business-ready tables. Next up, you will orchestrate these dbt runs with Dagster so they execute automatically as part of a managed pipeline.

Set Up Dagster to Orchestrate dbt

Your dbt models transform raw CDC data into clean analytics tables. Running dbt run manually works, but in a production pipeline you need something that schedules, monitors, and tracks dependencies between models.

Dagster is a data orchestrator that treats every data artifact as an "asset" with lineage and observability built in. In this step, you will wire your dbt project into Dagster so every model appears as a materializable asset in a visual graph.

In this step, get ready to:
  • Create the Dagster project structure with resource and asset definitions.
  • Define all dbt models as Dagster assets using the @dbt_assets decorator.
  • Launch the Dagster UI and trigger a full materialization of the asset graph.
Create the Dagster project files and configure resources

Dagster needs a Python module with specific files that tell it where your dbt project lives and how to run it. The DbtProject class points Dagster at your dbt directory, and DbtCliResource gives it the ability to invoke dbt commands.

  • Inside your dagster_project/ directory, create four files by running these commands:
touch dagster_project/__init__.py dagster_project/resources.py dagster_project/assets.py dagster_project/definitions.py

Why four files?

Dagster loads your code as a Python module. The __init__.py makes the folder importable. The other three files separate concerns: resources (connections), assets (data artifacts), and definitions (the entrypoint that wires everything together).

  • Confirm the files exist by running:
ls dagster_project/

You should see __init__.py, assets.py, definitions.py, and resources.py in the output.

  • Open dagster_project/resources.py in VS Code and paste this code:
from dagster_dbt import DbtCliResource, DbtProject
import os

# Point to the dbt project directory (one level up, then into dbt_project/)
DBT_PROJECT_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'dbt_project')

# Create a DbtProject instance for manifest resolution
dbt_project = DbtProject(project_dir=DBT_PROJECT_DIR)

# Create the resource Dagster will inject into assets
dbt_resource = DbtCliResource(project_dir=dbt_project)

What does this code do?

DBT_PROJECT_DIR builds an absolute path from the current file up one directory, then into dbt_project/. This keeps the path correct regardless of where you run Dagster from.

DbtProject represents your dbt project location and helps Dagster find the manifest.json that describes all models. DbtCliResource is the actual resource Dagster injects into your assets so they can call dbt commands.

  • Save resources.py.

Seeing an import error?

Make sure your virtual environment is activated. You should see (.venv) at the start of your terminal prompt. If not, run .venv\Scripts\activate (Windows) or source .venv/bin/activate (macOS/Linux).

help me fix a dagster-dbt import error in my resources.py file

Define dbt models as Dagster assets

Dagster's @dbt_assets decorator reads your dbt project's manifest and automatically registers every model as a trackable asset. This gives you lineage, observability, and the ability to trigger runs from the UI.

  • Before creating the assets file, confirm your dbt manifest exists by running:
ls dbt_project/target/manifest.json

What is the manifest?

When you ran dbt run in the previous step, dbt generated a manifest.json file inside dbt_project/target/. This file contains a complete map of all your models, their dependencies, and their configurations. Dagster reads it to build its asset graph.

  • Open dagster_project/assets.py and paste this code:
from dagster_dbt import dbt_assets, DbtProject
from .resources import dbt_project
import os

# Build the path to the dbt manifest generated by dbt run
MANIFEST_PATH = os.path.join(dbt_project.project_dir, 'target', 'manifest.json')

# Register all dbt models as Dagster assets
@dbt_assets(manifest=MANIFEST_PATH)
def ecommerce_dbt_assets(context, dbt: DbtCliResource):
    yield from dbt.cli(['run'], context=context).stream()

What does this code do?

The @dbt_assets decorator reads the manifest and creates one Dagster asset for each dbt model. When Dagster materializes these assets, it calls dbt.cli(['run']) and streams the output back to the Dagster UI in real time.

The dbt parameter is automatically injected by Dagster based on the resource you will wire up in definitions.py next.

  • Save assets.py.
  • Now open dagster_project/definitions.py and paste this code:
from dagster import Definitions
from .assets import ecommerce_dbt_assets
from .resources import dbt_resource

# Wire assets and resources into Dagster's entrypoint
defs = Definitions(
    assets=[ecommerce_dbt_assets],
    resources={'dbt': dbt_resource}
)

What does this code do?

The Definitions object is Dagster's top-level entrypoint. It registers your assets (the dbt models) and the resources they depend on (the DbtCliResource mapped to the key 'dbt').

The resource key 'dbt' matches the parameter name dbt in your ecommerce_dbt_assets function. Dagster uses this convention to inject the right resource automatically.

  • Save definitions.py.

Seeing a ModuleNotFoundError?

Make sure __init__.py exists inside dagster_project/. Without it, Python cannot treat the folder as an importable module. The file can be empty.

help me debug a ModuleNotFoundError in my Dagster project

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files before moving on.

ⓧ I'd like to double check the full code

Here is the complete code for each Dagster project file.

from dagster_dbt import DbtCliResource, DbtProject
import os

DBT_PROJECT_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'dbt_project')

dbt_project = DbtProject(project_dir=DBT_PROJECT_DIR)

dbt_resource = DbtCliResource(project_dir=dbt_project)
from dagster_dbt import dbt_assets, DbtProject
from .resources import dbt_project
import os

MANIFEST_PATH = os.path.join(dbt_project.project_dir, 'target', 'manifest.json')

@dbt_assets(manifest=MANIFEST_PATH)
def ecommerce_dbt_assets(context, dbt: DbtCliResource):
    yield from dbt.cli(['run'], context=context).stream()
from dagster import Definitions
from .assets import ecommerce_dbt_assets
from .resources import dbt_resource

defs = Definitions(
    assets=[ecommerce_dbt_assets],
    resources={'dbt': dbt_resource}
)
Launch Dagster and materialize the asset graph

With all the files in place, you can start Dagster's development server. It will load your module, discover the dbt assets, and serve a web UI where you can visualize and trigger runs.

  • Launch the Dagster development server by running this command:
dagster dev -m dagster_project

What does this command do?

dagster dev starts a local development server that includes the webserver UI, a daemon for schedules, and a code server. The -m dagster_project flag tells Dagster to load the dagster_project Python module as a code location.

  • Open your browser and navigate to http://localhost:3000.
  • Click Assets in the top navigation bar.

You should see all seven of your dbt models displayed as nodes in an asset graph. The lineage shows staging models feeding into the intermediate model, which feeds into the three mart models.

  • Click Materialize all in the top right corner of the asset graph.
  • Wait for the run to complete. You will see each asset turn green as it materializes successfully.

What just happened?

Dagster executed dbt run behind the scenes, streaming the output into its run log. Each dbt model ran in dependency order: staging first, then intermediate, then marts. Dagster recorded the run duration, status, and metadata for each asset.

Dagster UI not loading?

Check that nothing else is using port 3000. If your Docker Compose stack is running the dagster service container, it may be occupying that port. Stop it with docker compose stop dagster before running dagster dev locally.

Also confirm you are running the command from the data-pipeline project root (the folder that contains both dagster_project/ and dbt_project/).

help me troubleshoot why the Dagster UI is not loading at localhost:3000

Your dbt models are now fully orchestrated. Dagster tracks every materialization, shows lineage between assets, and can schedule runs on a cron. Next up, you will connect Grafana to ClickHouse and build dashboards that visualize the analytics your pipeline produces.

Configure Grafana Dashboards

Your data has travelled from PostgreSQL through Kafka into ClickHouse, been transformed by dbt, and orchestrated by Dagster. The final piece is making it visible to humans.

In this step, you will connect Grafana to ClickHouse and build a dashboard with three panels that visualize your e-commerce analytics in real time.

In this step, get ready to:
  • Connect Grafana to ClickHouse as a data source.
  • Create a dashboard with time series, bar chart, and pie chart panels.
  • Configure auto-refresh to keep the dashboard up to date.
Add the ClickHouse data source in Grafana

Grafana needs a data source connection before it can query your analytics tables. You will point it at ClickHouse using the native protocol on port 9000.

  • Open your browser and go to http://localhost:3001.
  • On the sign-in page, enter admin for both the username and password.
  • Click Sign in.
  • Grafana will prompt you to change the default password. You can either set a new password or click Skip for now.

Why port 3001?

In your docker-compose.yml, Grafana's internal port 3000 is mapped to 3001 on your machine. This avoids a conflict with Dagster, which already uses port 3000.

  • In the left-side menu, click Connections.
  • Click Add new connection.
  • Type ClickHouse in the search bar.
  • Select ClickHouse from the results.
  • Click Add new data source.

The data source configuration page opens. Fill in the connection details:

  • In the Server field, enter clickhouse.
  • Set Protocol to Native.
  • Set Port to 9000.
  • Leave Username as default.
  • Leave Password empty.
  • In the Default database field, enter analytics.

Why use the hostname clickhouse?

Grafana and ClickHouse are both running inside the same Docker network called pipeline. Docker automatically resolves container service names as hostnames, so Grafana can reach ClickHouse at clickhouse:9000.

  • Scroll to the bottom and click Save & test.

You should see a green banner that says the data source is working.

Connection test failed?

  • Make sure the ClickHouse container is running. Check with docker ps and look for the clickhouse service.
  • Confirm you entered the server as clickhouse (the Docker service name), not localhost. Grafana runs inside a container and cannot reach the host's localhost.
  • Verify the port is 9000 (native protocol), not 8123 (HTTP).

help me troubleshoot my Grafana ClickHouse data source connection

Create a dashboard with three analytics panels

With the data source connected, you can now build a dashboard that queries your dbt mart tables directly.

  • In the left-side menu, click Dashboards.
  • Click New and select New Dashboard.
  • On the empty dashboard, click + Add visualization.
  • Select your ClickHouse data source from the list.

The panel editor opens. You will create three panels one at a time.

Panel 1: daily revenue time series
  • In the query editor, switch to the SQL Editor tab if it is not already selected.
  • Enter the following query:
SELECT
    order_day AS time,
    total_revenue
FROM analytics.mart_revenue_by_day
ORDER BY order_day

What does this query do?

This selects the order_day column aliased as time (which Grafana uses for the x-axis) and total_revenue for the y-axis. The data comes from your dbt mart table mart_revenue_by_day.

  • Click Refresh (or press the run button) to execute the query.
  • In the visualization list on the right, select Time series.
  • Under Panel options, set the title to Daily Revenue.
  • Click Apply in the top-right corner to save the panel and return to the dashboard.

You should see a line chart showing revenue over time.

Panel 2: top 10 customers bar chart
  • On the dashboard, click Add in the header and select Visualization.
  • Select your ClickHouse data source.
  • In the SQL Editor, enter the following query:
SELECT
    customer_name,
    total_spend
FROM analytics.mart_top_customers
ORDER BY total_spend DESC
LIMIT 10

What does this query do?

This pulls the top 10 highest-spending customers from mart_top_customers. The LIMIT 10 ensures the bar chart stays readable.

  • Click Refresh to execute the query.
  • In the visualization list, select Bar chart.
  • Under Panel options, set the title to Top 10 Customers by Spend.
  • Click Apply.

A horizontal bar chart appears showing customer names and their total spend.

Panel 3: revenue by category pie chart
  • Click Add in the dashboard header and select Visualization.
  • Select your ClickHouse data source.
  • In the SQL Editor, enter the following query:
SELECT
    product_category,
    total_revenue
FROM analytics.mart_category_performance
ORDER BY total_revenue DESC

What does this query do?

This retrieves revenue grouped by product category from mart_category_performance. Grafana will display each category as a slice of the pie, sized by revenue.

  • Click Refresh to execute the query.
  • In the visualization list, select Pie chart.
  • Under Panel options, set the title to Revenue by Category.
  • Click Apply.

Your dashboard now has three panels showing daily revenue trends, top customers, and category distribution.

Panels showing no data?

  • Make sure you set the Default database to analytics in the data source settings. If you left it blank, prefix table names with analytics. in your queries.
  • Verify the mart tables have data by running dbt run again from your terminal if you are unsure.
  • Expand the dashboard time range using the time picker in the top-right corner. Set it to Last 2 years since the generated data spans the past two years.

help me fix my Grafana panels that show no data from ClickHouse

Save the dashboard and enable auto-refresh
  • Click Save dashboard (the floppy disk icon in the top-right corner).
  • Enter the title E-Commerce Analytics and click Save.
  • In the top-right corner, find the refresh interval dropdown (next to the time picker).
  • Set it to 30s.

Your dashboard will now automatically re-query ClickHouse every 30 seconds, keeping the visualizations current.

The full pipeline in action

Take a moment to appreciate what you have built. Data starts in PostgreSQL as transactional rows. Debezium captures changes via CDC and publishes them to Kafka.

The ClickHouse sink connector writes those events into an OLAP database. dbt transforms the raw data into analytics-ready marts. Dagster orchestrates the transformations. And Grafana makes it all visible.

Your analytics platform is fully operational. Next up, the Secret Mission challenges you to add real-time streaming so new orders flow through the entire pipeline automatically.

Secret mission

Add a Real-Time Streaming Demonstration

Your pipeline works for bulk data. But real data platforms handle continuous streams of new events. Can you make new orders flow from PostgreSQL through Kafka into ClickHouse in near real-time, then automate dbt refreshes so Grafana stays up to date?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally via Docker, so there are no ongoing cloud costs. However, the 7-service stack consumes 4-6 GB of RAM while running.

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

Cost warning

While there is no monetary cost, leaving Docker Compose running ties up 4-6 GB of RAM and significant CPU. If your machine has limited resources, choose Pause or Delete below.

Resources you used:

  • Docker containers (PostgreSQL, Kafka, Kafka Connect, Kafka UI, ClickHouse, Dagster, Grafana).
  • Docker volumes storing database data, Kafka logs, and ClickHouse tables.
  • Python virtual environment (.venv) with installed packages.
  • Project folder (data-pipeline) on your Desktop.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the pipeline.

  • Leave Docker Compose running.
  • Monitor resource usage by running docker stats in your terminal.
  • Your virtual environment stays active until you close the terminal or run deactivate.

Pause - I'll come back to this later

Shut down running containers to free up RAM and CPU, but keep all data intact so you can resume later.

  • Stop all containers by running this command from your data-pipeline folder:
docker compose stop
  • This halts all 7 services without removing any data or volumes.
  • When you are ready to resume, run this from the same folder:
docker compose start
  • All containers restart with your data intact.

Delete - I don't want to use this again

Remove all project resources and start fresh. This frees up all disk space and RAM.

  • Stop all containers and delete all volumes (including all database data) by running this from your data-pipeline folder:
docker compose down -v
  • Remove the custom Kafka Connect image that was built during the project:
docker image prune -f
  • Delete the entire project folder. On macOS/Linux:
rm -rf ~/Desktop/data-pipeline
  • On Windows (run in Command Prompt, not PowerShell):
rmdir /s /q %USERPROFILE%\Desktop\data-pipeline
  • This removes all project files, the Python virtual environment, and all installed pip packages.

That's a Wrap!

That's a Wrap!

Nice work! You just built a complete modern data stack from scratch, running seven interconnected services on your local machine.

Here's what you learned:

  • Deployed a 7-service data platform with Docker Compose, including PostgreSQL, Kafka, Kafka Connect, ClickHouse, Kafka UI, Dagster, and Grafana.
  • Configured a CDC pipeline using Debezium to capture PostgreSQL changes in real time, stream them through Apache Kafka, and land them in ClickHouse via the official sink connector.
  • Built a layered dbt project for analytics modelling, orchestrated transformations with Dagster, and visualized business intelligence in Grafana dashboards.
  • Secret Mission: Added a real-time streaming simulation to demonstrate the full CDC lifecycle from PostgreSQL insert to Grafana visibility.

Ready to quiz yourself?