Part 9 · Derived views

Feature Stores and ML Data Pipelines

Lesson 22 built the last retrieval view — a search and vector index that answers "what is relevant to this query." This lesson builds the last family of derived views in Part 9, and the one closest to the machine-learning product itself: the features a model reads as input, and the predictions it writes as output. Both are derived data. A feature is not a fact a user typed; it is a function computed over the raw event log — "this user's 7-day click rate," "this product's average rating." A prediction is a function of features and a model version. Treat them as derived views and the whole discipline of Part 9 applies: there is one source of truth (the raw event log and base tables), the views can go stale, and they can always be recomputed. Treat them carelessly and you get the two failures that ruin ML systems in production — label leakage (training on information that did not exist yet) and train-serve skew (the same feature computed two different ways offline and online). This lesson builds the machinery that prevents both: the DataFrame as the ML execution abstraction, point-in-time joins, and the offline/online feature store split.

DDIA source

Arc drawn from Martin Kleppmann, Designing Data-Intensive Applications — the batch-processing chapter (training-set materialization as a deterministic join, lesson 18) and the stream-processing chapter (event time, CDC, the stream-table join, lesson 19), with the 2nd-edition additions on DataFrames and dense numeric arrays as the analytics/ML execution model, and on ML training pipelines as derived-data systems. Original synthesis; the feature-store, point-in-time-join, and train-serve-skew framings are ours. The worked interview case for a feature store lives at lesson 31; this is the mechanism lesson behind it.

Linear position

Prerequisite: Lesson 18 (batch: building a training set is a deterministic join over an immutable input you can recompute), lesson 19 (event time vs processing time, CDC keeping derived views in sync, the stream-table "as of event time" join), and lesson 04 (columnar storage — why analytics and DataFrames scan columns, not rows). A hand-wave at encoding/schema (lesson 05) helps, since a feature definition is a schema two systems must share.
New capability: Design an ML data pipeline and feature store as correct derived data — pick the DataFrame abstraction that fits your scale, materialize a training set with a point-in-time join that cannot leak labels, split offline (batch) from online (low-latency lookup) feature serving, and guarantee the two stay consistent so the model sees at inference exactly what it saw in training.

The plan

Six moves. (1) The DataFrame — named columns over partitioned rows — as the ML/analytics execution abstraction; lazy vs eager evaluation; distributed (Spark) vs single-node (pandas/Polars); and the dense matrices/arrays/tensors a model actually consumes. (2) The ML data pipeline stages: ingest → clean/validate → feature engineering → training-set materialization → serving, and the ETL/ELT framing. (3) Point-in-time (as-of) joins — the one join you must get right, with a tiny worked example, because the obvious join leaks labels. (4) Offline vs online feature stores — batch training-set generation vs single-row low-latency lookup at inference. (5) Train-serve skew — the same feature computed two ways diverges; the fix is one definition plus shared encoding (lesson 05). (6) Predictions as derived data — a model's output is a view you can recompute and backfill. Then the "Where is truth?" callout and the hand-off to cloud-native storage.

1 · The DataFrame: named columns over partitioned rows

Almost all analytics and ML data work happens through one abstraction: the DataFrame. A DataFrame is a table of named, typed columns over a set of rows — like a SQL table, but as a first-class value in a program you can pass around, transform, and chain operations on. The operations are the relational ones you already know plus a few ML-flavored extras: filter (drop rows), select (project columns), join (combine on a key), groupBy ... agg (aggregate), withColumn (derive a new column). The reason this shape dominates ML is that feature engineering is column algebra: you take raw columns and compute new ones — "encode this category as an integer," "compute the 7-day rolling click rate," "bucket this price into deciles." Each is a column-to-column transform.

Two design axes separate the DataFrame libraries you will meet, and choosing wrong is a real production cost.

