Learn AI Series):Last episode we zoomed all the way out and learned to think in systems instead of models. We drew five layers -- problem, data, features, model, serving-and-monitoring -- and I made a big fuss about how the model is just one box of six. Today we climb down into two of the outer boxes (data pipeline and feature engineering) and get our hands properly dirty. Because those boxes are where real ML projects live or die, quietly, long after the exciting model-training part is over.
Way back in episode #14 we talked about data preparation: cleaning a CSV, filling missing values, encoding categories. That was the single-machine, single-dataset version -- data as a noun, a thing sitting still on your disk. Now we treat data as a verb: the pipes that move it, validate it, version it, and feed it to a model day after boring day, at scale, without silently rotting. If ML is a Ferrari engine, data engineering is the road system. A beautiful engine on a road full of potholes (bad data), missing signs (no schemas) and dead ends (broken pipelines) gets you exactly nowhere ;-)
As promised, let's clear last episode's three tasks before we start anything new. They all lean on MLProblemFrame, should_use_ml, latency_budget and population_stability_index from episode #117, so I'm assuming those are imported and in scope.
Exercise 1 asked you to frame three scruffy real-world problems, and -- the important bit -- to force a genuine non-ML baseline and a success_metric that is NOT accuracy for each:
# MLProblemFrame and should_use_ml come straight from episode #117.
fraud = MLProblemFrame(
business_goal="Cut fraudulent-transaction losses by 40%",
ml_objective="Binary classification: is this transaction fraud?",
input_data="Amount, merchant, device, geo, velocity features",
output="Fraud probability per transaction",
success_metric="Recall at a fixed 1% review budget (catch the most fraud a human team can actually review)",
baseline="Rule: flag any charge > $500 from a new device in a new country",
latency_requirement="Real-time -- under 100ms, it sits in the checkout path",
update_frequency="Daily -- fraud patterns mutate fast",
)
tickets = MLProblemFrame(
business_goal="Route support tickets to the right team automatically",
ml_objective="Multi-class classification: which of 12 queues?",
input_data="Ticket subject, body, customer plan tier",
output="Queue label plus a confidence score",
success_metric="Macro-F1 (the rare queues matter as much as the busy ones)",
baseline="Keyword routing: 'refund' -> billing, 'password' -> auth",
latency_requirement="Batch -- a few seconds after submission is perfectly fine",
update_frequency="Monthly -- queues change slowly",
)
server_load = MLProblemFrame(
business_goal="Provision capacity before tomorrow's traffic spikes",
ml_objective="Regression: predict peak requests/sec for the next 24h",
input_data="Historical load, day-of-week, holidays, marketing calendar",
output="Predicted peak load with a confidence interval",
success_metric="MAPE, penalised harder for UNDER-prediction (a smoking server costs more than an idle one)",
baseline="Last week's same-weekday peak, plus 10%",
latency_requirement="Batch -- one prediction per day",
update_frequency="Weekly retrain",
)
for name, frame in [("fraud", fraud), ("tickets", tickets), ("load", server_load)]:
verdict = should_use_ml({
"pattern_complexity": 3 if name == "fraud" else 2,
"data_availability": 2 if name == "load" else 3,
"value_of_accuracy": 3 if name == "fraud" else 2,
"maintenance_budget": 2,
})
print(f"{name:8s}: {verdict}")
The key insight I wanted you to feel: not one of those success_metric lines says "accuracy". Fraud cares about recall inside a fixed human-review budget. Ticket routing cares about macro-F1 so the rare queues don't get trampled. Load forecasting deliberately punishes under-prediction harder than over-prediction, because the two mistakes cost wildly different amounts of money. The metric IS the problem statement wearing a disguise -- we hammered that in episode #13, and it never stops being true.
Exercise 2 was about turning a latency budget into something that yells at you when one stage gets greedy:
def latency_budget(total_ms, stages, max_fraction=0.4):
"""Split a latency budget and flag any stage that hogs more than max_fraction."""
assert abs(sum(stages.values()) - 1.0) < 1e-6, "fractions must sum to 1.0"
budget = {}
for name, frac in stages.items():
budget[name] = round(total_ms * frac, 1)
if frac > max_fraction:
print(f"WARNING: '{name}' eats {frac:.0%} of the budget (cap {max_fraction:.0%})")
budget["_headroom_ms"] = round(total_ms - sum(budget.values()), 1)
return budget
# A two-stage recommender under a 100ms budget:
plan = latency_budget(100, {
"candidate_generation": 0.50, # 50ms -- the expensive nearest-neighbour search
"ranking": 0.35, # 35ms -- the cross-attention model
"post_processing": 0.15, # 15ms -- diversity + business rules
}, max_fraction=0.4)
print(plan)
Candidate generation trips the cap immediately at 50%. And the fix is the whole point of the exercise: you do NOT try to make the nearest-neighbour search twice as fast. You push it to batch -- precompute each user's candidate set offline, refresh it hourly, and the online request only pays for ranking. Suddenly your 50ms monster is a trivial cache lookup, and your latency budget breathes again. Optimise the plumbing, not the pump.
Exercise 3 asked you to actually summon training-serving skew and watch a monitor catch it:
import numpy as np
# population_stability_index from episode #117.
rng = np.random.default_rng(0)
raw = rng.normal(100, 15, 5000) # some underlying daily signal
# "training" feature: a smooth 30-day rolling mean
train_feature = np.convolve(raw, np.ones(30) / 30, mode="valid")
# "serving" feature: accidentically a 7-day rolling mean on the SAME data
serve_feature = np.convolve(raw, np.ones(7) / 7, mode="valid")
n = min(len(train_feature), len(serve_feature))
psi = population_stability_index(train_feature[:n], serve_feature[:n])
print("PSI:", round(psi, 3))
The 7-day mean is noisier -- a wider spread -- than the smooth 30-day mean, even though it's computed on identical raw numbers. So the PSI climbs up off the floor and past the 0.1 "something is shifting" line. Nobody changed the model. Nobody changed the data. Somebody just used a different window in the serving code, and the model quietly started seeing a distribution it never trained on. In production you catch this by logging PSI between training-time and live features on a schedule -- the alarm trips before a human ever notices the business metric sagging. Right, solutions done. On to the new stuff.
There are two fundamentally different ways to move data toward a model, and picking wrong is expensive. The first is batch: process everything in scheduled chunks. Think "every night at 2 AM, chew through yesterday's events."
import json
from pathlib import Path
class BatchPipeline:
"""Process data in scheduled chunks -- the workhorse of ML data engineering."""
def __init__(self, source_dir, output_dir):
self.source_dir = Path(source_dir)
self.output_dir = Path(output_dir)
def run(self, date):
"""The classic extract-transform-load, one day at a time."""
raw_data = self.extract(date)
clean_data = self.transform(raw_data)
self.load(clean_data, date)
def extract(self, date):
"""Read raw events for a specific date."""
file_path = self.source_dir / f"events_{date}.jsonl"
events = []
with open(file_path) as f:
for line in f:
events.append(json.loads(line))
return events
def transform(self, events):
"""Clean, validate, compute features."""
processed = []
for event in events:
if not self.validate(event):
continue
processed.append({
"user_id": event["user_id"],
"item_id": event["item_id"],
"action": event["action"],
"timestamp": event["timestamp"],
"session_duration": event.get("duration", 0),
})
return processed
def validate(self, event):
"""Reject malformed events before they poison the output."""
required = ["user_id", "item_id", "action", "timestamp"]
return all(k in event for k in required)
def load(self, data, date):
"""Write processed data to output (Parquet, BigQuery, S3, ...)."""
output_path = self.output_dir / f"features_{date}.parquet"
print(f"Wrote {len(data)} records to {output_path}")
Batch is simple, debuggable, cheap, and re-runnable (a job died at 3 AM? re-run it, nobody notices). It happily covers the overwhelming majority of real ML features. But sometimes once-a-day is just too slow -- a user's behaviour in the current session, a price from thirty seconds ago -- and for that you need streaming: process each event the instant it lands.
from collections import defaultdict
class StreamProcessor:
"""Process events as they arrive, keeping a little live state per user."""
def __init__(self):
self.user_states = defaultdict(lambda: {
"click_count": 0,
"last_action_time": 0,
"session_items": [],
})
def process_event(self, event):
"""Update state and emit fresh features for one incoming event."""
user_id = event["user_id"]
state = self.user_states[user_id]
gap = event["timestamp"] - state["last_action_time"]
state["click_count"] += 1
state["last_action_time"] = event["timestamp"]
state["session_items"].append(event["item_id"])
return {
"user_id": user_id,
"clicks_this_session": state["click_count"],
"seconds_since_last_action": gap,
"unique_items_viewed": len(set(state["session_items"])),
}
def run(self, event_stream):
"""Consume events from a stream (Kafka, Kinesis, Pulsar, ...)."""
for event in event_stream:
features = self.process_event(event)
self.publish_features(features)
def publish_features(self, features):
"""Push updated features to the online store (Redis, DynamoDB, ...)."""
pass
Here's the practical reality nobody puts on a conference slide: most teams run batch for 90% of their features and reserve streaming for the 10% that genuinely, provably need real-time freshness. Streaming is not free -- you now own message queues, watermarks, out-of-order events, exactly-once headaches, and a pager. Having said that, the correct move is almost always to start batch-only, ship it, and add streaming later for the handful of features that measurably move a metric. Reaching for Kafka on day one because it feels grown-up is how you spend three months building infrastructure for a model that could have run off a nightly cron job.
Here is the single scariest thing about bad data in ML: it does not throw an exception. A broken web request 500s and someone gets paged. A column that silently turns to nulls just... degrades your model, gently, for weeks, while every dashboard stays green. So we put a guard at the door and check every record against an explicit schema before it gets anywhere near the model.
from dataclasses import dataclass
from typing import Any
@dataclass
class SchemaField:
name: str
dtype: type
nullable: bool = False
min_value: Any = None
max_value: Any = None
allowed_values: list = None
class DataValidator:
"""Validate records against a schema at the pipeline's front door."""
def __init__(self, schema):
self.schema = {f.name: f for f in schema}
def validate_record(self, record):
errors = []
for name, field in self.schema.items():
value = record.get(name)
if value is None:
if not field.nullable:
errors.append(f"{name}: required field is null")
continue
if not isinstance(value, field.dtype):
errors.append(f"{name}: expected {field.dtype}, got {type(value)}")
continue
if field.min_value is not None and value < field.min_value:
errors.append(f"{name}: {value} below min {field.min_value}")
if field.max_value is not None and value > field.max_value:
errors.append(f"{name}: {value} above max {field.max_value}")
if field.allowed_values and value not in field.allowed_values:
errors.append(f"{name}: {value} not in {field.allowed_values}")
return len(errors) == 0, errors
def validate_batch(self, records):
valid, invalid = [], []
for record in records:
ok, errors = self.validate_record(record)
(valid if ok else invalid).append(record if ok else (record, errors))
rejection_rate = len(invalid) / max(len(records), 1)
if rejection_rate > 0.05:
print(f"WARNING: {rejection_rate:.1%} rejection rate -- investigate upstream!")
return valid, invalid
schema = [
SchemaField("user_id", str, nullable=False),
SchemaField("item_id", str, nullable=False),
SchemaField("action", str, allowed_values=["click", "view", "purchase"]),
SchemaField("price", float, nullable=True, min_value=0, max_value=100000),
SchemaField("timestamp", float, min_value=1600000000),
]
validator = DataValidator(schema)
Notice that 5% rejection threshold, because it is not a random number -- it's a real-world pattern. Some rejection is always normal (buggy clients, weird edge cases, a phone that lost signal mid-request). What you actually care about is the derivative: a steady 2% that suddenly jumps to 15% overnight. That spike almost always means something upstream changed under you -- a new app version sending a renamed field, a third-party API that quietly altered its date format, a database migration that went sideways. The validator doesn't just clean data, it's an early-warning tripwire for the rest of the pipeline. In the real world you'd reach for a library like Great Expectations or Pandera rather than hand-rolling this, but the principle is identical: make the schema explicit, and fail loud.
Retrain a model, and it comes out worse. Now answer me one question: did the data change, or did the code change? If you can't answer that instantly, you are debugging blindfolded. Code we already version with git. Data we version with DVC (Data Version Control), which is basically "git for big files that don't belong in git."
# DVC keeps a tiny pointer in git and the heavy data in remote storage.
# Initialise DVC inside your existing git repo
# $ dvc init
# $ dvc remote add -d storage s3://my-bucket/dvc-storage
# Track a dataset -- this creates a small .dvc pointer file
# $ dvc add data/training_set.parquet
# -> data/training_set.parquet.dvc (goes in git)
# -> the actual parquet (goes to remote storage)
# Typical workflow, every time the data changes:
# 1. Regenerate data/training_set.parquet
# 2. $ dvc add data/training_set.parquet
# 3. $ git add data/training_set.parquet.dvc
# 4. $ git commit -m "training data: added Jan 2026 events"
# 5. $ dvc push
# And the payoff -- reproduce ANY historical dataset exactly:
# $ git checkout v1.2.0
# $ dvc checkout
# -> data/ now holds the precise bytes that shipped with v1.2.0
That last pair of commands is the whole reason DVC exists. Six months from now, when a stakeholder asks "why did the model behave differently in March?", you check out March's git commit, run dvc checkout, and you are staring at the exact data that trained that exact model. No guessing, no "I think we had the older customer file loaded." Reproducibility stops being a virtue you aspire to and becomes a command you type. Nota bene: DVC also does pipelines -- you declare your data processing as a DAG and it only reruns the stages whose inputs actually changed. Think make, but for data science.
We name-dropped feature stores last episode as the cure for training-serving skew. Time to make it concrete. The most popular open-source one is Feast, and its whole job is to guarantee that the features you train on and the features you serve on are the same features.
# Feast feature definition (feature_repo/features.py)
from feast import Entity, FeatureView, Field, FileSource
from feast.types import Float32, Int64, String
from datetime import timedelta
# An entity is the "thing" we attach features to
user = Entity(name="user_id", join_keys=["user_id"])
# A feature view groups related features from one source
user_features = FeatureView(
name="user_activity_features",
entities=[user],
ttl=timedelta(days=1), # these features go stale after a day
schema=[
Field(name="total_clicks_7d", dtype=Int64),
Field(name="avg_session_duration", dtype=Float32),
Field(name="favorite_category", dtype=String),
Field(name="days_since_last_visit", dtype=Int64),
],
source=FileSource(
path="data/user_features.parquet",
timestamp_field="feature_timestamp",
),
)
# --- Retrieve features for TRAINING (point-in-time correct, from history) ---
# training_df = store.get_historical_features(
# entity_df=entity_df, # user_ids + the timestamps we care about
# features=["user_activity_features:total_clicks_7d",
# "user_activity_features:avg_session_duration"],
# ).to_df()
# --- Retrieve the SAME features for SERVING (live, low-latency) ---
# online = store.get_online_features(
# features=["user_activity_features:total_clicks_7d"],
# entity_rows=[{"user_id": "user_123"}],
# ).to_dict()
The magic property -- and it really is the whole ballgame -- is that get_historical_features and get_online_features return identical values for the same entity at the same point in time. One definition, two access paths, zero drift. That "point-in-time correct" bit on the training path is subtler than it looks: when you build a training row for a user as they were last Tuesday, Feast makes sure you get the feature values as they existed last Tuesday, not today's values leaking backward in time (which would be a lovely way to accidentically train on the future and then wonder why production is so much worse than your eval). This is the machinery that quietly murders training-serving skew, the silent killer from episode #117.
Your model is only ever as good as its labels, and labels are expensive -- slow, tedious, and surprisingly error-prone once a bored human is on hour six. So the smart move is not "label more"; it's "label the right examples." That's active learning: let the model itself tell you which unlabeled examples would teach it the most.
import numpy as np
class ActiveLearner:
"""Pick the most informative examples to send for labeling next."""
def __init__(self, model, unlabeled_pool):
self.model = model
self.unlabeled_pool = unlabeled_pool
def uncertainty_sampling(self, n_samples):
"""Grab the examples the model is LEAST sure about."""
probs = self.model.predict_proba(self.unlabeled_pool)
# High entropy == the model is genuinely confused == high value to label
entropy = -np.sum(probs * np.log(probs + 1e-10), axis=1)
return np.argsort(entropy)[-n_samples:]
def diversity_sampling(self, n_samples):
"""Grab examples that spread across the input space (avoid redundancy)."""
embeddings = self.model.encode(self.unlabeled_pool)
from sklearn.cluster import KMeans
kmeans = KMeans(n_clusters=n_samples).fit(embeddings)
indices = []
for center in kmeans.cluster_centers_:
distances = np.linalg.norm(embeddings - center, axis=1)
indices.append(int(np.argmin(distances)))
return indices
The loop is beautifully simple: the model flags the examples it's most confused about, a human labels just those, you retrain, repeat. In practice this cuts labeling costs by somewhere between 30% and 70% compared to labeling random examples, and the effect is strongest exactly where you feel the pain most -- early in a project, when you have a mountain of unlabeled data and a molehill of labeled data. The two strategies above pull in different directions on purpose: uncertainty_sampling chases the model's blind spots, while diversity_sampling makes sure you don't waste your whole budget re-labeling ten near-identical confusing examples. Good active-learning setups blend both.
When real data is scarce, private, or painfully imbalanced (fraud is the classic -- 0.1% positives), synthetic data can fill the gap. The idea: learn the statistical shape of your real data, then generate new samples that share that shape.
import numpy as np
class SimpleTabularSynthesizer:
"""A toy synthesizer -- learn per-column distributions, then sample from them."""
def __init__(self):
self.column_stats = {}
def fit(self, data):
"""Learn a distribution from each real column."""
for col, values in data.items():
values = np.asarray(values, dtype=float)
self.column_stats[col] = {
"mean": np.mean(values),
"std": np.std(values),
"min": np.min(values),
"max": np.max(values),
}
def generate(self, n_samples):
"""Sample new synthetic rows that mimic the originals."""
synthetic = {}
for col, stats in self.column_stats.items():
samples = np.random.normal(stats["mean"], stats["std"], n_samples)
samples = np.clip(samples, stats["min"], stats["max"])
synthetic[col] = samples.tolist()
return synthetic
Now, I want to be honest with you, because this is where people get burned. The toy above is deliberately naive: it models each column independently, which means it happily destroys the correlations between columns (in real data, "age" and "income" are related; this thing treats them as strangers). The grown-up tools -- CTGAN, TVAE, and friends -- model those joint relationships, handle mixed numeric/categorical types, and can even offer privacy guarantees. But the golden rule is the same at every level of sophistication: use synthetic data to augment and to test, never as a wholesale replacement for reality. A model trained entirely on synthetic data doesn't learn the world -- it learns your synthesizer's blind spots and biases, and then confidently applies them in production. Synthetic data is seasoning, not the meal ;-)
Zoom back out and the theme of the last two episodes rhymes: the model is one box, and everything around it -- how data arrives, how it's checked, how it's versioned, how features stay consistent -- is the unglamorous machinery that decides whether your system survives contact with reality. In the next stretch of episodes we'll keep working through that outer machinery: how you prove an experiment is reproducible, how you make a trained model fast enough to actually serve, and how you keep watch on a living system so it can't rot in silence. The road system, in stead of just the engine.
Three tasks to chew on before next time. As always -- wrestle with them yourself first, and I'll walk through full solutions in the next episode.
Explain the rejection. Extend DataValidator.validate_batch so that, alongside the valid/invalid split, it tallies which field caused each rejection and prints the top offending field for the batch (e.g. "62% of rejections were price out of range"). This is the difference between a monitor that says "something's wrong" and one that says "go look at the price field."
A real windowed feature. Give StreamProcessor a genuine time-windowed feature: "clicks in the last 60 seconds," using the event timestamps rather than a running total. Then write one paragraph on what breaks when events arrive out of order (they will), and how you'd defend against it.
Measure the synthetic tax. Fit SimpleTabularSynthesizer on a small real dataset (any toy tabular set from scikit-learn works), generate an equal-sized synthetic set, then train the same simple classifier once on real data and once on synthetic data, evaluating BOTH on a held-out slice of real data. Report the accuracy gap -- that gap is the price you pay for pretending. Bonus: swap in CTGAN and watch the gap shrink.
@scipio