🚀 Customer Purchase Prediction ML System

A MLOps pipeline that transforms e-commerce behavior data into real-time purchase predictions. Built on modern open-source technologies including Kafka, Flink, Spark, Ray, and MLflow, this project demonstrates a complete ML lifecycle from data ingestion through model deployment. The system features automated CDC, multi-layer data warehousing, real-time feature serving, and comprehensive observability.

📑 Table of Contents

• 📊 Dataset
• File Structure
• Event Types
• Modeling: Customer Purchase Prediction
• 🌐 Architecture Overview
• 1. Data Pipeline
  • 📤 Data Sources
  • ✅ Schema Validation
  • ☁️ Storage Layer
  • 🛒 Spark Streaming
• 2. Training Pipeline
  • 🌟 Distributed Training
  • 📦 Model Management
• 3. Serving Pipeline
  • ⚡ Model Serving
  • 🔍 Feature Service
• 4. Observability
  • 📡 Metrics & Monitoring
  • 🔒 Access Management
• 📖 Details
• 🔧 Setup Environment Variables
• 🏁 Start Data Pipeline
• ✅ Start Schema Validation Job
• ☁️ Start Data Lake
• 🔄 Start Orchestration
• Data and Training Pipeline
  • 🔄 Data Pipeline
  • 🤼‍♂️ Training Pipeline
  • 📦 Start Online Store
• 🚀 Start Serving Pipeline
• 🔎 Start Observability
  • 📈 SigNoz
  • 📉 Prometheus and Grafana
• 🔒 NGINX
• Contributing
• 📃 License

📊 Dataset

eCommerce Behavior Data from Multi Category Store

The dataset can be found here. This dataset contains behavior data from over 285 million user events on a large multi-category eCommerce website.

The data spans 7 months (October 2019 to April 2020) and captures user-product interactions like views, cart additions/removals, and purchases. Each event represents a many-to-many relationship between users and products.

The dataset was collected by the Open CDP project, an open source customer data platform that enables tracking and analysis of user behavior data.

File Structure

Field	Description
event_time	UTC timestamp when the event occurred
event_type	Type of user interaction event
product_id	Unique identifier for the product
category_id	Product category identifier
category_code	Product category taxonomy (when available for meaningful categories)
brand	Brand name (lowercase, may be missing)
price	Product price (float)
user_id	Permanent user identifier
user_session	Temporary session ID that changes after long user inactivity

Event Types

The dataset captures four types of user interactions:

• view: User viewed a product
• cart: User added a product to shopping cart
• remove_from_cart: User removed a product from shopping cart
• purchase: User purchased a product

Modeling: Customer Purchase Prediction

The core modeling task is to predict whether a user will purchase a product at the moment they add it to their shopping cart.

Feature Engineering

We transform the raw event data into meaningful features for our machine learning model. The analysis focuses specifically on cart addition events and their subsequent outcomes.

Key engineered features include:

Feature	Description
category_code_level1	Main product category
category_code_level2	Product sub-category
event_weekday	Day of week when cart addition occurred
activity_count	Total user activities in the current session
price	Original product price
brand	Product brand name
is_purchased	Target variable: whether cart item was eventually purchased

You can download the dataset and put it under the data folder.

🌐 Architecture Overview

The system comprises four main components—Data, Training, Serving, and Observability—alongside a Dev Environment and a Model Registry.

1. Data Pipeline

📤 Data Sources

• Kafka Producer: Continuously emits user behavior events to tracking.raw_user_behavior topic
• CDC Service: Uses Debezium to capture PostgreSQL changes, streaming to tracking_postgres_cdc.public.events

✅ Schema Validation

• Validates incoming events from both sources
• Routes events to:
• tracking.user_behavior.validated for valid events
• tracking.user_behavior.invalid for schema violations
• Handles ~10k events/second
• Alerts invalid events to Elasticsearch

☁️ Storage Layer

• Data Lake (MinIO):
• External Storage
• Stores data in time-partitioned buckets (year/month/day/hour)
• Supports checkpointing for pipeline resilience
• Data Warehouse (PostgreSQL):
• Organized in bronze → silver → gold layers
• Houses dimension/fact tables for analysis purposes
• Offline Store (PostgreSQL):
• Used for training and batch feature serving
• Periodically materialized to online store
• Online Store (Redis):
• Low-latency feature serving
• Updated through streaming pipeline
• Exposed via Feature Retrieval API

🛒 Spark Streaming

• Transforms validated events into ML features
• Focuses on session-based metrics and purchase behavior
• Dual-writes to online/offline stores

2. Training Pipeline

🌟 Distributed Training

• Ray Cluster:
• Handles distributed hyperparameter tuning via Ray Tune
• Executes final model training
• Integrates with MLflow for experiment tracking

📦 Model Management

• MLflow + MinIO + PostgreSQL:
• Tracks experiments, parameters, and metrics
• Versions model artifacts
• Provides model registry UI at localhost:5001

3. Serving Pipeline

⚡ Model Serving

• Ray Serve:
• Loads models from MLflow registry
• Automatically scales horizontally for high throughput
• Provides REST API for predictions
• Feature Service:
• FastAPI endpoint for feature retrieval
• Integrates with Redis for real-time features

4. Observability