Eager vs lazy evaluation. An eager DataFrame (pandas) executes each operation the instant you write it — df.filter(...) immediately materializes a new filtered frame in memory. Simple to reason about and debug, but it cannot optimize across operations: if you filter after a join, pandas still built the full join first. A lazy DataFrame (Spark, Polars in lazy mode) does not execute as you go; it records your operations as a query plan (a directed graph of transforms) and runs nothing until you call an action that needs a result — collect(), count(), write-to-disk. Before executing, an optimizer rewrites the plan: it pushes the filter down below the join so fewer rows are ever joined (predicate pushdown), drops unread columns (projection pushdown), and picks join strategies — the very batch optimizations of lesson 18. Lazy buys whole-pipeline optimization at the cost of "nothing happened until the action," which surprises people debugging.

Single-node vs distributed. A single-node DataFrame (pandas, Polars) holds all the data in one machine's memory and is bounded by that machine's RAM — wonderful up to tens of GB, a wall beyond it. A distributed DataFrame (Spark) partitions the rows across many machines (partitioning, lesson 18) and runs each operation in parallel on every partition; a join or groupBy triggers a shuffle — the same all-to-all repartition-by-key from lesson 18 — which is again where the cost lives. The rule of thumb: if your data fits comfortably on one box, a single-node lazy frame (Polars) is faster and far simpler than a cluster; reach for Spark only when the data genuinely exceeds one machine, because the shuffle and coordination overhead is not free.

Library	Evaluation	Scale	Use when
pandas	Eager	Single node, in-RAM	Exploration, small/medium data, subsecond iteration
Polars	Lazy (and eager)	Single node, larger-than-pandas	One big machine, want the optimizer without a cluster
Spark	Lazy	Distributed, partitioned	Data exceeds one machine; cluster-scale joins/aggregates

From DataFrame to tensor: what the model actually eats

A DataFrame is the preparation surface, but a model does not consume named columns — it consumes dense numeric arrays. The last step of any ML pipeline converts the typed, named, often-categorical DataFrame into a matrix (a 2-D array of numbers: rows = examples, columns = numeric features) for classical models, or a tensor (an N-dimensional array — e.g. a batch of images is 4-D: batch × height × width × channels) for deep models. This conversion is exactly where categorical encodings (one-hot, integer, embedding lookups) and normalization happen, and it is why columnar storage (lesson 04) underpins ML: a feature is a column, models read whole columns at once, and a columnar layout lets you scan one feature across millions of rows without touching the others. The DataFrame is the bridge from named, heterogeneous, human-meaningful data to the dense, homogeneous, numeric form the linear algebra needs.

2 · The ML data pipeline as derived data

An ML system's data substrate is a pipeline of derived views, each computed from the one before, all ultimately derived from the raw event log and base tables. The stages:

1Ingest. Land raw data from sources — event logs (clicks, views), operational DB tables (via CDC, lesson 19), third-party feeds — into a lake or warehouse. This raw layer is the system of record for everything downstream.

2Clean / validate. Enforce schema, types, ranges; drop or quarantine malformed rows; deduplicate. A bad value here silently poisons every model trained on it, so validation is a gate, not a nicety.

3Feature engineering. The column algebra of §1: derive features from cleaned data — aggregations, rolling windows, encodings, joins across entities. The output is feature tables keyed by an entity id and a timestamp.

4Training-set materialization. Join labels (from the event log) to feature values to produce a flat training table. This is the join that must be point-in-time correct (§3) or the model learns from the future.

5Serving. At inference, fetch the current feature values for one entity, run the model, write the prediction back as a derived view (§6).

The framing you will hear is ETL vs ELT. ETL (extract, transform, load) transforms data before loading it into the warehouse — you decide the shape up front. ELT (extract, load, transform) loads raw data first and transforms it inside the warehouse with SQL/DataFrame jobs, so the raw layer is preserved and transforms can be re-run and revised. ELT is the modern default for ML precisely because of lesson 18's lesson: keep the immutable raw input, and recompute the derived layers when a feature definition changes. The feature tables, the training set, and the predictions are all derived; the raw event log is the only thing you cannot regenerate.

