ML Systems
Design

// Training on historical data: label occurred at time T
// Must retrieve feature values AS OF time T, not current values
// Using future features to predict the past = data leakage = inflated offline metrics

// WRONG — uses today's feature value to predict a past label:
features = feature_store.get(entity_id=user_123, feature="avg_session_30d")
// This returns the feature as it is TODAY — but the label was from 6 months ago
// The feature includes 6 months of future data. Model appears great offline, fails in prod.

// CORRECT — point-in-time correct retrieval:
features = feature_store.get(
    entity_id=user_123,
    feature="avg_session_30d",
    as_of=label_timestamp  ← retrieve value AS OF when the label occurred
)

// Feature store must store full history of feature values with timestamps
// Offline store schema:
// (entity_id, feature_name, value, event_timestamp, created_timestamp)
// Query: SELECT value WHERE entity_id=X AND feature_name=Y
//        AND event_timestamp <= label_timestamp ORDER BY event_timestamp DESC LIMIT 1

// Stage 1: DATA INGESTION
// Pull labeled examples from data warehouse, join with feature store
data = bigquery.query("SELECT user_id, item_id, label, event_time FROM interactions WHERE date=?")
features = feature_store.batch_get(entity_ids=data.user_ids, as_of=data.event_times)
dataset = join(data, features)

// Stage 2: DATA VALIDATION
// Catch data quality issues before training (fail fast)
validator.check_schema(dataset)           // expected columns present?
validator.check_distributions(dataset)    // feature values in expected range?
validator.check_null_rates(dataset)       // null rate < threshold per feature?
validator.check_label_balance(dataset)    // not extreme class imbalance?

// Stage 3: FEATURE PREPROCESSING (saved as artifact → identical at serving time)
preprocessor = Pipeline([
    StandardScaler(cols=["age", "session_duration"]),
    OneHotEncoder(cols=["device_type", "country"]),
    MeanImputer(cols=["avg_purchase_value"])
])
preprocessor.fit_transform(train_data)    // saved as artifact, applied at serving

// Stage 4: MODEL TRAINING
model = TwoTowerModel(user_dim=256, item_dim=256)
trainer = DistributedTrainer(model, gpus=8)   // Horovod / PyTorch DDP
trainer.train(train_data, epochs=10)

// Stage 5: EVALUATION (must beat champion model)
metrics = evaluator.eval(model, test_data)
if metrics.recall_at_100 <= champion.recall_at_100:
    pipeline.fail("New model does not beat champion. Stopping.")

// Stage 6: MODEL REGISTRATION
registry.register(model, {
    "recall@100": metrics.recall_at_100,
    "dataset_version": dataset.version,
    "training_code_hash": git.sha(),
    "status": "candidate"   // awaiting A/B test
})

// 1. SHADOW MODE — new model runs alongside old, results NOT shown to users
//    Log new model's predictions for offline analysis. Zero user impact.
user_response = champion_model.predict(features)    ← shown to user
shadow_result  = challenger_model.predict(features)  ← logged only
logger.log({"shadow_prediction": shadow_result, "actual_label": label})

// 2. CANARY — small % of real traffic to new model
if hash(user_id) % 100 < 5:    ← 5% canary
    return challenger_model.predict(features)
else:
    return champion_model.predict(features)

// 3. BLUE-GREEN — both versions hot, instant traffic switch
//    Keep v1 (blue) running. Deploy v2 (green). Validate green.
//    Switch load balancer: 100% → green. Keep blue for 1hr (fast rollback).
load_balancer.set_backend("green")   ← instant switch

// Automated rollback trigger:
if metrics.p99_latency > champion.p99_latency * 1.1:   ← 10% regression
    load_balancer.set_backend("blue")  ← instant rollback
    alerting.page("Challenger model rolled back: latency regression")

