ML Pipelines and Feature Stores

A restaurant kitchen has three pipelines: the supply pipeline (ingredients arrive, are prepped, and stored in the walk-in fridge), the recipe pipeline (chefs create and test new dishes using stored ingredients), and the serving pipeline (orders come in, ingredients are pulled from storage, dishes are prepared and served). The walk-in fridge is the feature store. If the supply pipeline changes how the vegetables are cut (feature engineering), the recipes (training) and serving (inference) must use the same cut — or the dish comes out wrong.

Training-serving skew — when features are computed differently at training time vs inference time — is the number one silent failure mode in production ML. Feature stores exist specifically to prevent it by computing features once and serving the same values to both pipelines.

# pip install feast pandas pyarrow

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from pathlib import Path
import os, tempfile

# ── Simulate DoorDash delivery time prediction feature store ────────────

# ── Step 1: Define feature views (the schema) ─────────────────────────
# In a real Feast setup these are in feature_store.yaml + feature_repo/
# Here we show the concepts with pandas to illustrate the ideas

# Raw data — transactions from the data warehouse
raw_orders = pd.DataFrame({
    'order_id':        [f'ORD{i:04d}' for i in range(1000)],
    'restaurant_id':   np.random.choice(['RST001', 'RST002', 'RST003'], 1000),
    'driver_id':       np.random.choice([f'DRV{i:02d}' for i in range(20)], 1000),
    'event_timestamp': pd.date_range('2026-01-01', periods=1000, freq='1h'),
    'delivery_time_min': np.random.normal(35, 10, 1000).clip(10, 90),
    'distance_km':     np.random.exponential(3, 1000),
    'order_value':     np.random.exponential(400, 1000),
    'is_peak_hour':    np.random.randint(0, 2, 1000),
})

# ── Step 2: Compute features ──────────────────────────────────────────
def compute_restaurant_features(orders: pd.DataFrame) -> pd.DataFrame:
    """
    Aggregate features per restaurant — recomputed daily.
    These represent what a feature pipeline would compute.
    """
    feats = orders.groupby('restaurant_id').agg(
        restaurant_avg_delivery_time=('delivery_time_min', 'mean'),
        restaurant_7d_order_count=('order_id', 'count'),
        restaurant_acceptance_rate=('is_peak_hour', 'mean'),  # proxy
    ).reset_index()
    feats['feature_timestamp'] = datetime.now()
    return feats

def compute_driver_features(orders: pd.DataFrame) -> pd.DataFrame:
    """Aggregate features per driver."""
    feats = orders.groupby('driver_id').agg(
        driver_avg_delivery_time=('delivery_time_min', 'mean'),
        driver_completed_orders=('order_id', 'count'),
        driver_avg_distance=('distance_km', 'mean'),
    ).reset_index()
    feats['feature_timestamp'] = datetime.now()
    return feats

restaurant_features = compute_restaurant_features(raw_orders)
driver_features     = compute_driver_features(raw_orders)

