02. Feature Store + Training Pipeline — Low-Level Design¶

~22 min read · Part 3 of 4 (Overview → HLD → LLD → Q&A)

The HLD named the boxes. This file opens the three that carry the design's weight — the point-in-time join engine, the online serving path, and the materialization engine — and pins down the data models, the as-of join algorithm, and the concurrency corners where a feature store actually corrupts a model.

Data models¶

Everything hinges on two timestamps per feature value, so the schemas make them explicit. In the offline store, each feature view is a table (or partitioned dataset) shaped like this:

CREATE TABLE fv_card_velocity (          -- one table per feature view
    card_id            BIGINT     NOT NULL,   -- entity key
    event_timestamp    TIMESTAMP  NOT NULL,   -- when the value became TRUE in the world
    created_timestamp  TIMESTAMP  NOT NULL,   -- when WE computed/logged it
    txn_count_5m       INT,                    -- the feature(s)
    failed_auth_1h     INT
) PARTITION BY (DATE(event_timestamp));       -- prune scans by date

The two timestamps are the whole game. event_timestamp is when the value was true in the world (the transaction happened at 12:00:00). created_timestamp is when the pipeline produced the row — which, for a backfill or a late-arriving correction, can be days after the event. A point-in-time join that ignores created_timestamp will happily join a value that was computed with hindsight into a training row from before that hindsight existed, which is leakage. Partitioning by event_timestamp date is what lets a one-month training run read ~15 TB instead of the full 90 TB.

The online store holds only the latest value per entity per feature view, keyed for a single multiget:

key:   fv:{feature_view}:{entity_id}          # e.g. "fv:card_velocity:8675309"
value: { txn_count_5m: 3, failed_auth_1h: 0,
         event_timestamp: 2026-03-01T12:00:00.000Z }   # serialized, ~2 KB for the full model
ttl:   per feature-view (velocity: 600s; 90d-avg: 172800s)

Storing event_timestamp alongside the value is not optional metadata; it is what the serving path checks against TTL to decide whether a value is fresh enough to serve or must be returned as null. The TTL is per feature view because a 5-minute velocity count is worthless after 10 minutes while a 90-day average is fine for two days.

The registry ties definitions to versions and lineage:

CREATE TABLE feature_view (
    name           VARCHAR PRIMARY KEY,
    version        INT      NOT NULL,
    entities       JSONB    NOT NULL,   -- ["card_id", ...]
    features       JSONB    NOT NULL,   -- schema of produced features
    transformation TEXT     NOT NULL,   -- the ONE definition, compiled to both pipelines
    mode           VARCHAR  NOT NULL,   -- 'batch' | 'stream'
    ttl_seconds    INT      NOT NULL,
    online         BOOLEAN  NOT NULL    -- materialized to online store?
);

The training spine — the input to get-historical-features — is a table the caller supplies, not one we own:

entity_df: (card_id, user_id, merchant_id, device_id, event_timestamp, label)

The label for the fraud model is "was this transaction charged back as fraud," and its own maturity delay is a trap discussed under edge cases.

Component internals¶

Component 1 — Point-in-time (PIT) join engine¶

Responsibility: produce a training dataset in which every row's features are exactly what was knowable just before that row's event_timestamp — no leakage — respecting each feature's TTL.

class PointInTimeJoin:
    def build(entity_df, feature_service) -> DatasetURI
    def _asof_join(spine, fv_table, ttl) -> DataFrame   # per feature view

For each feature view in the service, the engine performs an as-of join between the spine and the feature-view table on the entity key, with the ordering predicate on time. The join is the defining algorithm and gets its own section below.

Component 2 — Online serving path¶

Responsibility: return the ~200-feature vector for the requested entities in under 10 ms, enforce freshness, and log the served vector for parity.

def get_online_features(feature_service, entities) -> FeatureVector:
    keys = build_keys(feature_service, entities)     # e.g. 4 entities → 4 keys
    rows = online_store.multiget(keys)               # ONE round trip, fan-out 4
    vector, ts = {}, {}
    for fv, row in zip(feature_service.views, rows):
        if row is None or is_expired(row, fv.ttl):   # older than TTL → null,
            vector.update(nulls_for(fv))             # matching training's null handling
        else:
            vector.update(row.features)
            ts[fv.name] = row.event_timestamp
    log_served_vector_async(feature_service, entities, vector, ts)  # parity: never blocks
    return FeatureVector(vector, ts)

Three points carry the weight. The multiget is one round trip fanning to four keys, not four sequential reads — that is the 10 ms budget. The freshness check treats an expired value as null, and it must produce the same null that the offline TTL logic produced at training time, or the model sees a value in production it never saw in training. And logging is asynchronous and fire-and-forget, so a slow offline store can never leak into the 10 ms read; the logged vector is the exact bytes the model consumed, which is what makes the next training run skew-free.

Component 3 — Materialization engine¶

Responsibility: keep the online store in agreement with the offline store by upserting the latest value per entity, without ever regressing a fresher value and without saturating the online store.

def materialize(feature_view, start_ts, end_ts):
    latest = offline_store.latest_per_entity(feature_view, start_ts, end_ts)
    for batch in chunked(latest, size=10_000):       # throttle: not 200M at once
        for row in batch:
            online_store.put_if_newer(               # guard against regressing
                key=f"fv:{feature_view}:{row.entity_id}",
                value=row.features,
                event_timestamp=row.event_timestamp) # compare on THIS, not wall clock
        rate_limit()                                  # protect the 10 ms live-read SLA

put_if_newer compares the incoming event_timestamp to the stored one and skips the write if the store already holds something fresher. This is what lets a slow nightly batch load and a per-second streaming update both write the same key safely — the fresher one always wins regardless of which arrives last.

Core algorithm — the as-of join, walked with the fraud numbers¶

