Build an ML Feature Platform
Build a production ML feature platform with batch, streaming, and agentic discovery.
Introduction
30 Second Summary
Every time an app recommends a song, personalizes an ad, or detects fraud in real time, an ML model is making a decision. Behind that model sits a feature platform that transforms raw event data into the signals the model actually uses to make predictions.
In this project, you will build a production-style ML feature platform that computes, stores, and serves features through batch and real-time pipelines. You will deploy a live API backed by Redis that anyone can hit to retrieve features on demand.
What You'll Build
A full ML feature platform with a declarative YAML registry, batch computation using pandas and DuckDB, real-time streaming with Apache Kafka to Redis, a FastAPI serving endpoint with drift detection, and agentic feature discovery powered by a local LLM.
By the end of this project, you'll have:
- A feature registry with versioned YAML definitions and a batch pipeline that computes features with point-in-time correctness to prevent data leakage.
- A real-time streaming pipeline using Apache Kafka that computes features from live events and materializes them to Redis within seconds.
- A deployed FastAPI serving API on Render with PSI-based drift monitoring and both online and offline feature retrieval.
- Secret Mission: an agentic feature discovery workflow that uses your local LLM to analyze raw data, propose new features, evaluate their quality, and automatically add top-scoring features to your registry.
Are there any prerequisites?
You should have basic Python knowledge and basic familiarity with APIs. No ML modeling experience is required.
You will also need Docker Desktop and LM Studio installed on your machine. Step 1 walks you through the full environment setup.
Before We Start
ML models are only as good as the features they receive. The infrastructure that transforms raw event data into reliable, versioned features is what separates production ML systems from prototypes.
In this step, you'll set up your development environment with Python, Docker containers for Redis and Apache Kafka, and a cloud Redis database for deployment.
In this step, get ready to:
- Create a Python virtual environment and install all project dependencies.
- Start local Docker containers for Redis and Kafka.
- Set up an Upstash Redis database and configure environment variables.
Create a Python virtual environment and install dependencies
The feature platform depends on multiple Python libraries for streaming, storage, and API serving. A virtual environment keeps these dependencies isolated from your system Python.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
First, confirm you have the right version of Python installed.
- Check your Python version by running this command:
python3 --version✔️ I see version 3.10 or higher
Your Python version is compatible. Continue to the next instruction below.
ⓧ I see an older version
The project requires Python 3.10 or higher. macOS often ships with an older Python that shadows newer installs.
- Install the latest Python from the official site: https://www.python.org/downloads/
- After installing, close and reopen your terminal.
- Run python3 --version again to confirm you see 3.10 or higher.
- If it still shows an older version, try running python3.12 --version or python3.11 --version to use the specific binary.
ⓧ Command not found
Python is not installed on your system.
- Download and install Python from the official site: https://www.python.org/downloads/
- During installation on Windows, check the box that says Add python.exe to PATH.
- Close and reopen your terminal after installation.
- Run python3 --version (macOS) or python --version (Windows) to confirm you see 3.10 or higher.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir ml-feature-platform
cd ml-feature-platform- Create a Python virtual environment and activate it by running these commands:
python3 -m venv venv
source venv/bin/activateWhat about Windows?
On Windows, activate the virtual environment with venv\Scripts\activate instead of the source command. You should see (venv) appear at the start of your terminal prompt on both platforms.
- Create the project directory structure by running this command:
mkdir registry pipelines store serving monitoring agent features- Confirm the folders exist by running:
lsYou should see agent, features, monitoring, pipelines, registry, serving, store, and venv listed in the output.
- Open VS Code in the current directory by running:
code .code command not found?
If the code command is not recognized, open VS Code manually. Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows) inside VS Code, type Shell Command: Install 'code' command in PATH, and select it. Then retry the command.
Help me open VS Code from the terminal
- In VS Code, create a new file called requirements.txt in the root of your ml-feature-platform folder.
- Paste the following dependency list into requirements.txt:
fastapi==0.139.0
uvicorn>=0.12.0
duckdb==1.5.4
redis==8.0.1
confluent-kafka==2.15.0
pandas>=2.0.0
pyyaml>=6.0
pydantic>=2.9.0
pydantic-settings>=2.0.0
openai==2.44.0
scipy>=1.10.0
pytest>=7.0.0
python-dotenv>=1.0.0What are these dependencies?
This project uses FastAPI for the serving API, DuckDB as the offline feature store, redis for the online store, and confluent-kafka for real-time streaming.
The openai SDK talks to your local LM Studio server for the agentic feature discovery agent. scipy handles the statistical drift detection calculations.
- Save requirements.txt.
- Install all the dependencies by running this command in your VS Code terminal:
pip install -r requirements.txtThis takes a minute or two. You should see a success message ending with Successfully installed followed by a list of package names.
Installation failing?
Make sure your virtual environment is activated. You should see (venv) at the start of your terminal prompt. If not, run source venv/bin/activate (macOS) or venv\Scripts\activate (Windows) again.
If confluent-kafka fails to build, ensure you have Python 3.10+ and that your system has a C compiler available (Xcode Command Line Tools on macOS, Visual Studio Build Tools on Windows).
Help me fix my pip install errors
Start Docker containers for Redis and Kafka
Your feature platform needs a local Redis instance for the online feature store and a Kafka broker for the real-time streaming pipeline. Docker Compose spins both up with a single command.
- In VS Code, create a new file called docker-compose.yml in the root of your ml-feature-platform folder.
- Paste the following configuration into docker-compose.yml:
services:
redis:
image: redis:7.4
ports:
- "6379:6379"
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 3s
retries: 5
zookeeper:
image: confluentinc/cp-zookeeper:7.9.0
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
kafka:
image: confluentinc/cp-kafka:7.9.0
depends_on:
- zookeeper
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1What does this set up?
This defines three services. redis runs the local Redis instance on port 6379 for fast feature lookups. zookeeper manages Kafka cluster coordination. kafka is the message broker on port 9092 that your streaming pipeline publishes events to.
The healthcheck on Redis confirms the container is ready before other services try to connect.
- Save docker-compose.yml.
- Make sure Docker Desktop is running on your machine.
- Start the containers by running this command in your VS Code terminal:
docker compose up -dWhat does this command do?
The -d flag runs the containers in detached mode so your terminal stays free. Docker pulls the Redis and Kafka images on first run, which may take a few minutes.
- Verify all containers are running by running:
docker compose psYou should see three services listed with a running or Up status: redis, zookeeper, and kafka.
Containers not starting?
Confirm Docker Desktop is running. Check the Docker Desktop icon in your system tray or menu bar. If you see a whale icon, it should be active.
If a port conflict error appears, another process is already using port 6379 or 9092. Stop the conflicting process or change the host port in docker-compose.yml.
Help me troubleshoot my Docker containers
Create an Upstash Redis database and configure your .env file
The local Redis container works for development, but your deployed API needs a cloud-hosted Redis. Upstash provides a free managed Redis database. No credit card required.
- Open https://console.upstash.com in your browser.
- Sign up for a free account (or log in if you already have one).
- From the Redis tab, click + Create Database in the upper right corner.
- Enter a name for your database (e.g. ml-feature-platform).
- Select a Primary Region close to your location (e.g. us-east-1).
- Click Next and select the Free plan.
- Your database is now ready. On the database details page, scroll down to find the REST API section.
- Copy the UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN values. Save them somewhere safe for the next step.
What is Upstash?
Upstash is a serverless Redis provider. The free tier gives you 256MB of storage and 500,000 commands per month. Your deployed feature serving API will connect to this database to read features with low latency.
- In VS Code, create a new file called .env in the root of your ml-feature-platform folder.
- Paste the following into your .env file, replacing the placeholder values with your Upstash credentials:
UPSTASH_REDIS_URL=your-upstash-redis-rest-url
UPSTASH_REDIS_TOKEN=your-upstash-redis-rest-token
LM_STUDIO_URL=http://localhost:1234/v1Why these variable names?
The UPSTASH_REDIS_URL and UPSTASH_REDIS_TOKEN variables connect your deployed API to the cloud Redis store. LM_STUDIO_URL points to your local LM Studio server, which exposes an OpenAI-compatible API at http://localhost:1234/v1.
The .env file keeps sensitive credentials out of your source code. The python-dotenv package loads these values automatically at runtime.
- Save .env.
Keep .env out of version control
When you push this project to GitHub later, add .env to your .gitignore file so your credentials are never committed.
✔️ Awesome, I've got everything!
Great. Double check you've saved all three files: requirements.txt, docker-compose.yml, and .env.
ⓧ I'd like to double check the full code
Here are the three files you should have at this point.
requirements.txt:
fastapi==0.139.0
uvicorn>=0.12.0
duckdb==1.5.4
redis==8.0.1
confluent-kafka==2.15.0
pandas>=2.0.0
pyyaml>=6.0
pydantic>=2.9.0
pydantic-settings>=2.0.0
openai==2.44.0
scipy>=1.10.0
pytest>=7.0.0
python-dotenv>=1.0.0docker-compose.yml:
services:
redis:
image: redis:7.4
ports:
- "6379:6379"
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 3s
retries: 5
zookeeper:
image: confluentinc/cp-zookeeper:7.9.0
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
kafka:
image: confluentinc/cp-kafka:7.9.0
depends_on:
- zookeeper
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1.env:
UPSTASH_REDIS_URL=<your-upstash-redis-url>
UPSTASH_REDIS_TOKEN=<your-upstash-redis-token>
LM_STUDIO_URL=http://localhost:1234/v1Your environment is ready. Next up, you'll build the feature registry and batch pipeline that transforms raw event data into versioned ML features.
Build the Feature Registry and Batch Pipeline
Your environment is ready with Docker containers running, dependencies installed, and your project structure in place. Now it's time to build the core of your feature platform: the system that defines, stores, and computes ML features.
ML models don't consume raw event data directly. They need transformed, versioned features (like "user clicks in the last 7 days") served reliably. In this step, you'll build the two foundational layers that make this possible: a feature registry that declares what features exist, and a batch pipeline that computes them from raw events.
In this step, get ready to:
- Create a YAML-based feature registry with versioned definitions and a Python class to manage them.
- Build a batch pipeline that computes features from raw event data and stores them in DuckDB.
- Implement point-in-time correct feature retrieval to prevent data leakage in training datasets.
Build the feature registry
A feature registry is the source of truth for what features exist in your platform. It defines each feature's name, version, data type, and how it's computed. Production platforms like Feast and Tecton use declarative definitions so teams can discover, share, and audit features across an organization.
- Create the feature definitions file by adding the following YAML to features/user_features.yaml:
feature_group: user_engagement
entity: user_id
features:
- name: user_click_count_7d
version: 1
dtype: int64
description: Total ad clicks by user in the last 7 days
transformation: count
window: 7d
source: ad_events
created_by: platform
- name: user_avg_session_duration
version: 1
dtype: float64
description: Average session duration in minutes over last 7 days
transformation: mean
window: 7d
source: session_events
created_by: platform
- name: ad_ctr_score
version: 1
dtype: float64
description: Click-through rate for ads shown to this user
transformation: ratio
numerator: click_count
denominator: impression_count
window: 7d
source: ad_events
created_by: platformWhat does this YAML define?
This file declares three features for the user_engagement group, all keyed by user_id.
- user_click_count_7d: counts ad clicks over a 7-day window.
- user_avg_session_duration: averages session duration over 7 days.
- ad_ctr_score: a ratio of clicks to impressions, giving each user a click-through rate.
Each feature carries lineage metadata (source, created_by) so you always know where a feature came from.
- Save features/user_features.yaml.
Now you need a Python class that can load these YAML definitions, validate them, and register them in a queryable database.
- Create registry/feature_registry.py and add the Pydantic models that define the shape of feature definitions:
import yaml
import duckdb
from pathlib import Path
from pydantic import BaseModel
from typing import Optional
class FeatureDefinition(BaseModel):
name: str
version: int
dtype: str
description: str
transformation: str
window: Optional[str] = None
source: str
created_by: str
class FeatureGroup(BaseModel):
feature_group: str
entity: str
features: list[FeatureDefinition]What do these models do?
FeatureDefinition is a Pydantic model that validates each feature's metadata fields (name, version, dtype, etc.). FeatureGroup wraps a collection of features under one entity key.
When you load the YAML file, Pydantic automatically validates that all required fields are present and correctly typed. Invalid definitions fail loudly instead of silently corrupting your feature store.
- Below the models, add the FeatureRegistry class that manages loading, storing, and querying feature definitions:
class FeatureRegistry:
def __init__(self, features_dir: str = "features", db_path: str = "feature_store.db"):
self.features_dir = Path(features_dir)
self.db_path = db_path
self.conn = duckdb.connect(db_path)
self._init_db()
self.groups: dict[str, FeatureGroup] = {}
def _init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS feature_registry (
name VARCHAR,
version INTEGER,
feature_group VARCHAR,
entity VARCHAR,
dtype VARCHAR,
description VARCHAR,
transformation VARCHAR,
source VARCHAR,
created_by VARCHAR,
registered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (name, version)
)
""")
def load_definitions(self):
for yaml_file in self.features_dir.glob("*.yaml"):
with open(yaml_file) as f:
data = yaml.safe_load(f)
group = FeatureGroup(**data)
self.groups[group.feature_group] = group
for feature in group.features:
self.conn.execute("""
INSERT OR REPLACE INTO feature_registry
(name, version, feature_group, entity, dtype, description, transformation, source, created_by)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", [
feature.name, feature.version, group.feature_group,
group.entity, feature.dtype, feature.description,
feature.transformation, feature.source, feature.created_by
])
def get_feature(self, name: str, version: Optional[int] = None) -> Optional[FeatureDefinition]:
for group in self.groups.values():
for feature in group.features:
if feature.name == name:
if version is None or feature.version == version:
return feature
return None
def list_features(self) -> list[dict]:
result = self.conn.execute(
"SELECT name, version, feature_group, entity, dtype FROM feature_registry"
).fetchall()
return [
{"name": r[0], "version": r[1], "group": r[2], "entity": r[3], "dtype": r[4]}
for r in result
]What does the FeatureRegistry class do?
- __init__ connects to a DuckDB database file and creates the feature_registry table if it doesn't exist.
- load_definitions() reads every YAML file in the features/ directory, validates each with Pydantic, and writes the metadata into DuckDB using INSERT OR REPLACE (upserts on the primary key).
- get_feature() looks up a single feature by name and optional version.
- list_features() queries DuckDB for all registered features and returns them as dictionaries.
DuckDB stores the registry persistently in feature_store.db, so feature metadata survives restarts and is queryable with SQL.
- Save registry/feature_registry.py.
- Verify the registry works by running this in your terminal:
python -c "from registry.feature_registry import FeatureRegistry; r = FeatureRegistry(); r.load_definitions(); print(r.list_features())"You should see a list of three feature dictionaries printed, each with a name, version, group, entity, and dtype.
Seeing a ModuleNotFoundError?
- Make sure you're running the command from your project root directory (where the features/ and registry/ folders live).
- Check that your virtual environment is activated. You should see its name in your terminal prompt.
- Add an empty __init__.py file inside the registry/ folder if Python can't find the module.
Help me debug my FeatureRegistry import error.
✔️ Awesome, I've got everything!
Great. Double check you've saved both files.
ⓧ I'd like to double check the full code
import yaml
import duckdb
from pathlib import Path
from pydantic import BaseModel
from typing import Optional
class FeatureDefinition(BaseModel):
name: str
version: int
dtype: str
description: str
transformation: str
window: Optional[str] = None
source: str
created_by: str
class FeatureGroup(BaseModel):
feature_group: str
entity: str
features: list[FeatureDefinition]
class FeatureRegistry:
def __init__(self, features_dir: str = "features", db_path: str = "feature_store.db"):
self.features_dir = Path(features_dir)
self.db_path = db_path
self.conn = duckdb.connect(db_path)
self._init_db()
self.groups: dict[str, FeatureGroup] = {}
def _init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS feature_registry (
name VARCHAR,
version INTEGER,
feature_group VARCHAR,
entity VARCHAR,
dtype VARCHAR,
description VARCHAR,
transformation VARCHAR,
source VARCHAR,
created_by VARCHAR,
registered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (name, version)
)
""")
def load_definitions(self):
for yaml_file in self.features_dir.glob("*.yaml"):
with open(yaml_file) as f:
data = yaml.safe_load(f)
group = FeatureGroup(**data)
self.groups[group.feature_group] = group
for feature in group.features:
self.conn.execute("""
INSERT OR REPLACE INTO feature_registry
(name, version, feature_group, entity, dtype, description, transformation, source, created_by)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", [
feature.name, feature.version, group.feature_group,
group.entity, feature.dtype, feature.description,
feature.transformation, feature.source, feature.created_by
])
def get_feature(self, name: str, version: Optional[int] = None) -> Optional[FeatureDefinition]:
for group in self.groups.values():
for feature in group.features:
if feature.name == name:
if version is None or feature.version == version:
return feature
return None
def list_features(self) -> list[dict]:
result = self.conn.execute(
"SELECT name, version, feature_group, entity, dtype FROM feature_registry"
).fetchall()
return [
{"name": r[0], "version": r[1], "group": r[2], "entity": r[3], "dtype": r[4]}
for r in result
]Build the batch feature pipeline
The registry defines what features exist. Now you need a pipeline that actually computes them from raw event data. The batch pipeline reads historical events, applies transformations (counts, averages, ratios), and writes the results to DuckDB as your offline feature store.
- Create pipelines/batch_pipeline.py and add the class initialization and table setup:
import pandas as pd
import duckdb
from datetime import datetime, timedelta
class BatchFeaturePipeline:
def __init__(self, db_path: str = "feature_store.db"):
self.conn = duckdb.connect(db_path)
self._init_tables()
def _init_tables(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS feature_values (
entity_id VARCHAR,
feature_name VARCHAR,
feature_version INTEGER,
value DOUBLE,
event_timestamp TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS raw_events (
event_id VARCHAR,
user_id VARCHAR,
ad_id VARCHAR,
event_type VARCHAR,
session_duration DOUBLE,
timestamp TIMESTAMP
)
""")
def ingest_events(self, events_df: pd.DataFrame):
self.conn.execute("INSERT INTO raw_events SELECT * FROM events_df")What does this set up?
- feature_values table stores computed features with timestamps. Each row is one feature value for one entity at one point in time.
- raw_events table holds the source event data (ad clicks, impressions, sessions) that features are computed from.
- ingest_events() leverages DuckDB's ability to read directly from a pandas DataFrame without manual row-by-row insertion.
- Below the ingest_events method, add the feature computation and backfill logic:
def compute_features(self, as_of_date: datetime):
window_start = as_of_date - timedelta(days=7)
click_counts = self.conn.execute("""
SELECT user_id, COUNT(*) as click_count
FROM raw_events
WHERE event_type = 'click'
AND timestamp BETWEEN ? AND ?
GROUP BY user_id
""", [window_start, as_of_date]).fetchdf()
for _, row in click_counts.iterrows():
self.conn.execute("""
INSERT INTO feature_values (entity_id, feature_name, feature_version, value, event_timestamp)
VALUES (?, 'user_click_count_7d', 1, ?, ?)
""", [row["user_id"], float(row["click_count"]), as_of_date])
avg_sessions = self.conn.execute("""
SELECT user_id, AVG(session_duration) as avg_duration
FROM raw_events
WHERE session_duration IS NOT NULL
AND timestamp BETWEEN ? AND ?
GROUP BY user_id
""", [window_start, as_of_date]).fetchdf()
for _, row in avg_sessions.iterrows():
self.conn.execute("""
INSERT INTO feature_values (entity_id, feature_name, feature_version, value, event_timestamp)
VALUES (?, 'user_avg_session_duration', 1, ?, ?)
""", [row["user_id"], float(row["avg_duration"]), as_of_date])
def backfill(self, start_date: datetime, end_date: datetime):
current = start_date
while current <= end_date:
self.compute_features(current)
current += timedelta(days=1)What does the computation logic do?
- compute_features() takes an as_of_date and looks back 7 days to calculate click counts and average session durations per user.
- Each computed value gets written to feature_values with the event_timestamp set to the as-of date. This is critical for point-in-time correctness later.
- backfill() iterates day-by-day over a date range, calling compute_features() for each day. This lets you recompute features historically when you add new transformations.
- Save pipelines/batch_pipeline.py.
- You won't see output yet. The next substep adds the retrieval method that lets you query these stored features.
Seeing an indentation error?
- Make sure compute_features and backfill are indented one level inside the BatchFeaturePipeline class (they're methods, not standalone functions).
- Check that the SQL strings inside triple quotes are not accidentally broken across lines in a way that loses the closing """.
Help me fix a syntax error in my batch_pipeline.py file.
Implement point-in-time correct retrieval
One of the most dangerous bugs in ML is data leakage. If your training data includes features computed from future events, your model looks great offline but fails in production. Point-in-time correct retrieval solves this by only returning feature values that existed at or before a given timestamp.
Why does point-in-time correctness matter?
Imagine training a model to predict clicks on January 15th. If you accidentally use a feature value computed on January 20th, you're leaking future information into the training set. The model learns patterns it could never see at inference time.
Point-in-time retrieval ensures you only ever see what was known at that moment. This is the difference between a model that works in production and one that mysteriously underperforms after deployment.
- In pipelines/batch_pipeline.py, find the backfill method.
- Add the following get_training_features method directly above backfill (between compute_features and backfill):
def get_training_features(self, entity_id: str, feature_names: list[str], timestamp: datetime) -> dict:
features = {}
for name in feature_names:
result = self.conn.execute("""
SELECT value FROM feature_values
WHERE entity_id = ?
AND feature_name = ?
AND event_timestamp <= ?
ORDER BY event_timestamp DESC
LIMIT 1
""", [entity_id, name, timestamp]).fetchone()
features[name] = result[0] if result else None
return featuresWhat does this method do?
For each requested feature, the query finds the most recent value where event_timestamp <= ? (the given timestamp). The ORDER BY event_timestamp DESC LIMIT 1 ensures you get the latest known value without leaking future data.
If no feature value exists before that timestamp, it returns None. This is intentional. A training pipeline calling this function will know that feature wasn't available yet for that entity.
- Save pipelines/batch_pipeline.py.
- Verify the full pipeline by running this test in your terminal:
python -c "
import pandas as pd
from datetime import datetime
from pipelines.batch_pipeline import BatchFeaturePipeline
pipeline = BatchFeaturePipeline(db_path='test_store.db')
events = pd.DataFrame({
'event_id': ['e1','e2','e3'],
'user_id': ['user_1','user_1','user_2'],
'ad_id': ['ad_1','ad_2','ad_1'],
'event_type': ['click','click','click'],
'session_duration': [3.5, 2.0, 5.0],
'timestamp': [datetime(2026,7,1)]*3
})
pipeline.ingest_events(events)
pipeline.compute_features(datetime(2026,7,2))
result = pipeline.get_training_features('user_1', ['user_click_count_7d','user_avg_session_duration'], datetime(2026,7,2))
print(result)
"You should see a dictionary with user_click_count_7d showing 2.0 and user_avg_session_duration showing 2.75 for user_1.
Seeing None values or an empty dictionary?
- Make sure get_training_features is placed inside the BatchFeaturePipeline class, not outside it.
- Check that the timestamp argument in get_training_features is on or after the date you passed to compute_features. Features computed for July 2nd won't appear if you query for July 1st.
- Verify there's no typo in the feature names. They must exactly match the strings written by compute_features.
Help me debug why get_training_features returns None.
- Clean up the test database by running:
rm test_store.db✔️ Awesome, I've got everything!
Great. Double check you've saved pipelines/batch_pipeline.py.
ⓧ I'd like to double check the full code
import pandas as pd
import duckdb
from datetime import datetime, timedelta
class BatchFeaturePipeline:
def __init__(self, db_path: str = "feature_store.db"):
self.conn = duckdb.connect(db_path)
self._init_tables()
def _init_tables(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS feature_values (
entity_id VARCHAR,
feature_name VARCHAR,
feature_version INTEGER,
value DOUBLE,
event_timestamp TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS raw_events (
event_id VARCHAR,
user_id VARCHAR,
ad_id VARCHAR,
event_type VARCHAR,
session_duration DOUBLE,
timestamp TIMESTAMP
)
""")
def ingest_events(self, events_df: pd.DataFrame):
self.conn.execute("INSERT INTO raw_events SELECT * FROM events_df")
def compute_features(self, as_of_date: datetime):
window_start = as_of_date - timedelta(days=7)
click_counts = self.conn.execute("""
SELECT user_id, COUNT(*) as click_count
FROM raw_events
WHERE event_type = 'click'
AND timestamp BETWEEN ? AND ?
GROUP BY user_id
""", [window_start, as_of_date]).fetchdf()
for _, row in click_counts.iterrows():
self.conn.execute("""
INSERT INTO feature_values (entity_id, feature_name, feature_version, value, event_timestamp)
VALUES (?, 'user_click_count_7d', 1, ?, ?)
""", [row["user_id"], float(row["click_count"]), as_of_date])
avg_sessions = self.conn.execute("""
SELECT user_id, AVG(session_duration) as avg_duration
FROM raw_events
WHERE session_duration IS NOT NULL
AND timestamp BETWEEN ? AND ?
GROUP BY user_id
""", [window_start, as_of_date]).fetchdf()
for _, row in avg_sessions.iterrows():
self.conn.execute("""
INSERT INTO feature_values (entity_id, feature_name, feature_version, value, event_timestamp)
VALUES (?, 'user_avg_session_duration', 1, ?, ?)
""", [row["user_id"], float(row["avg_duration"]), as_of_date])
def get_training_features(self, entity_id: str, feature_names: list[str], timestamp: datetime) -> dict:
features = {}
for name in feature_names:
result = self.conn.execute("""
SELECT value FROM feature_values
WHERE entity_id = ?
AND feature_name = ?
AND event_timestamp <= ?
ORDER BY event_timestamp DESC
LIMIT 1
""", [entity_id, name, timestamp]).fetchone()
features[name] = result[0] if result else None
return features
def backfill(self, start_date: datetime, end_date: datetime):
current = start_date
while current <= end_date:
self.compute_features(current)
current += timedelta(days=1)You now have a declarative feature registry and a batch pipeline that can compute, store, and safely retrieve features. Next up, you'll build a real-time streaming pipeline that computes features from live events as they arrive.
Build the Real-Time Streaming Pipeline
Your batch pipeline can compute features from historical data, but ML models in production need features that update in real time.
In this step, you will build a streaming pipeline using Apache Kafka that ingests live ad-click events, computes features on the fly, and writes them to Redis for low-latency serving.
In this step, get ready to:
- Implement a Redis online store abstraction that works locally and in production.
- Build a Kafka producer and consumer that compute features from live events.
- Test the full streaming loop and verify features appear in Redis.
Implement the online store abstraction
The online store is the low-latency layer that serves pre-computed features to ML models at prediction time. Your platform needs a single interface that connects to local Redis during development and Upstash Redis in production.
- Create the store/ directory inside your project folder by running this command:
mkdir -p store- Create a new file called online_store.py inside the store/ directory.
- Add the following code to store/online_store.py:
import os
import redis
class RedisOnlineStore:
def __init__(self):
redis_url = os.getenv("UPSTASH_REDIS_URL")
if redis_url:
self.client = redis.from_url(
redis_url,
password=os.getenv("UPSTASH_REDIS_TOKEN"),
decode_responses=True,
)
else:
self.client = redis.Redis(
host="localhost", port=6379, decode_responses=True
)
def write_features(self, entity_id: str, features: dict):
key = f"features:{entity_id}"
self.client.hset(key, mapping={k: str(v) for k, v in features.items()})
def read_features(self, entity_id: str, feature_names: list[str]) -> dict:
key = f"features:{entity_id}"
if feature_names:
values = self.client.hmget(key, feature_names)
return {
name: float(val) if val is not None else None
for name, val in zip(feature_names, values)
}
else:
all_features = self.client.hgetall(key)
return {k: float(v) for k, v in all_features.items()}
def health_check(self) -> bool:
try:
return self.client.ping()
except Exception:
return FalseWhat does this code do?
- The __init__ method checks for an UPSTASH_REDIS_URL environment variable. If it exists, the store connects to your Upstash cloud Redis. If not, it falls back to the local Docker Redis on port 6379.
- write_features() stores all features for an entity in a single Redis hash, keyed as features:{entity_id}. This lets you read any subset of features in one network round trip.
- read_features() retrieves specific features by name, or all features if no names are given. Values are stored as strings and converted back to floats on read.
- health_check() pings Redis to confirm the connection is alive. You will use this in the serving API later.
Why a single abstraction for two Redis instances?
Production feature platforms use different infrastructure in development and production. Your local Docker Redis is free and fast for testing. Upstash Redis is a managed service that stays available when your laptop is off.
By switching on an environment variable, the same code works in both environments without any changes to your pipeline logic.
- Save store/online_store.py.
- Verify the store connects to your local Redis by running this command from your project root:
python -c "from store.online_store import RedisOnlineStore; store = RedisOnlineStore(); print(store.health_check())"You should see True printed, confirming Redis is reachable.
Seeing a ConnectionError or False?
- Make sure your Docker containers are still running. Run docker compose ps and confirm the Redis container shows as healthy.
- If you see a ModuleNotFoundError for redis, confirm your virtual environment is activated and that you installed the dependencies from requirements.txt.
- Make sure you are running the command from your project root directory (the folder containing store/).
Why can't my script connect to Redis?
Build the Kafka producer and consumer
The streaming pipeline has two parts. A producer simulates live ad-click events and publishes them to a Kafka topic. A consumer reads from that topic, computes real-time aggregation features, and writes them to Redis.
- Create a new file called stream_pipeline.py inside the pipelines/ directory.
- Add the imports and the EventProducer class by pasting this code:
import json
import time
import random
from datetime import datetime, timezone
from confluent_kafka import Producer, Consumer, KafkaError
from store.online_store import RedisOnlineStore
class EventProducer:
def __init__(self, bootstrap_servers: str = "localhost:9092"):
self.producer = Producer({"bootstrap.servers": bootstrap_servers})
self.topic = "ad-events"
def produce_event(self, event: dict):
self.producer.produce(
self.topic,
key=event["user_id"].encode("utf-8"),
value=json.dumps(event).encode("utf-8"),
)
self.producer.flush()
def simulate_events(self, num_events: int = 100):
user_ids = [f"user_{i}" for i in range(1, 21)]
ad_ids = [f"ad_{i}" for i in range(1, 51)]
event_types = ["click", "impression", "impression", "impression"]
for _ in range(num_events):
event = {
"user_id": random.choice(user_ids),
"ad_id": random.choice(ad_ids),
"event_type": random.choice(event_types),
"session_duration": round(random.uniform(0.5, 15.0), 2),
"timestamp": datetime.now(timezone.utc).isoformat(),
}
self.produce_event(event)
time.sleep(0.05)What does the EventProducer do?
- produce_event() serializes an event as JSON and publishes it to the ad-events Kafka topic, keyed by user_id so all events for the same user land on the same partition.
- simulate_events() generates synthetic ad interactions across 20 users and 50 ads. The event type distribution (1 click to 3 impressions) mimics realistic click-through rates.
- Now add the FeatureConsumer class below the EventProducer class in the same file:
class FeatureConsumer:
def __init__(self, bootstrap_servers: str = "localhost:9092"):
self.consumer = Consumer({
"bootstrap.servers": bootstrap_servers,
"group.id": "feature-pipeline",
"auto.offset.reset": "earliest",
})
self.consumer.subscribe(["ad-events"])
self.online_store = RedisOnlineStore()
self.user_clicks: dict[str, int] = {}
self.user_impressions: dict[str, int] = {}
def process_events(self, max_messages: int = 200, timeout: float = 1.0):
processed = 0
while processed < max_messages:
msg = self.consumer.poll(timeout)
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
break
event = json.loads(msg.value().decode("utf-8"))
self._update_features(event)
processed += 1
self.consumer.close()
print(f"Processed {processed} events, materialized features for {len(self.user_clicks)} users")
def _update_features(self, event: dict):
user_id = event["user_id"]
event_type = event["event_type"]
if event_type == "click":
self.user_clicks[user_id] = self.user_clicks.get(user_id, 0) + 1
self.user_impressions[user_id] = self.user_impressions.get(user_id, 0) + 1
click_count = self.user_clicks.get(user_id, 0)
impression_count = self.user_impressions.get(user_id, 0)
ctr = click_count / impression_count if impression_count > 0 else 0.0
features = {
"user_click_count_7d": float(click_count),
"ad_ctr_score": round(ctr, 4),
}
self.online_store.write_features(user_id, features)What does the FeatureConsumer do?
- The consumer subscribes to the ad-events topic with auto.offset.reset: earliest, meaning it reads all available messages from the beginning of the topic.
- _update_features() maintains running counters of clicks and impressions per user. After each event, it recomputes the click-through rate (CTR) and writes both user_click_count_7d and ad_ctr_score to Redis.
- process_events() polls Kafka in a loop, gracefully handling partition-end signals and stopping after max_messages are consumed.
- Save pipelines/stream_pipeline.py.
Seeing an import error for confluent_kafka?
- Confirm your virtual environment is activated. Run pip show confluent-kafka to check it is installed.
- On some systems, confluent-kafka requires the librdkafka C library. The pip wheel bundles it for most platforms, but if you see a build error, check the confluent-kafka installation guide for your OS.
Help me fix my confluent-kafka import error.
✔️ Awesome, I've got everything!
Great. Make sure you have saved both store/online_store.py and pipelines/stream_pipeline.py.
ⓧ I'd like to double check the full code
Here is the complete store/online_store.py:
import os
import redis
class RedisOnlineStore:
def __init__(self):
redis_url = os.getenv("UPSTASH_REDIS_URL")
if redis_url:
self.client = redis.from_url(
redis_url,
password=os.getenv("UPSTASH_REDIS_TOKEN"),
decode_responses=True,
)
else:
self.client = redis.Redis(
host="localhost", port=6379, decode_responses=True
)
def write_features(self, entity_id: str, features: dict):
key = f"features:{entity_id}"
self.client.hset(key, mapping={k: str(v) for k, v in features.items()})
def read_features(self, entity_id: str, feature_names: list[str]) -> dict:
key = f"features:{entity_id}"
if feature_names:
values = self.client.hmget(key, feature_names)
return {
name: float(val) if val is not None else None
for name, val in zip(feature_names, values)
}
else:
all_features = self.client.hgetall(key)
return {k: float(v) for k, v in all_features.items()}
def health_check(self) -> bool:
try:
return self.client.ping()
except Exception:
return FalseHere is the complete pipelines/stream_pipeline.py:
import json
import time
import random
from datetime import datetime, timezone
from confluent_kafka import Producer, Consumer, KafkaError
from store.online_store import RedisOnlineStore
class EventProducer:
def __init__(self, bootstrap_servers: str = "localhost:9092"):
self.producer = Producer({"bootstrap.servers": bootstrap_servers})
self.topic = "ad-events"
def produce_event(self, event: dict):
self.producer.produce(
self.topic,
key=event["user_id"].encode("utf-8"),
value=json.dumps(event).encode("utf-8"),
)
self.producer.flush()
def simulate_events(self, num_events: int = 100):
user_ids = [f"user_{i}" for i in range(1, 21)]
ad_ids = [f"ad_{i}" for i in range(1, 51)]
event_types = ["click", "impression", "impression", "impression"]
for _ in range(num_events):
event = {
"user_id": random.choice(user_ids),
"ad_id": random.choice(ad_ids),
"event_type": random.choice(event_types),
"session_duration": round(random.uniform(0.5, 15.0), 2),
"timestamp": datetime.now(timezone.utc).isoformat(),
}
self.produce_event(event)
time.sleep(0.05)
class FeatureConsumer:
def __init__(self, bootstrap_servers: str = "localhost:9092"):
self.consumer = Consumer({
"bootstrap.servers": bootstrap_servers,
"group.id": "feature-pipeline",
"auto.offset.reset": "earliest",
})
self.consumer.subscribe(["ad-events"])
self.online_store = RedisOnlineStore()
self.user_clicks: dict[str, int] = {}
self.user_impressions: dict[str, int] = {}
def process_events(self, max_messages: int = 200, timeout: float = 1.0):
processed = 0
while processed < max_messages:
msg = self.consumer.poll(timeout)
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
break
event = json.loads(msg.value().decode("utf-8"))
self._update_features(event)
processed += 1
self.consumer.close()
print(f"Processed {processed} events, materialized features for {len(self.user_clicks)} users")
def _update_features(self, event: dict):
user_id = event["user_id"]
event_type = event["event_type"]
if event_type == "click":
self.user_clicks[user_id] = self.user_clicks.get(user_id, 0) + 1
self.user_impressions[user_id] = self.user_impressions.get(user_id, 0) + 1
click_count = self.user_clicks.get(user_id, 0)
impression_count = self.user_impressions.get(user_id, 0)
ctr = click_count / impression_count if impression_count > 0 else 0.0
features = {
"user_click_count_7d": float(click_count),
"ad_ctr_score": round(ctr, 4),
}
self.online_store.write_features(user_id, features)Test the full streaming loop
With both the producer and consumer ready, you can now run the complete pipeline end-to-end. Events flow from the producer into Kafka, the consumer picks them up, computes features, and writes them to Redis.
- First, run the producer to push 100 simulated events into Kafka by running:
python -c "from pipelines.stream_pipeline import EventProducer; p = EventProducer(); p.simulate_events(100); print('Produced 100 events to ad-events topic')"This takes about 5 seconds (a small delay between each event). You should see the confirmation message once all events are published.
- Now run the consumer to process those events and materialize features into Redis:
python -c "from pipelines.stream_pipeline import FeatureConsumer; c = FeatureConsumer(); c.process_events(max_messages=100)"You should see output similar to: Processed 100 events, materialized features for 20 users. The exact user count may vary since events are randomly distributed.
- Verify features landed in Redis by reading them back for a sample user:
python -c "from store.online_store import RedisOnlineStore; store = RedisOnlineStore(); print(store.read_features('user_1', []))"You should see a dictionary with user_click_count_7d and ad_ctr_score values for that user.
What just happened?
You ran the exact same pipeline architecture that powers real-time ML features at scale. Events arrived in Kafka, a consumer computed aggregation features on the fly, and those features were immediately available in Redis for model serving.
In production, the producer would be your ad platform publishing real user interactions. The consumer runs continuously, keeping Redis in sync with the latest user behavior.
Consumer stuck or showing 0 events processed?
- Make sure you ran the producer first. The consumer reads from the topic, so events must exist before the consumer starts.
- Check that Kafka is running by running docker compose ps. The kafka container should show as running.
- If the consumer hangs and never finishes, it may be waiting for more messages to reach max_messages. Press Ctrl+C to stop it, then re-run with a smaller max_messages value (e.g. 50).
Why is my Kafka consumer not receiving any messages?
Your feature platform now has both batch and real-time paths. The batch pipeline handles historical feature computation, and the streaming pipeline keeps features fresh as new events arrive. Next up, you will build the serving API that exposes these features to ML models with drift monitoring.
Build the Feature Serving API and Drift Detection
Your feature platform can now compute features in batch and in real time. The next challenge is serving those features to ML models in production with the right guarantees.
In this step, you will build a FastAPI serving layer that routes requests to either the online store (Redis) or the offline store (DuckDB) depending on whether the caller needs real-time features or historical point-in-time features. You will also add drift detection so the platform can alert when feature distributions shift.
In this step, get ready to:
- Build a FastAPI serving API with online and offline feature retrieval.
- Implement drift detection using Population Stability Index (PSI).
- Deploy the serving API live on Render with Upstash Redis as the online store.
Build the FastAPI serving endpoint
The serving API is the interface between your feature platform and the ML models that consume features. It needs two retrieval modes: low-latency online reads from Redis for inference, and point-in-time offline reads from DuckDB for training.
- Create a new file called serving/app.py in your project. Add the imports and app initialization by pasting this code:
import os
from datetime import datetime
from fastapi import FastAPI, Query
from dotenv import load_dotenv
load_dotenv()
from registry.feature_registry import FeatureRegistry
from store.online_store import RedisOnlineStore
from pipelines.batch_pipeline import BatchFeaturePipeline
app = FastAPI(title="ML Feature Platform", version="1.0.0")
# Initialize platform components
registry = FeatureRegistry()
registry.load_definitions()
online_store = RedisOnlineStore()
batch_pipeline = BatchFeaturePipeline()What does this code do?
This sets up the FastAPI app and initializes the three core components the serving layer depends on: the feature registry (to know which features exist), the online store (Redis for real-time reads), and the batch pipeline (DuckDB for historical reads).
The load_dotenv() call loads your .env file so the RedisOnlineStore can find Upstash credentials when deployed.
- Below the initialization code in the same file, add the main feature retrieval endpoint:
@app.get("/features/{entity_id}")
def get_features(
entity_id: str,
feature_names: str = Query(default="", description="Comma-separated feature names"),
timestamp: str = Query(default=None, description="ISO timestamp for point-in-time retrieval"),
):
names = [n.strip() for n in feature_names.split(",") if n.strip()]
# Route to offline store for historical queries, online store for real-time
if timestamp:
ts = datetime.fromisoformat(timestamp)
features = batch_pipeline.get_training_features(entity_id, names, ts)
source = "offline"
else:
features = online_store.read_features(entity_id, names)
source = "online"
return {
"entity_id": entity_id,
"features": features,
"source": source,
"timestamp": timestamp or "latest",
}What does this endpoint do?
This single endpoint handles both real-time and historical feature retrieval. When a caller passes a timestamp query parameter, the endpoint reads from DuckDB using point-in-time correct logic. This prevents data leakage when building training datasets.
When no timestamp is provided, the endpoint reads the latest feature values from Redis for low-latency model inference.
- Add the health check endpoint below the get_features function:
@app.get("/features/health")
def health():
return {
"status": "healthy",
"registry_features": len(registry.list_features()),
"online_store": online_store.health_check(),
}What does the health endpoint do?
The /features/health endpoint returns the number of registered features and whether Redis is reachable. This is the endpoint monitoring tools hit to confirm your platform is operational.
- Save serving/app.py.
- Test the serving API locally by running this command:
uvicorn serving.app:app --reload --port 8000- Open your browser and navigate to http://localhost:8000/features/user_1. You should see a JSON response with the features stored in Redis from the streaming pipeline.
Not seeing feature data?
Make sure your Docker containers are still running. The RedisOnlineStore connects to local Redis on port 6379.
If you see an import error, confirm you are running uvicorn from the project root directory (the folder containing the serving/, registry/, and store/ folders).
help me debug my FastAPI serving endpoint
- Stop the server with Ctrl+C when you have confirmed it works.
Implement drift detection with PSI
Feature drift happens when the distribution of feature values in production shifts away from what your model was trained on. Detecting drift early prevents silent model degradation.
- Create a new file called monitoring/drift_detector.py. Add the PSI computation function by pasting this code:
import numpy as np
from scipy.stats import entropy
def compute_psi(reference: list[float], current: list[float], bins: int = 10) -> float:
ref_array = np.array(reference)
curr_array = np.array(current)
# Create shared bin edges across both distributions
min_val = min(ref_array.min(), curr_array.min())
max_val = max(ref_array.max(), curr_array.max())
bin_edges = np.linspace(min_val, max_val, bins + 1)
# Histogram each distribution into the same bins
ref_hist, _ = np.histogram(ref_array, bins=bin_edges)
curr_hist, _ = np.histogram(curr_array, bins=bin_edges)
# Add smoothing to avoid division by zero
ref_pct = (ref_hist + 1) / (ref_hist.sum() + bins)
curr_pct = (curr_hist + 1) / (curr_hist.sum() + bins)
# PSI formula: sum of (current - reference) * ln(current / reference)
psi = np.sum((curr_pct - ref_pct) * np.log(curr_pct / ref_pct))
return float(psi)What is Population Stability Index?
PSI compares two distributions by binning both into the same histogram and measuring how much the proportions shifted. A PSI below 0.1 means the distributions are stable. A PSI above 0.1 signals meaningful drift that could degrade model performance.
The +1 smoothing prevents log(0) errors when a bin has zero observations in one distribution but not the other.
- Below the compute_psi function in the same file, add the DriftMonitor class:
class DriftMonitor:
def __init__(self, threshold: float = 0.1):
self.threshold = threshold
self.reference_distributions: dict[str, list[float]] = {}
def set_reference(self, feature_name: str, values: list[float]):
self.reference_distributions[feature_name] = values
def check_drift(self, feature_name: str, current_values: list[float]) -> dict:
if feature_name not in self.reference_distributions:
return {"feature": feature_name, "status": "no_reference", "psi": None}
reference = self.reference_distributions[feature_name]
psi = compute_psi(reference, current_values)
drifted = psi > self.threshold
return {
"feature": feature_name,
"psi": round(psi, 4),
"threshold": self.threshold,
"drifted": drifted,
"status": "drift_detected" if drifted else "stable",
}What does DriftMonitor do?
The DriftMonitor stores a baseline (reference) distribution for each feature. When you call check_drift, it computes PSI between the reference and the current values, then flags whether the threshold of 0.1 has been exceeded.
- Save monitoring/drift_detector.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved the file.
ⓧ I'd like to double check the full code
import numpy as np
from scipy.stats import entropy
def compute_psi(reference: list[float], current: list[float], bins: int = 10) -> float:
ref_array = np.array(reference)
curr_array = np.array(current)
min_val = min(ref_array.min(), curr_array.min())
max_val = max(ref_array.max(), curr_array.max())
bin_edges = np.linspace(min_val, max_val, bins + 1)
ref_hist, _ = np.histogram(ref_array, bins=bin_edges)
curr_hist, _ = np.histogram(curr_array, bins=bin_edges)
ref_pct = (ref_hist + 1) / (ref_hist.sum() + bins)
curr_pct = (curr_hist + 1) / (curr_hist.sum() + bins)
psi = np.sum((curr_pct - ref_pct) * np.log(curr_pct / ref_pct))
return float(psi)
class DriftMonitor:
def __init__(self, threshold: float = 0.1):
self.threshold = threshold
self.reference_distributions: dict[str, list[float]] = {}
def set_reference(self, feature_name: str, values: list[float]):
self.reference_distributions[feature_name] = values
def check_drift(self, feature_name: str, current_values: list[float]) -> dict:
if feature_name not in self.reference_distributions:
return {"feature": feature_name, "status": "no_reference", "psi": None}
reference = self.reference_distributions[feature_name]
psi = compute_psi(reference, current_values)
drifted = psi > self.threshold
return {
"feature": feature_name,
"psi": round(psi, 4),
"threshold": self.threshold,
"drifted": drifted,
"status": "drift_detected" if drifted else "stable",
}Now wire the drift monitor into the serving API.
- In serving/app.py, find the line that says from pipelines.batch_pipeline import BatchFeaturePipeline.
- Add this import directly below it:
from monitoring.drift_detector import DriftMonitor- Find the line batch_pipeline = BatchFeaturePipeline().
- Add this line directly below it:
drift_monitor = DriftMonitor()- At the bottom of serving/app.py, add the drift detection endpoint:
@app.get("/features/drift")
def check_drift():
results = []
for feature_info in registry.list_features():
name = feature_info["name"]
if name in drift_monitor.reference_distributions:
# Collect current values from online store for all users
current_values = []
for i in range(1, 21):
val = online_store.read_features(f"user_{i}", [name]).get(name)
if val is not None:
current_values.append(val)
if current_values:
results.append(drift_monitor.check_drift(name, current_values))
return {"drift_report": results}What does the drift endpoint do?
The /features/drift endpoint iterates through every registered feature. For each feature that has a reference distribution set, it collects the current values from Redis across all 20 users and computes PSI against the baseline.
The response includes a drift report with PSI scores and a flag indicating whether each feature has drifted past the 0.1 threshold.
- Save serving/app.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved the file.
ⓧ I'd like to double check the full code
import os
from datetime import datetime
from fastapi import FastAPI, Query
from dotenv import load_dotenv
load_dotenv()
from registry.feature_registry import FeatureRegistry
from store.online_store import RedisOnlineStore
from pipelines.batch_pipeline import BatchFeaturePipeline
from monitoring.drift_detector import DriftMonitor
app = FastAPI(title="ML Feature Platform", version="1.0.0")
registry = FeatureRegistry()
registry.load_definitions()
online_store = RedisOnlineStore()
batch_pipeline = BatchFeaturePipeline()
drift_monitor = DriftMonitor()
@app.get("/features/{entity_id}")
def get_features(
entity_id: str,
feature_names: str = Query(default="", description="Comma-separated feature names"),
timestamp: str = Query(default=None, description="ISO timestamp for point-in-time retrieval"),
):
names = [n.strip() for n in feature_names.split(",") if n.strip()]
if timestamp:
ts = datetime.fromisoformat(timestamp)
features = batch_pipeline.get_training_features(entity_id, names, ts)
source = "offline"
else:
features = online_store.read_features(entity_id, names)
source = "online"
return {
"entity_id": entity_id,
"features": features,
"source": source,
"timestamp": timestamp or "latest",
}
@app.get("/features/health")
def health():
return {
"status": "healthy",
"registry_features": len(registry.list_features()),
"online_store": online_store.health_check(),
}
@app.get("/features/drift")
def check_drift():
results = []
for feature_info in registry.list_features():
name = feature_info["name"]
if name in drift_monitor.reference_distributions:
current_values = []
for i in range(1, 21):
val = online_store.read_features(f"user_{i}", [name]).get(name)
if val is not None:
current_values.append(val)
if current_values:
results.append(drift_monitor.check_drift(name, current_values))
return {"drift_report": results}Deploy the serving API to Render
Your API works locally. The next step is to package it in a Docker container and deploy it to Render so interviewers can hit a live endpoint backed by Upstash Redis.
- Create a file called Dockerfile in your project root with this content:
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "serving.app:app", "--host", "0.0.0.0", "--port", "10000"]Why port 10000?
Render expects web services to listen on port 10000 by default for Docker deployments. The CMD instruction starts uvicorn bound to 0.0.0.0:10000 so Render can route traffic to your container.
- Create a file called render.yaml in your project root with this content:
services:
- type: web
name: ml-feature-platform
runtime: docker
plan: free
envVars:
- key: UPSTASH_REDIS_URL
sync: false
- key: UPSTASH_REDIS_TOKEN
sync: falseWhat does sync: false mean?
The sync: false setting tells Render to prompt you for these values in the dashboard rather than hardcoding them in the YAML. This keeps your Upstash credentials out of version control.
- Save both files.
Seeing a port mismatch error on Render?
Make sure your Dockerfile CMD uses port 10000 and not 8000. Render routes incoming traffic to port 10000 by default.
help me fix my Dockerfile port configuration for Render
Now push your project to GitHub and deploy on Render.
- Create a new repository on GitHub (public or private).
- Initialize git and push your project by running these commands:
git init
git add .
git commit -m "ML feature platform with serving API"
git branch -M main
git remote add origin [[YOUR_REPO_URL="https://github.com/your-username/ml-feature-platform.git"]]
git push -u origin mainExclude local files from git
Create a .gitignore file before committing. Add .env, feature_store.db, __pycache__/, and venv/ to prevent pushing secrets and local artifacts to GitHub.
- Open https://render.com and sign in (or create a free account).
- In the Render Dashboard, click + New in the top-right corner and select Web Service.
- Connect your GitHub account if you have not already.
- Select the repository you just pushed and click Connect.
- Set the Language field to Docker.
- Choose the Free instance type.
- Click Create Web Service.
Render will start building your Docker image. While it builds, add your Upstash credentials.
- In the Render Dashboard, select your new service.
- Click Environment in the left pane.
- Under Environment Variables, click + Add Environment Variable.
- Add UPSTASH_REDIS_URL with the URL from your Upstash dashboard.
- Add UPSTASH_REDIS_TOKEN with the token from your Upstash dashboard.
- Click Save Changes.
How long does the first deploy take?
The first build typically takes 2-3 minutes as Render installs all Python dependencies. Subsequent deploys are faster because of Docker layer caching.
The free tier spins down after 15 minutes of inactivity. The first request after idle takes about 1 minute to respond while the container restarts.
- Once the deploy finishes, find your service URL in the Render Dashboard. It will look like https://ml-feature-platform-xxxx.onrender.com.
- Verify the live endpoint by running this command (replace the URL with your actual Render URL):
curl [[RENDER_URL="https://your-service.onrender.com"]]/features/health- You should see a JSON response showing the service is healthy and the online store is connected.
Deploy failing on Render?
Check the build logs in the Render Dashboard. The most common issues are a missing file in the repository (make sure all folders like registry/, store/, monitoring/ are committed) or an incorrect requirements.txt path.
If the health endpoint returns online_store: false, double check your UPSTASH_REDIS_URL and UPSTASH_REDIS_TOKEN environment variables in the Render dashboard.
help me debug my Render deployment
Your feature platform is now live. Next up, you will build an agentic feature discovery system that uses your local LLM to automatically propose new features from raw data.
Secret mission
Build an Agentic Feature Discovery Agent
Your feature platform handles storage, serving, and drift monitoring. But what about discovering NEW features? In production ML teams, data scientists spend weeks manually exploring raw data for predictive signals. What if your platform could do that automatically? In this secret mission, you'll build an AI agent that analyzes raw event data, proposes new features, evaluates their quality, and writes the best ones directly into your feature registry.
Clean Up Your Resources
Clean Up Your Resources
This project uses free-tier cloud services and local Docker containers. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Upstash Redis database (free tier, 256MB storage, 500,000 commands/month).
- Render web service (free tier, 750 instance hours/month, spins down after 15 minutes of inactivity).
- Local Docker containers (Redis + Kafka via Confluent Platform).
- Local project files and Python virtual environment.
Will I be charged?
Both Upstash Redis and Render are on free tiers with no credit card required. You will not be charged unless you manually upgrade your plan or exceed the free-tier limits.
Upstash free-tier databases are automatically archived after 30 days of inactivity. Render free-tier services spin down after 15 minutes of inactivity and cost nothing while idle.
Keep everything running
No action needed. Choose this if you are still actively building or want to keep your live API as a portfolio demo for job applications.
- Your Render service stays deployed and accessible at your Render URL.
- Your Upstash Redis database remains available as the online feature store.
- Your local Docker containers continue running until you stop them or restart your machine.
- LM Studio remains installed on your machine.
Pause - I'll come back to this later
Stop local Docker containers to free up memory and CPU. Render and Upstash stay free and idle with no action required.
- Stop and remove your local Docker containers by running:
docker compose down- Your Render service automatically spins down after 15 minutes of inactivity. No action needed.
- Your Upstash Redis database stays idle on the free tier. No action needed.
- You can restart everything later with docker compose up -d from your project folder.
Delete - I don't want to use this again
Remove all project resources and start fresh. Follow each step below to fully clean up.
- Delete your Render web service:
- Open the Render Dashboard and click on your ml-feature-platform service.
- Click Settings in the left sidebar.
- Scroll to the bottom of the Settings page and click Delete Web Service.
- Confirm the deletion when prompted.
- Delete your Upstash Redis database:
- Open the Upstash Console and select your Redis database.
- Scroll to the bottom of the Details page.
- Click Delete this database and follow the confirmation instructions.
- Remove local Docker containers and volumes:
docker compose down -v- Delete your local project files:
cd ~/Desktop
rm -rf ml-feature-platform- Deactivate and remove your Python virtual environment by running:
deactivateThat's a Wrap!
That's a Wrap!
Nice work. You just built and deployed a production-style ML feature platform that ingests raw event data, transforms it into versioned features through batch and real-time pipelines, and serves them through a live API with drift monitoring.
You've learned how to:
- Build an ML feature platform with a declarative YAML registry, a batch pipeline (pandas + DuckDB) that enforces point-in-time correctness, and a real-time streaming pipeline using Apache Kafka that materializes features to Redis.
- Create a FastAPI serving API with both online (Redis) and offline (DuckDB) feature retrieval, plus PSI-based drift detection for feature governance and health monitoring.
- Deploy the feature serving API live on Render with Upstash Redis as the online store, giving you a public endpoint that interviewers can hit directly.
- Secret Mission: Built an agentic feature discovery system using a local LLM that automatically proposes new features from raw data, evaluates their quality, and writes top-scoring proposals to the registry.
Ready to quiz yourself?