// STAGE 1: RETRIEVAL — find top-500 from 10M items in ~25ms
user_emb = user_tower.embed(user_features)     // 256-dim vector, ~5ms
candidates = ann_index.search(user_emb, k=500)  // ScaNN ANN search, ~15ms
// ANN index: 10M items × 256 dims × 4 bytes = ~10 GB → fits in RAM
// ScaNN achieves ~95% recall@100 at 10ms for 10M items

// STAGE 2: RANKING — score each of 500 candidates precisely
for item_id in candidates:
    item_features = feature_store.get(item_id)    // Redis batch fetch
    cross_features = compute_cross(user, item)    // interaction features
    score = ranking_model.predict(user_features, item_features, cross_features)

ranked = sorted(candidates, key=score, reverse=True)

// STAGE 3: POST-PROCESSING — business rules on top-100
final = post_processor.apply(ranked[:100], rules=[
    FilterWatched(user_id),         // don't show already-watched
    EnforceDiversity(max_per_topic=3), // not 10 Taylor Swift videos
    FreshnessBoost(hours=24),         // boost content <24h old
    SafetyFilter()                    // remove policy violations
])
return final[:20]  // top 20 to user

// Split by USER (not by request) — same user always sees same model
// Splitting by request = same user gets both models = contamination
def get_model(user_id):
    bucket = hash(user_id) % 100
    if bucket < 5:   ← 5% treatment
        return challenger_model
    else:              ← 95% control
        return champion_model

// Sample size calculation:
// baseline CTR = 2%, minimum detectable effect = 0.2% (10% relative lift)
// power = 80%, significance = 0.05
n_per_group = sample_size(
    baseline=0.02, mde=0.002, power=0.8, alpha=0.05
)  → ~156,000 users per group → ~312K total

// Primary metric: business metric (CTR, watch time, conversion rate)
// NOT offline AUC — model can improve AUC while hurting business metrics
// Guardrail metrics: must not regress (latency p99, revenue, crash rate)

// Peeking problem: DO NOT stop early because p < 0.05 after 3 days
// Each time you check, you increase false positive rate.
// Fix: pre-commit to run duration (2 weeks), use sequential testing if early stopping needed

// PSI = Population Stability Index
// Compares distribution of a feature between training (baseline) and production (current)
// Higher PSI = more drift

// Formula:
PSI = Σ (P_current_i - P_baseline_i) × ln(P_current_i / P_baseline_i)

// Where P_i = proportion of observations in bucket i (e.g., 10 equal-frequency buckets)

// Thresholds (industry standard):
PSI < 0.1   → No significant change, model stable
PSI 0.1–0.2 → Minor shift, investigate further
PSI > 0.2   → Significant shift, retrain model

// Monitoring stack (6 daily metrics to track):
1. PSI per feature           ← data drift
2. Prediction score dist     ← model output drift
3. Null / missing rate       ← pipeline health
4. Business metric (CTR)     ← concept drift proxy
5. Label distribution        ← label drift
6. Inference latency p99     ← serving health

// FEATURE COMPUTATION (daily Spark jobs):
user_features  = compute(watch_history_30d, search_history_7d, demographics)
video_features = compute(view_count, like_ratio, avg_watch_pct, transcript_emb)
// → Written to BigQuery (offline) + Bigtable (online, low-latency lookup)

// TRAINING DATA GENERATION:
positives = events WHERE watch_pct > 0.5        // user watched >50% of video
negatives = sample(shown_but_not_clicked, n=10) // 10 negatives per positive
dataset   = join_pit_correct(positives + negatives, feature_store)

// TWO-TOWER RETRIEVAL TRAINING:
// Goal: Recall@100 (are ground-truth videos in top-100 from 10M?)
// Pre-index: all 10M video embeddings → ScaNN index (~10 GB in RAM)

// RANKING MODEL TRAINING:
// Input: (user_emb, video_emb, cross_features, context)
// Output: predicted watch time (regression)
// Eval: NDCG@20 on held-out test set

// A/B TEST PROMOTION:
// New model → 5% canary → primary metric: watch time per session
// Guardrail: p99 latency must not exceed 110ms
// Run 2 weeks → evaluate → ramp 5% → 10% → 50% → 100%