Feature store

Deep dive

How it works

Parts:

• registry — feature definitions: entity key, source, transformation, TTL, version
• offline store — append-only history of (entity, feature_ts, value) in columnar files or a warehouse, partitioned by date
• online store — key-value store (Redis, DynamoDB, Cassandra class) holding the latest value per entity key
• ingestion and materialization — batch jobs and stream jobs (windowed aggregates, change feeds) that write both stores from one definition
• serving API — get_online(entity keys, feature names) and get_historical(label rows, feature names)
• monitoring — freshness, null rate, drift against training data

1. Define. Register the feature once; both paths run this definition.
2. Backfill. Compute history into the offline store from raw events.
3. Training set. For each label row (entity, label_ts), take the newest value with feature_ts <= label_ts (and label_ts - feature_ts <= ttl); missing means null.
4. Materialize. On a schedule or per stream event, write the newest value per entity to the online store with its feature_ts.
5. Serve. Fetch all features for the request's entity keys in one batched read.
6. Log. Record served values with the request id: training data and a skew check.

raw events ─► transform (batch / stream) ─┬─► offline store: all versions ──► as-of join ──► training set
   (one registered definition)            │
                                          └─► materialize latest ─► online store ──► get_online ──► model
                                                  (entity -> value, feature_ts)   (one batched read per request)

Detail

• As-of join. Per entity, keep versions sorted by feature_ts; for a label time binary-search the last version at or before it. Bulk: sort labels and features by (entity, ts) and merge. See point-in-time join.
• Event time, not ingestion time. A late value keeps the feature_ts of when it became true. Training sees it from then; live serving only after arrival. That small skew is what the log step measures.
• TTL. A value older than the TTL counts as missing, so a stale entity does not look fresh. Set it from how fast the signal decays.
• Freshness vs cost. Streaming gives seconds of freshness for always-on compute; hourly or daily batch is cheaper and fine for slow features. Pick per feature.
• Streaming features. Windowed counts and sums (stream-windowed-aggregation) are written to both stores; the offline history is the replayable record.
• Versions. A feature is addressed by (entity key, name, version); a changed transformation is a new version, kept until models trained on the old one retire.

Code

Sorted per-entity log, as-of lookup by binary search, materialization to a hash map. Compiled and run; asserts cover leakage (the naive join returns a future value), TTL, late arrival and offline/online agreement.

use std::collections::HashMap;

type Entity = u64;
type Ts = u64;

/// Offline store: append-only history per entity, kept sorted by feature timestamp.
#[derive(Default)]
pub struct Offline {
    rows: HashMap<Entity, Vec<(Ts, f64)>>,
}

impl Offline {
    pub fn append(&mut self, e: Entity, ts: Ts, v: f64) {
        let h = self.rows.entry(e).or_default();
        // late/out-of-order arrival: insert at the right place, not at the end
        let i = h.partition_point(|&(t, _)| t <= ts);
        h.insert(i, (ts, v));
    }

    /// As-of join for one label row (e, ts): newest value with feature_ts <= ts,
    /// and not older than `ttl` (None = no limit).
    pub fn point_in_time(&self, e: Entity, ts: Ts, ttl: Option<Ts>) -> Option<f64> {
        let h = self.rows.get(&e)?;
        let i = h.partition_point(|&(t, _)| t <= ts); // first index with t > ts
        let (ft, v) = *h.get(i.checked_sub(1)?)?;
        match ttl {
            Some(l) if ts - ft > l => None,
            _ => Some(v),
        }
    }

    /// WRONG on purpose: "latest value" join, ignores the label time (leaks the future).
    pub fn naive_latest(&self, e: Entity) -> Option<f64> {
        self.rows.get(&e)?.last().map(|&(_, v)| v)
    }

    /// Materialization: newest value per entity at time `now` into the online store.
    pub fn materialize(&self, now: Ts, ttl: Option<Ts>) -> Online {
        let mut o = Online::default();
        for &e in self.rows.keys() {
            if let Some(v) = self.point_in_time(e, now, ttl) {
                o.kv.insert(e, v);
            }
        }
        o
    }
}

/// Online store: one value per entity key, O(1) reads (stands in for Redis/DynamoDB).
#[derive(Default)]
pub struct Online {
    kv: HashMap<Entity, f64>,
}

impl Online {
    pub fn get_online(&self, e: Entity) -> Option<f64> {
        self.kv.get(&e).copied()
    }
}