The as-of join is where point-in-time correctness lives or dies. Take one row from the fraud model's 200M-row spine: a transaction on card 8675309 at event_timestamp = 2026-03-01 12:00:00.000, whose label (fraud, via a chargeback that arrived 2026-04-20) is being joined to the card_velocity feature view. The feature view's offline table holds, for that card, several computed values:

card_id   event_timestamp            created_timestamp          txn_count_5m
8675309   2026-03-01 11:58:30.000    2026-03-01 11:58:31.100    2
8675309   2026-03-01 11:59:50.000    2026-03-01 11:59:51.050    3
8675309   2026-03-01 12:01:10.000    2026-03-01 12:01:11.000    4   ← FUTURE, must not leak
8675309   2026-03-01 09:00:00.000    2026-04-20 15:00:00.000    9   ← BACKFILL, must not leak

The join steps:

Filter by event time. Keep only rows with event_timestamp <= 2026-03-01 12:00:00.000. This drops the 12:01:10 row — a value that became true after the transaction. Joining it would tell the model the card had already made a fourth transaction it had not yet made. Leakage.
Filter by created time. From what remains, keep only rows with created_timestamp <= 2026-03-01 12:00:00.000. This drops the 09:00:00 / created 2026-04-20 backfill row: its event time is in the past, but we did not know that value at scoring time — it was computed 50 days later. Ignoring created_timestamp is the most common subtle leak, because the row looks innocent on event time alone.
Order and take latest. Of the survivors (11:58:30 → 2 and 11:59:50 → 3), order by event_timestamp descending and take the first: txn_count_5m = 3, event_timestamp 11:59:50.
Apply TTL. The card_velocity TTL is 600 s. 12:00:00 − 11:59:50 = 10 s, well within TTL, so the value stands. Had the latest surviving value been older than 600 s, the join would emit null — the same null the serving path produces for a stale online value, which is exactly why serving checks TTL too.

So this spine row trains on txn_count_5m = 3. And here is the payoff for the threaded scenario: when this same card is scored live at 12:00:00, the serving path reads the online store, finds the value written by the streaming pipeline at 11:59:50 (event_timestamp 11:59:50, within its 600 s TTL), and returns txn_count_5m = 3 — the identical value, from the identical logic, that the training join selected. The model reads at 10 ms what it trained on. No skew, no leakage. That equivalence is the entire deliverable.

Across the full 200M-row spine, step 1's time filter and the date partitioning let the engine prune to the relevant partitions per row's month, so the join scans ~15 TB per month of spine rather than the full 90 TB, and distributes the as-of merge across the Spark cluster keyed by entity.

Sequence diagram — a live fraud score under the 10 ms budget¶

Fraud svc     Serving API      Online store        Offline log
   │  score(card,user,mer,dev) │                     │
   ├──────────────────────────▶│                     │
   │              │  multiget([4 keys])               │
   │              ├──────────────────────▶│           │
   │              │◀── 4 rows (+event_ts)─┤           │
   │              │  freshness check vs TTL           │
   │              │  (any stale → null,               │
   │              │   matching training)              │
   │              │  log_served_vector_async ─────────┼──────▶│ (fire & forget)
   │◀── vector ───┤  (~200 features, ~2 KB)           │       │
   │  (p99 < 10ms)│                     │             │

One multiget, an in-process freshness check, an asynchronous log, and return. The log write is off the critical path, so the offline store can be slow or briefly down without touching the 10 ms budget; the vector it logs is byte-identical to what the model consumed, which seeds the next skew-free training run.

Concurrency and edge cases¶

Streaming vs batch write race on the same online key. The streaming pipeline writes txn_count_5m per transaction; the nightly batch materialization writes the same key. If the batch load lands after a fresher streaming update, a naive last-writer-wins would regress the value to something hours old. put_if_newer resolves this by comparing event_timestamp, not arrival order — the fresher event always wins regardless of which write executes last.
Idempotent materialization. Re-running materialize over a time range must be safe, because backfills and retries happen. put_if_newer keyed by (entity_id, event_timestamp) makes re-application a no-op: writing the same or an older value changes nothing.
Point-in-time leakage from late-arriving data. Covered in the join walk: the created_timestamp filter is what excludes backfilled and corrected values that were not knowable at the row's instant. Any feature value must carry an honest created_timestamp, or the leakage guard has nothing to filter on.
Label maturity, not just feature leakage. The fraud label matures late — a chargeback can arrive 60–90 days after the transaction. If a training run over "the last 30 days" treats not-yet-charged-back transactions as legitimate, it mislabels frauds that simply have not been reported yet, poisoning the training set. The fix is a label-maturity window: only include spine rows whose event_timestamp is old enough for labels to have settled, independent of the feature-leakage guard.
Partial feature vectors for new entities. A brand-new card has no velocity history, so its online read returns null. This must produce the same null-handling the training join produced for cold-start rows, or a cold card is scored on a distribution the model never trained on — a skew that only shows up for new users, the population fraud most targets.
TTL boundary consistency. The serving freshness check and the offline join's TTL filter must use the identical TTL value from the registry. If serving uses 600 s and the training join used 900 s, the two paths disagree on when a value becomes null — skew hiding inside the freshness logic itself.
Skew detection loop. Because every served vector is logged with its event_timestamp, a background job can recompute those same features offline from raw events and diff them against what was served. A non-zero diff is a divergence between the compiled batch and streaming definitions — the alarm that a "single definition" has quietly stopped being single.
Read-your-writes is not required. Serving is allowed to be eventually consistent within each feature's freshness SLA; there is no requirement that a feature written this millisecond is readable the next. The only hard consistency requirement is value parity between the training join and the serving read, which logging and the shared TTL guarantee — not read-after-write ordering.