This pipeline ingests the Olist Brazilian E-Commerce dataset from a source PostgreSQL database and streams it in real time through Kafka into a clean analytics warehouse — with no manual refresh needed.

Source DB
    │
    │  WAL (Write-Ahead Log)
    ▼
Debezium CDC  ──►  Kafka Topics  ──►  Spark Streaming
                                            │
                                            ▼
                                    Silver Tables  (ecommerce_db)
                                            │
                                            ▼
                                    Gold Materialized Views

Any insert or update in the source DB propagates to the gold analytics layer within ~10 seconds — automatically.

Raw, cleaned mirror of source tables. Upserted on every Kafka batch.

customers · orders · order_items · order_payments · order_reviews · products · sellers

Aggregated and dashboard-ready. Auto-refreshed whenever orders or payments are updated.

• Docker Desktop
• PostgreSQL 14+ — running locally
• Taskfile — task runner

npm install -g @go-task/cli

PostgreSQL must have logical WAL enabled — required for Debezium CDC.

Connect to PostgreSQL and run:

-- Enable logical replication
ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_replication_slots = 4;
ALTER SYSTEM SET max_wal_senders = 4;

-- Grant replication permission to your user
ALTER USER postgres REPLICATION;

Then restart PostgreSQL for the changes to take effect:

# Windows (run as Administrator)
net stop postgresql-x64-17
net start postgresql-x64-17

# Linux / macOS
sudo systemctl restart postgresql

Verify:

SHOW wal_level;  -- should return 'logical'

git clone https://github.com/your-username/ecommerce-dashboard.git
cd ecommerce-dashboard
cp .env.example .env

Edit .env:

# Silver / Gold database
DB_URL=jdbc:postgresql://host.docker.internal:5432/ecommerce_db
DB_HOST=host.docker.internal
DB_NAME=ecommerce_db
DB_USER=postgres
DB_PASS=your_password

# Source database
SOURCE_DB_NAME=source_db
TOPIC_PREFIX=kafka_topic_prefix

# Kafka
KAFKA_BOOTSTRAP_SERVERS=kafka:29092

On Linux, replace host.docker.internal with 172.17.0.1

psql -U postgres -c "CREATE DATABASE ecommerce_db;"
psql -U postgres -d ecommerce_db -f init.sql
psql -U postgres -d ecommerce_db -f gold_layer.sql

Download the dataset from Kaggle, place the CSV files in csvs/, then:

task load

This loads all CSVs into the source database.

task up

That's it. Debezium will snapshot the source DB, push all records through Kafka, and Spark will populate the silver and gold layers automatically.

task up                # Start all services
task down              # Stop all services
task restart           # Stop then start
task build             # Rebuild Docker images
task load              # Load CSVs into source DB (once)

task status            # Container status
task connector         # Debezium connector status
task logs-spark        # Spark streaming logs
task logs-kafka        # Kafka logs
task logs-debezium     # Debezium connector logs

task db-source         # psql into source DB 
task db-silver         # psql into silver DB (ecommerce_db)
task refresh-gold      # Manually refresh all gold views

task clean             # Stop + delete all volumes
task clean-checkpoints # Clear Spark checkpoints

Gold views are empty

psql -U postgres -d ecommerce_db -c "SELECT COUNT(*) FROM silver_orders WHERE order_purchase_timestamp IS NOT NULL;"

If count is 0, timestamps failed to parse. Clear checkpoints and rebuild:

task clean-checkpoints
docker compose up -d --build streaming-app

Debezium connector in FAILED state

task connector   # check status
curl -X POST http://localhost:8083/connectors/postgres-connector/restart

Kafka broker conflict on restart

task clean       # wipes volumes
task up

Inspect a raw Kafka message

docker exec ecommerce-dashboard-kafka-1 kafka-console-consumer \
  --bootstrap-server localhost:29092 \
  --topic kafka_topic_prefix.public.orders \
  --from-beginning --max-messages 1

ecommerce-dashboard/
├── csvs/                          # Source CSV files (Kaggle dataset)
├── debezium/
│   └── postgres-connector.json    # Debezium connector config 
|
├── dockerfiles/
│   ├── csv_to_db.Dockerfile
│   └── spark_app.Dockerfile
├── scripts/
│   ├── csv_to_db.py               # CSV loader
│   └── spark_app.py               # Spark streaming job
|
├── init.sql                       # Silver layer schema
├── gold_layer.sql                 # Gold materialized views
├── compose.yml                    # Main services
├── load-compose.yml               # CSV loader service
├── Taskfile.yml
└── .env.example

• Debezium sends PostgreSQL timestamps as microseconds since epoch (MicroTimestamp). Spark converts them with from_unixtime(col / 1000000).
• Clearing checkpoints causes Spark to re-read Kafka from the beginning. Upsert logic prevents duplicate data.
• Gold views refresh automatically when orders or order_payments batches are processed. Use task refresh-gold to trigger manually.
• host.docker.internal is a Docker DNS name for the host machine. Not available on Linux by default — use 172.17.0.1 or add extra_hosts to compose.yml.