Feast Feature Store: Offline/Online Feature Management

Defining Features & the Feature Registry

Install with pip install feast[redis,gcp], initialize a repo, then define features in Python:

# feast init my_feature_repo && cd my_feature_repo
# feature_repo/features.py

from datetime import timedelta
from feast import Entity, FeatureView, Field, FileSource, FeatureStore
from feast.types import Float32, Int64, String

# Entities are the primary keys of your data (user_id, item_id, etc.)
user = Entity(name="user_id", description="User entity")
item = Entity(name="item_id", description="Item entity")

# Data source — Parquet for local, BigQuery/Redshift for cloud
user_stats_source = FileSource(
    path="data/user_stats.parquet",
    timestamp_field="event_timestamp",    # required for point-in-time joins
)

# FeatureView — defines a group of features from a single source
user_stats_view = FeatureView(
    name="user_statistics",
    entities=[user],
    ttl=timedelta(days=7),                # max feature age
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_order_value_30d", dtype=Float32),
        Field(name="days_since_last_purchase", dtype=Int64),
        Field(name="preferred_category", dtype=String),
    ],
    source=user_stats_source,
    tags={"team": "recommendations", "version": "v2"},
)

# Apply to registry: feast apply

Offline Store: Point-in-Time Training Data

Generate training datasets with correct historical feature values at each event's timestamp:

import pandas as pd
from feast import FeatureStore
from datetime import datetime

store = FeatureStore(repo_path=".")

# Entity DataFrame: for each (user_id, event_timestamp), retrieve features
# as they existed AT that timestamp (no future leakage)
entity_df = pd.DataFrame({
    "user_id": [1001, 1002, 1003, 1004],
    "event_timestamp": [
        datetime(2024, 1, 15, 10, 0),
        datetime(2024, 2, 3, 14, 30),
        datetime(2024, 3, 22, 9, 15),
        datetime(2024, 4, 1, 11, 45),
    ],
    "label": [1, 0, 1, 0],    # your target variable
})

# Point-in-time join — retrieves historical feature values for each row
training_data = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_statistics:purchase_count_7d",
        "user_statistics:avg_order_value_30d",
        "user_statistics:days_since_last_purchase",
        "user_statistics:preferred_category",
    ],
).to_df()

print(training_data.head())
# user_id | event_timestamp | purchase_count_7d | avg_order_value_30d | label
# 1001    | 2024-01-15      | 3                 | 45.20               | 1

Online Store: Low-Latency Feature Serving

Materialize features to Redis, then serve at sub-millisecond latency in production:

from feast import FeatureStore
import subprocess
from datetime import datetime

store = FeatureStore(repo_path=".")

# Materialize features to online store (schedule as cron job)
# CLI: feast materialize-incremental $(date -u +%Y-%m-%dT%H:%M:%S)
store.materialize_incremental(end_date=datetime.utcnow())

# --- In your serving code (FastAPI, gRPC, etc.) ---
from fastapi import FastAPI

app = FastAPI()

@app.post("/predict")
async def predict(user_id: int):
    # Retrieve latest features from Redis — ~0.5ms latency
    feature_vector = store.get_online_features(
        features=[
            "user_statistics:purchase_count_7d",
            "user_statistics:avg_order_value_30d",
            "user_statistics:days_since_last_purchase",
        ],
        entity_rows=[{"user_id": user_id}],
    ).to_dict()

    # Identical features to what the model saw during training
    X = [[
        feature_vector["purchase_count_7d"][0],
        feature_vector["avg_order_value_30d"][0],
        feature_vector["days_since_last_purchase"][0],
    ]]
    prediction = model.predict(X)[0]
    return {"user_id": user_id, "prediction": float(prediction)}

Feast vs Tecton vs Hopsworks vs Vertex Feature Store