📡 Metrics & Monitoring

• SigNoz:
• Collects OpenTelemetry data
• Provides service-level monitoring
• Accessible at localhost:3301
• Ray Dashboard:
• Monitors training/serving jobs
• Available at localhost:8265
• Prometheus + Grafana:
• Tracks Ray cluster metrics
• Visualizes system performance
• Accessible at localhost:3009
• Superset:
• Visualize the data in the Data Warehouse
• Accessible at localhost:8089
• Elasticsearch:
• Alert invalid events

🔒 Access Management

• NGINX Proxy Manager:
• Reverse proxy for all services
• SSL/TLS termination
• Access control and routing

The architecture prioritizes reliability, scalability, and observability while maintaining clear separation of concerns between pipeline stages. Each component is containerized and can be deployed independently using Docker Compose.

---

📖 Details

All available commands can be found in the Makefile.

In this section, we will dive into the details of the system.

🔧 Setup Environment Variables

Please run the following command to setup the .env files:

cp .env.example .env
cp ./src/cdc/.env.example ./src/cdc/.env
cp ./src/model_registry/.env.example ./src/model_registry/.env
cp ./src/orchestration/.env.example ./src/orchestration/.env
cp ./src/producer/.env.example ./src/producer/.env
cp ./src/streaming/.env.example ./src/streaming/.env

Note: I don't use any secrets in this project, so run the above command and you are good to go.

🏁 Start Data Pipeline

I will use the same network for all the services, first we need to create the network.

make up-network

🐟 Start Kafka

make up-kafka

The last service in the docker-compose.kafka.yaml file is kafka_producer, this service acts as a producer and will start sending messages to the tracking.raw_user_behavior topic.

To check if Kafka is running, you can go to localhost:9021 and you should see the Kafka dashboard. Then go to the Topics tab and you should see tracking.raw_user_behavior topic.

To check if the producer is sending messages, you can click on the tracking.raw_user_behavior topic and you should see the messages being sent.

Here is an example of the message's value in the tracking.raw_user_behavior topic:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "name": "event_time",
        "type": "string"
      },
      {
        "name": "event_type",
        "type": "string"
      },
      {
        "name": "product_id",
        "type": "long"
      },
      {
        "name": "category_id",
        "type": "long"
      },
      {
        "name": "category_code",
        "type": ["null", "string"],
        "default": null
      },
      {
        "name": "brand",
        "type": ["null", "string"],
        "default": null
      },
      {
        "name": "price",
        "type": "double"
      },
      {
        "name": "user_id",
        "type": "long"
      },
      {
        "name": "user_session",
        "type": "string"
      }
    ]
  },
  "payload": {
    "event_time": "2019-10-01 02:30:12 UTC",
    "event_type": "view",
    "product_id": 1306133,
    "category_id": "2053013558920217191",
    "category_code": "computers.notebook",
    "brand": "xiaomi",
    "price": 1029.37,
    "user_id": 512900744,
    "user_session": "76b918d5-b344-41fc-8632-baf222ec760f"
  }
}

🔄 Start CDC (2)

make up-cdc

Next, we start the CDC (Change Data Capture) service using Docker Compose. This setup includes the following components:

• Debezium: Monitors the Backend DB for any changes (inserts, updates, deletes) and captures those changes.
• PostgreSQL: The database where the changes are being monitored.
• A Python service: Registers the connector, creates the table, and inserts the data into PostgreSQL.

Steps involved:

• Debezium monitors the Backend DB for any changes. (2.1)
• Debezium captures these changes and pushes them to the Raw Events Topic in the message broker. (2.2)

The data is automatically synced from PostgreSQL to the tracking_postgres_cdc.public.events topic. To confirm this, go to the Connect tab in the Kafka UI; you should see a connector named cdc-postgresql.

Return to localhost:9021; there should be a new topic called tracking_postgres_cdc.public.events.

✅ Start Schema Validation Job

make schema_validation

This is a Flink job that will consume the tracking_postgres_cdc.public.events and tracking.raw_user_behavior topics and validate the schema of the events. The validated events will be sent to the tracking.user_behavior.validated topic and the invalid events will be sent to the tracking.user_behavior.invalid topic, respectively. For easier understanding, I don't push these Flink jobs into a Docker Compose file, but you can do it if you want. Watch the terminal to see the job running, the log may look like this:

We can handle 10k RPS, noting that approximately 10% of events are failures. I purposely make the producer send invalid events to the tracking.user_behavior.invalid topic. You can check this at line 127 in src/producer/produce.py.

After starting the job, you can go to localhost:9021 and you should see the tracking.user_behavior.validated and tracking.user_behavior.invalid topics.

Beside that, we can also start the alert_invalid_events job to alert the invalid events.

make alert_invalid_events

Note: This feature of pushing the invalid events to Elasticsearch is not implemented yet, I will implement it in the future, but you can do it easily by modifying the src/streaming/jobs/alert_invalid_events_job.py file.

🔄 Transformation Job (4)

First, we need to start the Data Warehouse and the Online Store.

make up-dwh
make up-online-store

📦 Data Warehouse

The Data Warehouse is just a PostgreSQL instance.

📦 Online Store

The Online Store is a Redis instance.

Look at the `docker-co