fn main() {
    let mut off = Offline::default();
    for (ts, v) in [(10, 1.0), (20, 2.0), (30, 3.0)] {
        off.append(7, ts, v);
    }
    // (1) no leakage: label at ts=25 sees 2.0 (ts=20), never 3.0 (ts=30)
    assert_eq!(off.point_in_time(7, 25, None), Some(2.0));
    assert_eq!(off.naive_latest(7), Some(3.0)); // naive join leaks the future
    assert_eq!(off.point_in_time(7, 5, None), None); // nothing known yet
    // (2) TTL: value from ts=20 is 5 old at ts=25 -> fresh for ttl=10, expired for ttl=3
    assert_eq!(off.point_in_time(7, 25, Some(10)), Some(2.0));
    assert_eq!(off.point_in_time(7, 25, Some(3)), None);
    // late arrival is placed in order
    off.append(7, 22, 2.5);
    assert_eq!(off.point_in_time(7, 25, None), Some(2.5));
    // (3) online value == PIT value at "now"
    let on = off.materialize(100, None);
    assert_eq!(on.get_online(7), off.point_in_time(7, 100, None));
    assert_eq!(on.get_online(8), None);
    println!("tests ok");
}

Complexity and constants

• As-of lookup: O(log h) per label row (h = versions of that entity); a sort-merge bulk join is O((n + m) log) for n label rows and m feature rows.
• Online read: O(1) hash lookup; per request the cost is one network round trip (batched across features); its latency depends on the store and the network, so measure it in your own setup.
• Materialization: O(entities changed).
• Measured on this Mac (Apple M4 Pro, one core, release build, std HashMap, everything in RAM): 1M rows (100k entities x 10 versions), 1M random as-of lookups with a TTL: about 1.8-2.0 x 10^7 lookups/s (about 50 ns each); 1M online gets over the same entities: about 1.3-1.4 x 10^8/s (about 7 ns). Three runs; harness = the code above plus an Instant loop. CPU-only: a networked store is bound by round trips.

Efficiency

• Cost: per training row, one as-of lookup (about 50 ns in-process, measured above; in practice bound by scanning the offline files). Per inference request, one batched online read, bound by network latency and per-key cost; fan-out over entity keys and features multiplies it. Materialization compute is the standing cost.
• Levers:
  • Batch reads of all features for a request in one call — turns N round trips into 1 — whenever a request needs several features.
  • Group features sharing an entity key into one row — one read returns many — always.
  • Partition offline data by date, prune by label time range — scans only needed files — large backfills.
  • Stream only features that need seconds, batch the rest — less always-on compute — mixed feature sets.
• Naive → standard → state of the art:
  • Copy-pasted SQL for training, service code for serving · skew and leakage found only in production
  • One definition, offline join with as-of semantics, scheduled materialization to key-value store · same values, hours of freshness
  • Same definition on streaming and batch, served-feature logging, freshness and skew monitors · seconds of freshness, skew measured

Implementation notes

• Store feature_ts next to every online value so the service can check age and fall back to a default.
• Cost drivers: online read latency, fan-out (entities x features per request), round trips, backfill compute, streaming compute.
• Offline: columnar files (Parquet class), sorted by (entity, feature_ts) inside partitions, so joins merge instead of probe.
• Entity keys need the same type and normalisation (case, padding) in both stores; mismatches return silent nulls.
• Make the online write idempotent and last-write-wins by feature_ts, so replays and out-of-order stream events do not move a value back: a timestamp-guarded upsert gives the effect of exactly-once processing without transactions.
• Count defaults served for missing values; a rising null rate is the first symptom of a broken pipeline.

Pitfalls

• Joining "latest value" into training rows: the model learns from the future; offline metrics look great, production does not. Test with a case like the one in the code.
• Window aggregates that include the label event itself: the window must end before label_ts.
• Two code paths for one feature (SQL and service code) that diverge.
• TTL left at infinity: a dead entity serves an old value as current.
• Changing a feature in place: models trained on the old meaning silently get the new one.

Variants and alternatives

In production

Uber's Michelangelo platform includes a shared feature store; open-source stores: Feast, Hopsworks, Featureform; managed ones exist on the major clouds and data platforms.

Implementations

Sources

• Sculley et al., Hidden Technical Debt in Machine Learning Systems, NeurIPS 2015. paper
• Uber Engineering, Meet Michelangelo: Uber's Machine Learning Platform, 2017 (blog post; link not verified).
• Feast documentation, Point-in-time joins. docs