print("Feature store contents:")
print(f"  Restaurant features: {restaurant_features.shape}")
print(f"  Driver features:     {driver_features.shape}")
print(f"
Sample restaurant features:")
print(restaurant_features.head(3).to_string(index=False))

# ── Step 3: Point-in-time correct training dataset ────────────────────
def create_training_dataset(
    orders: pd.DataFrame,
    restaurant_feats: pd.DataFrame,
    driver_feats: pd.DataFrame,
    cutoff_date: datetime,
) -> pd.DataFrame:
    """
    Create a training dataset with point-in-time correct features.
    Only use features available as of cutoff_date — no future data.
    """
    # Only include orders before the cutoff
    train_orders = orders[orders['event_timestamp'] < cutoff_date].copy()

    # Only use features computed from data before the cutoff
    # In a real feature store (Feast), this is a time-travel query
    hist_orders = orders[orders['event_timestamp'] < cutoff_date]
    hist_rest_feats = compute_restaurant_features(hist_orders)
    hist_drv_feats  = compute_driver_features(hist_orders)

    # Join features to training events
    dataset = train_orders.merge(hist_rest_feats, on='restaurant_id', how='left')
    dataset = dataset.merge(hist_drv_feats, on='driver_id', how='left')

    return dataset

cutoff = datetime(2026-03-1)
train_df = create_training_dataset(raw_orders, restaurant_features,
                                    driver_features, cutoff)
print(f"
Training dataset (before {cutoff.date()}):")
print(f"  Rows:    {len(train_df)}")
print(f"  Columns: {list(train_df.columns)}")

# ── Step 4: Online serving — retrieve latest features for a request ───
ONLINE_STORE = {
    'RST001': {'restaurant_avg_delivery_time': 33.2, 'restaurant_7d_order_count': 245},
    'RST002': {'restaurant_avg_delivery_time': 38.7, 'restaurant_7d_order_count': 189},
    'DRV01':  {'driver_avg_delivery_time': 31.5, 'driver_completed_orders': 892},
}

def get_online_features(restaurant_id: str, driver_id: str) -> dict:
    """Real-time feature retrieval — Redis lookup in production (~1ms)."""
    features = {}
    features.update(ONLINE_STORE.get(restaurant_id, {}))
    features.update(ONLINE_STORE.get(driver_id, {}))
    return features

features = get_online_features('RST001', 'DRV01')
print(f"
Online features for RST001/DRV01:")
for k, v in features.items():
    print(f"  {k}: {v}")

# pip install prefect scikit-learn pandas

from prefect import task, flow, get_run_logger
from prefect.tasks import task_input_hash
from datetime import timedelta, datetime
import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_absolute_error
import pickle, json, os

# ── Task: Extract features from feature store ─────────────────────────
@task(
    name='extract-features',
    retries=3,
    retry_delay_seconds=60,
    cache_key_fn=task_input_hash,   # cache by input hash — skip if same inputs
    cache_expiration=timedelta(hours=1),
)
def extract_features(cutoff_date: str) -> pd.DataFrame:
    logger = get_run_logger()
    logger.info(f"Extracting features up to {cutoff_date}")

    # In production: query feature store / data warehouse
    np.random.seed(42)
    n = 5000
    df = pd.DataFrame({
        'restaurant_avg_delivery_time':  np.random.normal(35, 8, n),
        'restaurant_7d_order_count':     np.random.randint(50, 500, n),
        'driver_avg_delivery_time':      np.random.normal(32, 7, n),
        'driver_completed_orders':       np.random.randint(100, 1000, n),
        'distance_km':                   np.random.exponential(3, n),
        'is_peak_hour':                  np.random.randint(0, 2, n),
        'order_value':                   np.random.exponential(400, n),
        'target_delivery_time':          np.random.normal(35, 10, n).clip(10, 90),
    })
    logger.info(f"Extracted {len(df)} rows, {df.shape[1]} columns")
    return df

# ── Task: Validate data quality ────────────────────────────────────────
@task(name='validate-data')
def validate_data(df: pd.DataFrame) -> dict:
    logger = get_run_logger()
    issues = []

    # Null checks
    null_rates = df.isnull().mean()
    for col, rate in null_rates.items():
        if rate > 0.05:
            issues.append(f"High null rate in {col}: {rate:.2%}")

    # Range checks
    if (df['target_delivery_time'] < 0).any():
        issues.append("Negative delivery times found")
    if (df['distance_km'] < 0).any():
        issues.append("Negative distances found")

    # Row count check
    if len(df) < 1000:
        issues.append(f"Too few rows: {len(df)} (minimum 1000)")

    result = {
        'passed':     len(issues) == 0,
        'row_count':  len(df),
        'issues':     issues,
        'null_rates': null_rates.to_dict(),
    }
    if issues:
        logger.warning(f"Data validation issues: {issues}")
    else:
        logger.info(f"Data validation passed: {len(df)} rows")
    return result

# ── Task: Train model ─────────────────────────────────────────────────
@task(name='train-model')
def train_model(df: pd.DataFrame) -> dict:
    logger = get_run_logger()

    feature_cols = [c for c in df.columns if c != 'target_delivery_time']
    X = df[feature_cols]
    y = df['target_delivery_time']

    X_train, X_val, y_train, y_val = train_test_split(
        X, y, test_size=0.2, random_state=42,
    )

    model = GradientBoostingRegressor(
        n_estimators=200, max_depth=4,
        learning_rate=0.05, random_state=42,
    )
    model.fit(X_train, y_train)

    train_mae = mean_absolute_error(y_train, model.predict(X_train))
    val_mae   = mean_absolute_error(y_val,   model.predict(X_val))

    logger.info(f"Train MAE: {train_mae:.2f} min  Val MAE: {val_mae:.2f} min")

    return {
        'model':      model,
        'train_mae':  train_mae,
        'val_mae':    val_mae,
        'feature_cols': feature_cols,
        'trained_at': datetime.now().isoformat(),
    }

# ── Task: Evaluate and gate ────────────────────────────────────────────
@task(name='evaluate-model')
def evaluate_model(result: dict,
                    max_val_mae: float = 8.0,
                    max_overfit_gap: float = 3.0) -> bool:
    logger = get_run_logger()
    val_mae  = result['val_mae']
    train_mae = result['train_mae']
    overfit   = train_mae - val_mae   # negative means val > train (fine)

    passed = val_mae <= max_val_mae and abs(overfit) <= max_overfit_gap

    logger.info(
        f"Quality gate: val_mae={val_mae:.2f} (max={max_val_mae}) "
        f"overfit_gap={overfit:.2f} (max={max_overfit_gap}) "
        f"→ {'PASS' if passed else 'FAIL'}"
    )
    return passed

# ── Task: Register model ───────────────────────────────────────────────
@task(name='register-model')
def register_model(result: dict, run_date: str) -> str:
    logger    = get_run_logger()
    model_dir = f'/tmp/models/delivery_time_{run_date}'
    os.makedirs(model_dir, exist_ok=True)

    model_path = f'{model_dir}/model.pkl'
    meta_path  = f'{model_dir}/metadata.json'

    with open(model_path, 'wb') as f:
        pickle.dump(result['model'], f)

    metadata = {
        'model_path':   model_path,
        'val_mae':      result['val_mae'],
        'train_mae':    result['train_mae'],
        'feature_cols': result['feature_cols'],
        'trained_at':   result['trained_at'],
        'registered_at': datetime.now().isoformat(),
        'version':      run_date,
    }
    with open(meta_path, 'w') as f:
        json.dump(metadata, f, indent=2)

    logger.info(f"Model registered at {model_path}")
    return model_path

# ── Flow: Assemble the pipeline ───────────────────────────────────────
@flow(name='delivery-time-training-pipeline')
def training_pipeline(run_date: str = None):
    logger   = get_run_logger()
    run_date = run_date or datetime.now().strftime('%Y%m%d')
    logger.info(f"Starting training pipeline for {run_date}")

    # Step 1: Extract
    df = extract_features(cutoff_date=run_date)

    # Step 2: Validate
    validation = validate_data(df)
    if not validation['passed']:
        logger.error(f"Data validation failed: {validation['issues']}")
        return {'status': 'failed', 'reason': 'data_validation'}

    # Step 3: Train
    result = train_model(df)

    # Step 4: Evaluate
    passed = evaluate_model(result)
    if not passed:
        logger.warning("Model failed quality gate — not registering")
        return {'status': 'rejected', 'val_mae': result['val_mae']}

    # Step 5: Register
    model_path = register_model(result, run_date)
    return {'status': 'success', 'model_path': model_path,
             'val_mae': result['val_mae']}

# ── Run locally ────────────────────────────────────────────────────────
print("Running training pipeline:")
result = training_pipeline(run_date='20260328')
print(f"
Pipeline result: {result}")

# pip install apache-airflow

from datetime import datetime, timedelta

# ── Airflow DAG for ML training pipeline ─────────────────────────────
# In production: save as ml_training_dag.py in your Airflow dags/ folder

DAG_CODE = '''
from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.email import EmailOperator
from airflow.utils.dates import days_ago
from datetime import timedelta
import logging

# Default arguments applied to all tasks
DEFAULT_ARGS = {
    "owner":            "ml-team",
    "depends_on_past":  False,
    "email":            ["ml-alerts@company.com"],
    "email_on_failure": True,
    "email_on_retry":   False,
    "retries":          2,
    "retry_delay":      timedelta(minutes=5),
    "execution_timeout": timedelta(hours=2),
}

def extract_features_fn(**context):
    """Extract and compute features for training."""
    run_date = context["ds"]   # execution date as YYYY-MM-DD
    logging.info(f"Extracting features for {run_date}")
    # ... feature extraction logic ...
    # Push result to XCom for downstream tasks
    context["ti"].xcom_push(key="row_count", value=50000)

def validate_data_fn(**context):
    """Validate feature quality — branch on result."""
    row_count = context["ti"].xcom_pull(
        task_ids="extract_features", key="row_count"
    )
    if row_count < 10000:
        return "send_data_alert"   # branch to alert task
    return "train_model"           # branch to training task

def train_model_fn(**context):
    run_date = context["ds"]
    logging.info(f"Training model for {run_date}")
    val_mae = 5.8   # would come from actual training
    context["ti"].xcom_push(key="val_mae", value=val_mae)

def evaluate_and_register_fn(**context):
    val_mae = context["ti"].xcom_pull(task_ids="train_model", key="val_mae")
    if val_mae > 8.0:
        logging.warning(f"Model failed quality gate: MAE={val_mae}")
        return
    logging.info(f"Registering model with MAE={val_mae}")
    # ... model registration ...

with DAG(
    dag_id="delivery_time_training",
    default_args=DEFAULT_ARGS,
    description="Daily retraining of DoorDash delivery time model",
    schedule_interval="0 2 * * *",   # 2 AM daily
    start_date=days_ago(1),
    catchup=False,
    tags=["ml", "delivery", "training"],
) as dag:

    extract = PythonOperator(
        task_id="extract_features",
        python_callable=extract_features_fn,
    )

    validate = BranchPythonOperator(
        task_id="validate_data",
        python_callable=validate_data_fn,
    )

    train = PythonOperator(
        task_id="train_model",
        python_callable=train_model_fn,
    )

    evaluate = PythonOperator(
        task_id="evaluate_and_register",
        python_callable=evaluate_and_register_fn,
    )

    alert = EmailOperator(
        task_id="send_data_alert",
        to="ml-team@company.com",
        subject="[ALERT] Training data validation failed",
        html_content="Feature pipeline produced insufficient data. Check sources.",
    )

    # DAG structure: extract → validate → train → evaluate
    #                                   ↘ alert (if validation fails)
    extract >> validate >> [train, alert]
    train   >> evaluate
'''

print("Airflow DAG structure:")
print("  extract_features")
print("       ↓")
print("  validate_data (branch)")
print("    ↙              ↘")
print("  train_model    send_alert")
print("       ↓")
print("  evaluate_and_register")

print("
Key Airflow concepts:")
concepts = [
    ('schedule_interval', '"0 2 * * *" = 2 AM daily. "0 */6 * * *" = every 6 hours.'),
    ('catchup=False',     'Do not backfill missed runs when DAG is enabled.'),
    ('retries=2',         'Retry failed tasks twice before marking as failed.'),
    ('XCom',              'Pass data between tasks via push/pull key-value store.'),
    ('BranchPython',      'Conditional logic — returns task_id(s) to run next.'),
    ('execution_timeout', 'Kill task if it runs too long — prevents zombie tasks.'),
]
for concept, detail in concepts:
    print(f"  {concept:<22}: {detail}")

# pip install feast

# ── Feast feature store setup ─────────────────────────────────────────
# In production: files live in a feature_repo/ directory

FEATURE_STORE_YAML = """
project: doordash_ml
registry: data/registry.db
provider: local
online_store:
  type: sqlite           # Redis in production
  path: data/online_store.db
offline_store:
  type: file             # BigQuery/Snowflake in production
"""

RESTAURANT_FEATURE_VIEW = """
from feast import Entity, Feature, FeatureView, FileSource, ValueType
from datetime import timedelta

# Entity: the thing we're computing features for
restaurant = Entity(
    name="restaurant_id",
    value_type=ValueType.STRING,
    description="Unique restaurant identifier",
)

# Data source: where the feature data lives
restaurant_source = FileSource(
    path="data/restaurant_features.parquet",
    event_timestamp_column="event_timestamp",
    created_timestamp_column="created_timestamp",
)

# Feature view: named collection of features for an entity
restaurant_features = FeatureView(
    name="restaurant_stats",
    entities=["restaurant_id"],
    ttl=timedelta(days=7),           # features expire after 7 days
    features=[
        Feature(name="avg_delivery_time_min", dtype=ValueType.FLOAT),
        Feature(name="order_count_7d",        dtype=ValueType.INT64),
        Feature(name="acceptance_rate",        dtype=ValueType.FLOAT),
        Feature(name="avg_order_value",        dtype=ValueType.FLOAT),
    ],
    online=True,                     # materialise to online store for serving
    source=restaurant_source,
)
"""

print("Feast feature store structure:")
print("""
feature_repo/
├── feature_store.yaml        ← registry, provider, online/offline store config
├── entities.py               ← entity definitions (restaurant, driver, user)
├── data_sources.py           ← where raw data lives (S3, BigQuery, Parquet)
└── feature_views.py          ← feature view definitions (what to compute)
""")

# ── Simulate Feast API patterns ───────────────────────────────────────
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# Training: point-in-time historical feature retrieval
def simulate_feast_historical_features(
    entity_df: pd.DataFrame,
    feature_view: str,
    features: list[str],
) -> pd.DataFrame:
    """
    In production:
    store = FeatureStore(repo_path='./feature_repo')
    training_df = store.get_historical_features(
        entity_df=entity_df,
        features=['restaurant_stats:avg_delivery_time_min', ...],
    ).to_df()

    Point-in-time correct: for each row in entity_df,
    return feature values as of entity_df['event_timestamp'].
    """
    np.random.seed(42)
    result = entity_df.copy()
    result['avg_delivery_time_min'] = np.random.normal(35, 8, len(entity_df))
    result['order_count_7d']        = np.random.randint(50, 500, len(entity_df))
    result['acceptance_rate']       = np.random.uniform(0.7, 0.99, len(entity_df))
    return result

# Inference: online feature retrieval — low latency
def simulate_feast_online_features(
    entity_rows: list[dict],
    features: list[str],
) -> dict:
    """
    In production:
    store = FeatureStore(repo_path='./feature_repo')
    feature_vector = store.get_online_features(
        features=['restaurant_stats:avg_delivery_time_min', ...],
        entity_rows=[{'restaurant_id': 'RST001'}],
    ).to_dict()

    Returns latest feature values from Redis — ~1ms latency.
    """
    return {
        'restaurant_id':              ['RST001'],
        'avg_delivery_time_min':      [33.2],
        'order_count_7d':             [245],
        'acceptance_rate':            [0.94],
    }

# Training: build entity dataframe with timestamps
entity_df = pd.DataFrame({
    'restaurant_id':   ['RST001', 'RST002', 'RST003'] * 100,
    'event_timestamp': pd.date_range('2026-01-01', periods=300, freq='1h'),
})

train_features = simulate_feast_historical_features(
    entity_df,
    'restaurant_stats',
    ['avg_delivery_time_min', 'order_count_7d'],
)
print(f"Training features shape: {train_features.shape}")

# Inference: get latest features for a prediction request
online_features = simulate_feast_online_features(
    [{'restaurant_id': 'RST001'}],
    ['avg_delivery_time_min', 'order_count_7d'],
)
print(f"
Online features for serving:")
for k, v in online_features.items():
    print(f"  {k}: {v}")

# Materialisation: write historical features to online store
print("""
# Materialise features to online store — run this daily via Airflow
# feast materialize 2026-03-27T00:00:00 2026-03-28T00:00:00
# OR in Python:
# store.materialize(
#     start_date=datetime(2026, 3, 27),
#     end_date=datetime(2026, 3, 28),
# )
# Copies latest feature values from offline store (Parquet/BigQuery)
# to online store (Redis) for low-latency serving
""")

# ── The decision most teams actually face before adopting a feature store ──
#
# "Do we need a feature store, or is a shared feature library enough?"
#
# A shared Python library (import compute_restaurant_features from a
# common package, call it in both the training notebook and the serving
# API) solves training/serving skew just as well AS LONG AS:
#   - one team owns both the training and serving code paths
#   - features do not need to be reused across other teams' models
#   - low-latency online lookups are not required (batch scoring is fine)
#
# A feature store earns its operational overhead once ANY of these hold:
#   - multiple teams want to reuse the same features (restaurant stats
#     used by the delivery-time model AND the fraud model AND the
#     recommendation model)
#   - real-time serving needs sub-10ms feature lookups
#   - feature definitions need a central registry so a change to one
#     feature doesn't silently break three other teams' models
#
# Most companies get the first cut wrong in one direction: adopting a
# full feature store platform for a single model with one consumer, or
# refusing to adopt one long after three teams are already hand-copying
# the same feature computation into three different notebooks.

A plain database gives you storage and lookups; it does not give you the specific capabilities that make a feature store worth building. Point-in-time correct joins for training data, a synchronised pair of an offline store and an online store that agree on values, a schema registry so feature definitions are shared rather than reinvented per team, and a materialisation pipeline that keeps the online copy fresh — none of that comes for free from Postgres or S3 alone. The feature store is the system built around a database (often two databases) specifically to solve training-serving consistency, which a database on its own has no concept of.

Having a timestamp column does not by itself prevent leakage — the join logic has to actually use it correctly, and it is easy to get subtly wrong. A common mistake is joining on the feature's created_timestamp instead of the event's event_timestamp, or forgetting that a feature's time-to-live means a value from ninety days ago should not be joined against a training event today. Real point-in-time correctness requires an as-of join that explicitly asks 'what was true at time T for this entity,' implemented deliberately — Feast's get_historical_features() does this work, but a manual pandas merge on entity ID alone, even with timestamps sitting right there in the columns, will silently leak future values if the join does not filter on them.

Reuse saves engineering time but creates a real coupling cost: every model that consumes a shared feature is now depending on its exact definition staying stable. Change how restaurant_avg_delivery_time is computed to fix a bug for the delivery-time model, and the fraud model and the recommendation model that also consume it can silently degrade without anyone on those teams touching a line of their own code. This is an ownership and versioning problem as much as a technical one — production feature stores need change review and consumer visibility (who depends on this feature) precisely because reuse is not the free lunch it looks like at the point of definition.

An online store being low-latency in principle does not mean every lookup against it is fast in practice. A cold-start entity that has not been materialised yet returns nothing or a stale default. Fetching many features across many entities in one request without batching the lookups turns a single 1ms Redis call into dozens of round trips. Connection pool exhaustion under load can turn a normally-fast store into the slowest part of the request. Treating 'it is in the online store' as equivalent to 'it will be fast' skips the actual engineering work of batching lookups, warming caches, and monitoring materialisation freshness that production latency budgets depend on.

What actually needs to match is the transformation logic that produces the values, not just the values that happen to be sitting in each store today. If the offline feature is computed by a nightly batch Spark job and the online feature is computed independently by a separate streaming job written by a different engineer, small differences in windowing, null handling, or aggregation order will produce different numbers for what is supposedly 'the same feature' — and this happens even inside systems that are correctly labelled a feature store. The materialisation pattern (compute once offline, copy the same values to the online store) exists specifically to close this gap; maintaining two independent computation paths for one feature reopens exactly the skew problem the feature store was built to prevent.

It happens whenever the code path that computes a feature for training diverges from the code path that computes it for serving — different aggregation windows, different null handling, different library versions, or simply two engineers implementing the 'same' feature independently in a notebook and in a serving API. Prevention means removing the duplication entirely: define each feature once, in one place, and have both training and serving read from that single definition — which is exactly what a feature store provides structurally. Where a full feature store is not yet in place, the minimum fix is a shared feature-computation library imported by both pipelines, never two independent implementations of the same logic.

A point-in-time join retrieves, for each training example at its own event timestamp, only the feature values that were actually available at that moment — not the latest values as of today. Without this, a training row from January can get joined against a feature value computed from March data, which means the model is trained on information that would not have existed yet at prediction time. The model looks excellent in offline evaluation because it is effectively looking at the future, then performs far worse in production once only present-moment features are available. Feast and similar tools implement this as an as-of join keyed on event_timestamp specifically to prevent that class of data leakage.

They are optimised for opposite access patterns. Training needs to scan large historical ranges across millions of rows, which is what columnar formats like Parquet or warehouses like BigQuery are built for, but a single key lookup against them can take seconds. Serving needs the opposite: a single entity's latest feature values in low single-digit milliseconds, which is what an in-memory or key-value store like Redis is built for, but storing years of history there is prohibitively expensive per gigabyte. Using one store for both jobs means being bad at one of them; the split exists purely to match each access pattern to the storage technology that is actually good at it, with a materialisation job syncing values from offline to online on a schedule.

I would start with the entities the organisation actually predicts about (restaurant, driver, user) and define feature views per entity with an explicit schema and a time-to-live. I would pick an offline store that matches where the raw data already lives (often the existing warehouse) and an online store optimised for low-latency key lookups, then build a materialisation job — scheduled through the same orchestrator already running the training pipelines — that copies fresh values from offline to online on a fixed cadence. Critically, I would build the point-in-time query path for training before worrying about anything else, since that is the capability a shared feature library cannot easily replicate and the reason most teams adopt a real feature store instead of one.

First, know who depends on it — a feature registry should make it possible to list every model consuming a given feature view before touching its definition. Then version it rather than mutating it in place: register the changed logic as a new feature view or a new version of the existing one, backfill it into the offline store, and let consuming teams migrate on their own schedule instead of finding out from a production incident. Only retire the old definition once every consumer has confirmed the migration. Silently changing a shared feature's computation in place is exactly the kind of hidden coupling that makes 'free' feature reuse expensive later.