A production-shaped ML feature platform built around one constraint: a model trained on feature values must receive the same feature values at inference, computed by the same code, joined at the correct point in time. No exceptions, no workarounds.
Training-serving skew is the leading cause of silent model degradation in production. It happens when the feature values used to train a model differ from those served at inference time. The gap has two common sources:
Separate compute paths. Teams write one transformation for offline training pipelines and a different one for online serving. The implementations drift. The model trains on one distribution and predicts on another.
Incorrect join timing. A label recorded at time T should be joined with the feature value that existed at time T, not the current value. Using a later value leaks information the model could not have had at prediction time, inflating offline metrics while production performance degrades.
Neither problem is caught by standard model evaluation. Both are invisible until the model fails in production.
One transform function, two runtimes. Every feature computation is a pure Python function in featureplatform/compute/transforms.py. The Flink streaming job and the offline backfill job both import and call the same function. Divergence between training and serving compute is structurally impossible: there is no online transform to drift.
Point-in-time correct training joins. The training join (training/pit_join.py) answers: "what was the value of feature F for entity E at time T?" It returns the most recent feature row whose business-event timestamp is at or before the label timestamp. It never uses the current value. Every join cell is audited for four correctness violations:
| Violation | Classification | Blocks training? |
|---|---|---|
| Feature timestamp after label timestamp | FUTURE_LEAKAGE | Yes |
| Row computed by a different version | VERSION_MISMATCH | Yes |
| Transform path changed within a version | HASH_MISMATCH | Yes |
| No feature data within the lookback window | STALENESS_MISS | No (uses default) |
dataset.validate() raises PITCorrectnessError on any hard violation before training begins.
Version and hash guards on every serving call. The online store stamps transform_hash and version on every write. Every get() call verifies both match the current registry spec. A stale deployment writing v1 values while the registry expects v2 raises SkewError, not a silent wrong prediction.
Skew monitor with rollback signal. The serving layer logs every feature vector it serves. The SkewMonitor compares that log against the offline training distribution using KS (distribution shape) and PSI (magnitude of shift). A rollback signal fires when PSI exceeds the critical threshold and the KS p-value confirms the shift is statistically significant. The monitor uses integer-aware binning for count features so sparse integer distributions produce correct PSI values, not false zeros.
Freshness SLA with two tiers. Each spec carries a FreshnessPolicy with a warn threshold and a critical threshold. Staleness is measured from the explicit computed_at_ms timestamp stamped at write time, not from Redis TTL jitter.
The platform separates the interfaces from the backends so each layer can be replaced independently.
Online store. The DictBackend used in tests implements the same three-method interface as RedisBackend (set, get, ttl). Swapping backends requires changing one constructor call. In production, Redis Cluster with hash-slot sharding distributes entity keys across nodes. The serving layer calls get_batch once per inference request; replacing the serial loop with a Redis pipeline reduces network round-trips from O(features) to O(1).
Offline store. The OfflineFeatureStore in CI uses a per-entity sorted index with bisect_right lookups (O(log k) per call). In production this maps to PyIceberg with partition pruning by event_ts_ms date. The training join interface is identical: latest_before(feature_name, version, entity_value, as_of_ts_ms). Swapping the backend changes no join logic.
Streaming ingest. The Flink job (ingest/flink_feature_job.py) is real PyFlink DDL. It uses Kafka as the event source, a 7-day tumbling window for aggregations, an Iceberg sink with exactly-once checkpoint semantics, and a Redis sink for low-latency serving. Scaling the job means adding Flink task manager slots; the Iceberg table is partitioned by date so historical joins do not scan the full dataset.
Skew monitor. SkewMonitor.run() takes numpy arrays. Building those arrays from the serving log and the offline store is a separate ETL step (an Airflow DAG in production) that runs on a schedule. The monitor itself is stateless and can run on any compute.
pip install -r requirements.txt
make test # 76 tests, no cluster needed, completes in under 2 seconds
make demo # end-to-end walkthrough: register, backfill, serve, PIT join, skew detectExpected demo output:
[1/6] Registering feature specs
registered: ['purchase_count_7d:1', 'avg_order_value_30d:1']
[2/6] Backfilling offline + online stores
backfilled 200 users x purchase_count_7d, 200 users x avg_order_value_30d
[3/6] Assembling PIT-correct training dataset
labels: 500, misses: 0, violations: CLEAN
PIT join validated: no leakage, no version mismatch
[leakage test] injecting 3 future-leaked labels
leakage violations detected: 6
[4/6] Serving feature vector for online inference
user 42 features: {'purchase_count_7d': 12, 'avg_order_value_30d': 112.79}
[5/6] Checking freshness SLAs
40 checks: 40 violations, 0 warnings
[6/6] Running skew monitor
[clean] 2 features checked | all stable
[skewed] 2 features checked | ROLLBACK triggered: ['purchase_count_7d']
rollback signal correctly triggered
The streaming stack (Kafka, Flink, Iceberg, Redis) runs in Docker:
make up # start all services
make down # tear down and remove volumessrc/featureplatform/
registry/ FeatureSpec + FeatureRegistry (immutable, versioned)
compute/ transforms.py: shared online and offline compute functions
store/
online/ Redis-backed online store with version and hash guards
offline/ Iceberg-backed offline store with indexed PIT lookup
training/ point_in_time_join() with four-violation classification
serving/ FeatureServingLayer with bounded circular sample log
monitor/ SkewMonitor (PSI + KS) and FreshnessMonitor
sla/ Per-feature freshness enforcement with two severity tiers
ingest/ Flink streaming job and offline backfill
tests/ 76 tests covering all correctness invariants, no cluster needed
docs/adr/ Architecture decision records for the join boundary and compute path
test_pit_join.py Join correctness: clean path, all four violations, boundary cases
test_skew_monitor.py PSI values, KS sensitivity, rollback trigger, edge cases
test_online_store.py Version guard, hash guard, staleness measurement, defaults
test_correctness.py Transforms, backfill, freshness SLA, serving layer, end-to-end
test_registry.py Immutability, version enforcement, duplicate registration
test_review_failures.py Regression tests: tie-breaking, missing keys, bounded log, O(log n) index