3 · The point-in-time join: the one you must get right

Here is the join at the center of every feature store, and the one that quietly destroys models. You have labels — each is an event with a timestamp: "user 42 was shown product 91 at 10:00 and clicked." You have features that change over time — "user 42's 7-day click rate," which was 0.10 last week and 0.30 today. To build a training row for that 10:00 label, you must attach the feature value as it was at 10:00 — not its value now. Reading "whatever the feature table says today" is the same mistake as the stream-table join in lesson 19: you must read the feature as of the label's event time.

Why it matters has a name: label leakage. If you join the 10:00 label to today's feature value (0.30), you have leaked information from after the prediction moment into the training row. The model learns a relationship that was not knowable at prediction time; offline metrics look fantastic, and the model fails in production because at serving time the future is, of course, unavailable. The fix is a point-in-time join (also called an as-of join): for each label, take the most recent feature value whose timestamp is ≤ the label's event time.

Worked as-of join — leakage vs correct

A feature table records user 42's "7-day click rate" as it was recomputed over time, and we have two labels to build training rows for:

feature history (user 42, "7d_click_rate"): ts=09:00 value=0.10 ts=10:30 value=0.30 <- recomputed later in the day labels to join: label A: shown@10:00 clicked=1 label B: shown@11:00 clicked=0 WRONG ("join to current value", 0.30 for both): label A -> feature 0.30 (LEAK: 0.30 did not exist until 10:30, after 10:00) label B -> feature 0.30 (ok by luck) RIGHT (as-of join: latest feature ts <= label ts): label A@10:00 -> latest ts <= 10:00 is 09:00 -> feature 0.10 (knowable at 10:00) label B@11:00 -> latest ts <= 11:00 is 10:30 -> feature 0.30 (knowable at 11:00)

Label A is the tell: the naive join gives it 0.30, a value that was not computed until 10:30 — half an hour after the prediction it is supposed to inform. The as-of join gives it 0.10, the value that genuinely existed at 10:00. Get this wrong across a million rows and your offline AUC is a fiction.

Mechanically, the point-in-time join is a join on entity_id plus an inequality on time (feature.ts ≤ label.ts), keeping the row with the maximum feature timestamp per label. It is more expensive than an equality join — engines implement it as a sort-merge on (entity, time) so it stays a sequential scan rather than a per-row lookup (lesson 18) — but it is the price of not leaking. Every serious feature store (Feast, Tecton, Databricks) makes the point-in-time join the primitive of training-set generation precisely because hand-rolled joins get it wrong.

4 · Offline vs online: two stores, two access patterns

The same logical features are read in two completely different ways, and a feature store is, at heart, the system that serves both from one definition.

Offline store (batch)

Generates training sets. Access pattern: scan millions of rows across the whole history with point-in-time joins (§3). Lives in a columnar lake/warehouse (lesson 04) — Parquet on S3, BigQuery. Throughput-optimized; latency irrelevant; a job runs for minutes.

Online store (serving)

Serves features at inference. Access pattern: fetch the current feature vector for one entity by key, in single-digit milliseconds, under request load. Lives in a low-latency key-value store — Redis, DynamoDB, Cassandra. Point-lookup-optimized; only the latest value per entity is kept.

The same feature ("7-day click rate") is therefore materialized into two physical stores: a wide columnar history for training, and a single hot row per entity for serving. A batch or streaming job computes the feature and writes it to both — and that dual materialization is exactly where the next failure mode lives.

