« Phase 27 README · Warmup · Hitchhiker's · Principal Deep Dive · Core Contributor · Staff Notes

Phase 27 — Deep Dive: Data & Feature Engineering at Scale

The load-bearing idea across all three labs is a single one wearing three costumes: a feature value is an assertion about an entity at a moment in time, and every correctness bug in ML data engineering is a moment-in-time bug. Leakage is reading a value stamped after the event. Skew is two code paths disagreeing about the value at the same moment. A watermark is the runtime's belief about which moment is complete. This doc traces the actual data structures and control flow that make those moments unambiguous — because that is exactly what the labs force you to implement.

The point-in-time join is a max under a filter, and the filter is the whole point

Lab 01's offline store is dict[str, list[dict]] — one append-only list of timestamped rows per feature view. ingest copies each row and stamps a monotonic _seq counter. That _seq is not decoration; it is the tie-break that makes the join a total order over rows sharing a timestamp.

_point_in_time_row(view, entity_val, event_ts) is the mechanism, and it is deliberately three lines of logic:

for r in self._offline[view.name]:
    if r[view.entity] != entity_val:      continue
    if r[EVENT_TS] > event_ts:            continue   # the FUTURE — never eligible
    key = (r[EVENT_TS], r["_seq"])
    if best is None or key > (best[EVENT_TS], best["_seq"]): best = r

Read it as: argmax of (feature_timestamp, _seq) over the set of rows for this entity whose feature_timestamp is <= event_ts. The r[EVENT_TS] > event_ts: continue line is the invariant that prevents label leakage — it excludes the future from the candidate set before the max is taken. Remove that one line and the join returns the latest value overall, which is precisely the leakage bug. Then a TTL gate: if event_ts - best[EVENT_TS] > ttl, return None — the newest value that was known is too old to trust, so an honest null beats a stale number.

Trace it on the lab's own timeline. user_profile has u1 rows at t=0 ("US") and t=100 ("CA"); user_activity (ttl=10) has u1 rows at t=10, t=40, t=105. Query the event u1@45:

  • profile: t=0 is <= 45 (eligible), t=100 is > 45 (excluded as future). argmax → t=0 → "US". The "CA" correction at t=100 is a fact from the future of the event and never enters the max.
  • activity: t=10 and t=40 are eligible, t=105 excluded. argmax → t=40 (txn_count_7d=9). Age is 45 - 40 = 5 <= 10, within TTL. Kept.

Move the event to u1@60: activity argmax is still t=40, but age 60 - 40 = 20 > 10 → stale → None, while profile (no TTL) still resolves to "US". Move it to u2@45: u2's only profile row is t=50, which is > 45, excluded — the candidate set is empty, best is None → cold-start None. Every one of those outcomes is a consequence of the same argmax-under-filter, and the tests pin all four boundaries: at-or-before is inclusive (a value stamped exactly at the event time is usable; one tick later is not), stale returns null, cold start returns null.

Why the naive join fails at the mechanism level. A plain equijoin against the feature table has no notion of the event's clock — it takes whatever row the key matches, which after any update is the newest. "Join the nearest timestamp" is worse-because-subtler: nearest can be after the event, so it reaches into the future exactly when the feature moved late. Both fail because they compute a max (or a nearest) over the unfiltered set. The correctness lives entirely in the <= event_ts predicate; everything else is bookkeeping.

Complexity and why real systems don't scan. The lab's join is O(rows) per (entity, event) — an honest linear scan, fine for a miniature. At warehouse scale you have millions of entity rows and billions of feature rows; the scan becomes a sort-merge (as-of) join: sort both sides by (entity, timestamp), then sweep the feature side with a cursor that only ever advances, so each entity row binary-searches or merge-walks to its at-or-before value in amortized log or constant time. This is what merge_asof, DuckDB/Snowflake ASOF JOIN, and Feast's compiled get_historical_features all do. The semantics the lab implements are identical; only the access path changes.

Materialization is the same argmax with the clock pinned to "now"