one feature definition, two materializations, one model +-------------------------+ raw event log -->| feature transform | (ONE definition; §5) base tables -->| (e.g. 7d_click_rate) | +-----------+-------------+ | +---------------+----------------+ v v OFFLINE store (batch) ONLINE store (serving) columnar history, all ts KV: latest value per entity Parquet/BigQuery Redis/DynamoDB | | point-in-time join point lookup by key over millions of rows one entity, ~1-5 ms v v +---------------+ +---------------+ | training set |---> train ----> | MODEL | ---> prediction +---------------+ (model v1) +---------------+ (derived view, §6) | same v1 reads the SAME features here as in training -> if offline and online disagree = TRAIN-SERVE SKEW

Keeping the two consistent is the core feature-store problem. The clean pattern is the CDC/log approach of lesson 19: one feature-computation job writes to the offline columnar table, and the online store is kept in sync from that same computation (a materialization job pushes the latest value per entity into the KV store, or both subscribe to one feature stream). The freshness gap between them — how stale the online value is relative to the offline history — is the online store's lag budget, the streaming-vs-batch trade of lessons 18–19 applied to features.

5 · Train-serve skew: the same feature, computed twice

Train-serve skew is the silent killer of deployed models: the feature value the model saw in training differs from the value it sees at serving, even for the identical entity and time, because the two were computed by two different code paths. The classic origin: the training feature is computed in a Spark/SQL batch job ("7-day click rate" as a windowed aggregate over the event log), while the serving feature is computed in application code in the request path ("count this user's clicks in the last 7 days" via a different query, a different time-zone handling, a different null policy). Each looks right in isolation; they disagree at the third decimal place; the model — which is exquisitely sensitive to its input distribution — degrades, and nothing in your tests catches it because each path passes its own unit tests.

How skew sneaks in (and the worked tell)

Training (batch): 7d_click_rate = clicks_7d / impressions_7d where the 7-day window is computed in UTC, and a user with zero impressions yields NULL (dropped). Serving (app code): the same ratio, but the window is computed in the server's local time zone, and zero impressions yields 0.0 (kept). For most users the two agree; for users near a day boundary or with sparse activity they diverge — say training says 0.25 and serving says 0.20 for the same user at the same instant. The model was fit assuming 0.25; it receives 0.20; predictions drift. The bug is invisible per-path: both "work." The only reliable detection is an offline/online parity audit — log the served feature vector alongside the prediction, recompute the same feature from the offline definition for those entities/times, and assert they match within tolerance.

The fix is structural, not "be more careful": one feature definition, shared by both paths. You define the transform once — as a function, a SQL/DataFrame expression, or a feature-store-managed definition — and both the offline training-set job and the online serving path execute that same definition over the same inputs, with the same encoding. That shared encoding is exactly lesson 05's point: a feature is a schema (name, type, null policy, units, time zone) that two systems must agree on, and the way you keep two systems agreeing on a schema is to make the schema a single shared artifact, not two re-implementations. Feature stores enforce this by being the registry where a feature is defined once and both stores are populated from that one definition. When the definition changes, you bump its version and recompute — the derived-data discipline of lesson 20.

6 · Predictions are derived data too

The final move closes the loop. A model's output — a score, a ranking, a class, an embedding — is itself a derived view: a deterministic function of (input features, model version). That reframing buys the same superpowers as every other derived view in Part 9. Predictions can be recomputed: ship model v2 and re-score the backlog. They can be backfilled: generate v2's predictions over all of history to compare against v1 before you roll out. They can be cached as a materialized view (precompute recommendations nightly) or computed on demand, the freshness-vs-cost trade of the whole part. And they are auditable: record the feature vector, the model version, and the timestamp alongside each prediction, and any output is reproducible — you can re-derive exactly what the model saw and said. The source of truth was never the prediction; it was the event log and the model artifact, and the prediction is a regenerable function of them.