materialize walks the offline history and keeps, per entity, the row with the greatest (event_timestamp, _seq) — the argmax with no upper filter, i.e. an event happening later than every recorded feature time. It writes {"_ts": feature_ts, feature: value, ...} into _online. That is why the training-serving consistency test holds: a historical query with an event_timestamp at or after the newest feature row returns the same row materialization chose. Offline-latest and online are two evaluations of one function at the same clock value — the equality is the no-skew guarantee, and Lab 01 asserts it directly. get_online_features re-runs the TTL gate at serve time (now - rec["_ts"] > ttl), so a stalled materialization pipeline makes features disappear (visible, alarmable) rather than serving three-day-old counts (invisible, corrosive).

Windowing is assignment; the watermark is emission; conflating them is the classic bug

Lab 02's data model is Element(value, timestamp, window) and a frozen Window(start, end) half-open interval. Two orthogonal operations act on it, and the whole Dataflow model is keeping them orthogonal:

Assignment (window_into) is pure and stateless: FixedWindows(size).assign(ts) returns the single window [⌊ts/size⌋·size, +size); SlidingWindows(size, period).assign(ts) returns every window [start, start+size) containing ts where start is a multiple of period, walking downward while start > ts - size. That loop yields ceil(size/period) windows, so with size=10, period=5, t=7 lands in both [0,10) and [5,15). One element becomes several elements — the overlap is a real fan-out cost, each copy aggregated once per window it joined. Assignment refuses an element with no timestamp (raises WindowError): there is nothing to window by.

The shuffle is group_by_key, and the single most important line is the group key:

gk = (e.window, key)   # NOT key alone

Grouping is per (window, key). The same user in two 10-minute windows is two groups; the same user in the overlapping [0,10) and [5,15) from a sliding assignment is two groups. combine_per_key is group_by_key followed by combine_fn on each group's value list — the associativity that lets a real runner pre-combine before the network move lives here, though the lab computes it in one place.

Emission is the StreamingRunner. State is dict[(window,key), list] plus a _fired: set and a monotonic watermark. add(value, ts) routes the element into every assigned window's accumulator. advance_watermark(wm) is the trigger:

if self.watermark is not None and wm < self.watermark: raise WindowError  # never backwards
for gk in self._order:
    if gk in self._fired: continue
    window, key = gk
    if window.end <= wm:
        fired.append(Element((key, combine(state[gk])), _window_ts(window), window))
        self._fired.add(gk)

window.end <= watermark is the emit-on-watermark trigger: a window fires when event time has provably advanced past its end. _fired guarantees exactly once per (window, key) — the pane's effect is applied once even as the watermark keeps advancing. The backwards guard encodes that a watermark is monotonic by definition.

The batch = streaming proof, mechanically. build_word_count runs the transforms over a bounded source; as_result_dict normalizes panes to {(window,key): count}. The streaming path feeds the identical (word, 1) elements into a StreamingRunner and calls advance_watermark(10_000) — jumping the watermark past every window at once. Both produce the same dict; the assert passes. That is the theorem stated as code: a bounded input is a stream whose watermark leaps to infinity when the source ends. The naive alternative — one codebase for batch, another for streaming — has no such invariant to test, which is why the two silently drift and become skew.

The TRANSFORM's fitted statistics are the anti-skew mechanism, structurally

Lab 03's Transform.fit computes, on the training rows only, each numeric column's mean and population variance (sum((v-mean)**2)/len, dividing by n, not n-1) and each categorical's sorted vocabulary. feature_names() fixes a stable output order (passthrough, then standardize, then one-hot blocks) so weight indices never shift. apply uses only those stored statistics: a constant column (std 0) maps to 0.0 instead of dividing by zero; an unseen category at predict time yields an all-zeros indicator block rather than crashing. Because the same fitted object runs inside create_model (to build X) and inside predict_proba (to score serving rows), the two paths cannot diverge — training-serving skew for preprocessing is prevented by object identity, not by remembering to reapply a scaler. The model is deterministic end to end: hash_split buckets rows by md5(str(key)) % 100 so a row never changes sides, weights zero-initialize, and full-batch gradient descent (g_j = (1/n)Σ(p_i − y_i)x_ij, fixed epochs and learning rate) reproduces identical weights on every run — the assert in main() proves it.

What to hold onto

Three labs, one mechanism: pin the clock, then take the max under a filter. The feature store filters candidates to at-or-before the event and maxes on (timestamp, seq). The dataflow runner filters windows to end <= watermark and fires each once. The warehouse model fits its statistics at one clock (train) and freezes them for the other (serve). Learn where the clock and the filter live in each, and every leakage, skew, and late-data bug becomes the same bug.