This is why a prediction log is not just a debugging convenience — it is the evidence layer. It is what powers the parity audit of §5 (compare served vs recomputed features), what lets you attribute a metric regression to a model version versus a data shift, and what makes "predictions as derived data" an operational reality rather than a slogan. The model registry (lesson 31's sibling case at lesson 33) is the catalog of model versions that makes this reproducibility complete.

Failure modes

Label leakage. Joining a label to a feature value computed after the label's event time; offline metrics are inflated, production fails. Symptom: a model that is "too good" offline and disappoints live. Fix: point-in-time / as-of join (§3).
Train-serve skew. The same feature computed by two code paths (batch vs request-path) diverges. Symptom: live performance below offline, no failing test. Fix: one shared definition + shared encoding (§5).
Stale online store. The online KV value lags the offline definition past its budget; the model serves on old features. Symptom: features "frozen" after a pipeline stall. Fix: monitor materialization lag (§4).
Eager frame OOM. A pandas join on data that no longer fits RAM. Symptom: the box swaps and dies. Fix: lazy + larger node (Polars) or distribute (Spark) (§1).
Silent data corruption. A malformed value passes ingest unvalidated and poisons every downstream model. Symptom: gradual metric rot with no code change. Fix: validation gate at stage 2 (§2).
Untracked feature version. Changing a feature definition in place so old training sets cannot be reproduced. Fix: version definitions, recompute (§5, lesson 20).

Decision checklist

Does the data fit one machine? If yes, lazy single-node (Polars); if no, distributed (Spark) and watch the shuffle (§1).
Is every training-set join a point-in-time join on (entity, time ≤ label time)? Can you state, for each feature, that it was knowable at the label's event time? (§3)
Are offline and online features written from one definition, or two code paths? (§5)
Is there an offline/online parity audit logging served features and recomputing them? (§5, §6)
What is the online store's freshness budget, and do you monitor materialization lag against it? (§4)
Are feature definitions versioned so a training set is reproducible and recomputable after a change? (§2, §6)
Do you log (features, model version, timestamp) with every prediction so outputs are auditable and backfillable? (§6)

Checkpoint exercise

Try it

You are building the feature substrate for a click-prediction model. Features include user_7d_click_rate and item_avg_rating, both changing over time; labels are impression events with a timestamp and a clicked flag. (a) You have 4 billion impression events and a feature history far larger than one machine — pick the DataFrame engine and say what the expensive step in the training-set join will be. (b) Write, in words, the join condition that builds a leak-free training row for an impression at time t, and give a two-row worked example where the naive "current value" join would leak. (c) The model will serve at 5 ms p99 per request — describe the online store and how it is kept consistent with the offline store from one feature definition. (d) Three weeks in, live CTR is below the offline estimate but no test fails. Name the most likely cause and the one audit that would have caught it.

Where is truth?

System of record: the raw event log and base tables (clicks, impressions, ratings, operational rows) — the only data you cannot regenerate. Copies / derived views: feature tables (offline columnar history + online KV latest-value), the materialized training set, and the model's predictions — all functions of the SoR plus a code/model version. Freshness budget: the online store's materialization lag behind the feature definition (CDC/batch lag, lessons 18–19); the offline history is as fresh as its last batch run. Owner: the feature definition is the single owned artifact (the feature-store registry), shared by both stores. Deletion path: delete from the SoR (the event log) and recompute the derived feature tables and predictions; per-user deletion propagates through the recompute. Reconciliation / repair: recompute features and re-materialize both stores from the raw log; re-score predictions with the current model version. Evidence it is correct: a no-leakage check (every feature timestamp ≤ its label's event time in the materialized training set) and an offline/online parity audit (served feature vectors logged with predictions and re-derived from the offline definition match within tolerance).

Where this points next

Part 9 has now built every family of derived view — materialized views and their correctness (lesson 20), analytics query execution (21), search and vector retrieval (22), and now features and predictions for ML. Every one of them — the columnar warehouse, the vector index, the offline feature store, the prediction log — lives on the same physical substrate: cloud object storage. The next lesson (lesson 24, Cloud-Native Storage as a Subsystem) opens that substrate up. It asks how a system like S3 actually delivers durability, elasticity, and the separation of storage from compute that lets a Spark cluster, a warehouse, and a feature store all read the same Parquet files — and what the data-intensive designer gives up (latency, strong consistency guarantees, per-request cost) in exchange. The feature store you just designed assumes that substrate exists; lesson 24 builds it.

Takeaway

Features and predictions are the last family of derived views: a feature is a function of the raw event log, a prediction is a function of features and a model version, and the raw log is the only source of truth. The execution abstraction is the DataFrame — named columns over rows — with two axes: eager (pandas) vs lazy (Spark/Polars, which optimize the whole plan), and single-node (RAM-bound) vs distributed (partitioned, pays a shuffle); the final step converts the DataFrame into the dense matrices/tensors a model consumes, which is why columnar storage (lesson 04) underpins ML. The pipeline ingests → cleans → engineers features → materializes a training set → serves, in ETL/ELT form. The join at its center must be a point-in-time (as-of) join — attach each label the feature value as of the label's event time (lesson 19) — or you get label leakage and an offline metric that is a fiction. Features are served from two stores: an offline columnar history for batch training-set generation and an online KV store for millisecond point lookups at inference, kept consistent by computing from one definition. Computing a feature two ways (batch vs request-path) causes train-serve skew; the fix is one shared definition plus shared encoding (lesson 05), audited by an offline/online parity check. And because predictions are themselves a derived view, they can be recomputed, backfilled, cached, and audited — provided you log features, model version, and timestamp with every one.

Interview prompts

What is a DataFrame, and what do the eager/lazy and single-node/distributed axes buy you? (§1 — named typed columns over rows; lazy (Spark/Polars) records a plan and optimizes the whole pipeline (pushdown, join choice) before any action, eager (pandas) runs each op immediately; single-node is RAM-bound and simple, distributed partitions rows and parallelizes but pays a shuffle on joins/aggregates. Fit one machine → Polars; exceed it → Spark.)
Why does ML ultimately need columnar storage and tensors, not just a DataFrame? (§1 — a model consumes dense numeric matrices/tensors, not named columns; a feature is a column and models read whole columns across millions of rows, so columnar layout (lesson 04) lets you scan one feature without the others; the DataFrame is the bridge that encodes categoricals/normalizes into the dense array.)
What is a point-in-time / as-of join and why is it mandatory for training sets? (§3 — join each label to the most recent feature value with timestamp ≤ the label's event time (a join on entity + a time inequality, lesson 19's "as of event time"); joining to the current value leaks future information (label leakage), inflating offline metrics while the model fails live.)
Contrast the offline and online feature stores. (§4 — offline = columnar history (Parquet/warehouse), scans millions of rows with point-in-time joins to build training sets, throughput-optimized; online = low-latency KV (Redis/DynamoDB), point-lookup the latest vector for one entity in ~ms at inference; same feature materialized into both from one definition.)
What is train-serve skew, how does it arise, and what is the structural fix? (§5 — the feature value in training differs from serving because two code paths (batch vs request-path) compute it differently (time zone, null policy, windowing); the fix is one shared definition + shared encoding (lesson 05), not "be careful," detected by an offline/online parity audit.)
In what sense is a model's prediction "derived data," and why care? (§6 — a prediction is a deterministic function of (features, model version); so it can be recomputed, backfilled for a new model, cached as a materialized view, and audited — provided you log features + model version + timestamp; the SoR is the event log and model artifact, never the prediction.)
Your offline AUC is excellent but live performance is mediocre and no test fails — what are the two prime suspects and how do you tell them apart? (§3, §5 — label leakage (a feature used a value not knowable at the label's event time → check every feature ts ≤ label ts with a no-leakage audit) and train-serve skew (offline and online features diverge → run an offline/online parity audit logging served vectors and recomputing them).)