Personalize this lesson
Adapt explanations and teaching visuals to your background and preferred voice.
Versioned datasets with clean splits still aren't model-ready by themselves. A prediction model can't consume a raw event log directly. To predict whether a training job will miss its SLA, it needs a fixed row of measurements available at the moment the promise is made.
Those measurements are features. A feature such as hours_since_last_heartbeat compresses many scheduler events into one input value. Inventing columns is easy compared with making sure each value means the same thing during training and while serving live requests.
09:00, the contract emits the exact six-field row [a100-pool, 42.0, 0.5, 18.0, 0, 0]. The noon heartbeat, finished timestamp, and post-SLA escalation remain beyond the cutoff and can't enter the model input.Start With a Prediction Timestamp
Suppose the product asks at 2026-05-01 09:00: will job J-204 miss its SLA? Job heartbeats after that timestamp aren't available to the prediction service and can't appear in its training row.
| Candidate field | Known at prediction time? | Use as feature? |
|---|---|---|
| runner pool | yes | yes, categorical |
| queued minutes | yes | yes, numeric |
| minutes since most recent heartbeat | yes | yes, numeric |
| cluster queue backlog | yes | yes, numeric |
| finished timestamp | no | no, it defines the eventual label |
| post-SLA escalation | no | no, it leaks the outcome |
The label may be computed later as missed_sla = 1. A feature must be computed from history ending at the prediction timestamp. If a training row contains the post-SLA escalation, feature leakage rewards the model for reading the answer.
Sculley et al. describe production ML systems as networks of data and configuration dependencies where hidden feedback and undeclared consumers create technical debt.[1] Feature definitions are one of those dependencies: when their time boundary is unclear, the model's impressive score doesn't survive deployment.
Why must a feature definition begin with a prediction timestamp?
Answer
The timestamp defines which source facts were knowable when the prediction was made. Without it, a training row can silently include future outcomes or later updates that live serving couldn't access.
Define one row before training
Build a small contract for job-SLA prediction:
| Feature | Type | Missing rule | Why it can help |
|---|---|---|---|
queued_minutes | numeric | reject if absent | longer queue waits expose more scheduling risk |
hours_since_last_heartbeat | numeric | cap at 180 | stale heartbeats signal job risk |
queue_backlog | numeric | use measured queue only | queue pressure affects scheduling and preemption |
runner_pool | categorical | reject if absent; map unseen to other | runner pools have different networks |
priority_job | boolean | default false only when source guarantees it | priority changes SLA and retry policy |
A missing value is a product decision. Filling missing queue_backlog with zero says "unknown congestion means no congestion," which is rarely defensible. Store an additional queue_backlog_missing indicator or stop scoring until the feed recovers.
Categorical values need a policy too. If a new runner pool appears after training, the online encoder can't invent a new model column. An other bucket provides stable behavior while a new model candidate is evaluated. Missing runner pool data is different: it may signal a broken source feed, so don't silently fold it into other.
Prove the Time Boundary
A feature job should make the prediction timestamp explicit, then discard later events before aggregation. This first lab keeps only heartbeats whose event timestamps aren't later than the prediction. It deliberately assumes heartbeats arrive immediately. The next lesson adds a separate ingestion timestamp so a replay can't use an earlier heartbeat that arrived late.
1from datetime import datetime
2
3prediction_time = datetime.fromisoformat("2026-05-01T09:00:00")
4heartbeats = [
5 datetime.fromisoformat("2026-05-01T01:00:00"),
6 datetime.fromisoformat("2026-05-01T08:30:00"),
7 datetime.fromisoformat("2026-05-01T12:00:00"),
8]
9
10visible_heartbeats = [heartbeat for heartbeat in heartbeats if heartbeat <= prediction_time]
11print("visible heartbeats:", len(visible_heartbeats))
12print("latest visible:", max(visible_heartbeats).isoformat())1visible heartbeats: 2
2latest visible: 2026-05-01T08:30:00Filtering source history is necessary but not sufficient. The row contract must also enforce output keys, types, missing-value meaning, fitted category mappings, and numeric bounds. In this example:
queue_backlog=0means a measured empty queue; missing backlog emitsqueue_backlog_missing=1.- an unseen but valid runner pool maps to fitted category
other; an absent pool is rejected. - heartbeat age is capped at the fitted 180-hour limit, while a future heartbeat is rejected.
- an output allowlist blocks future-only keys such as
finished_at.
The integrated row builder below exercises these rules together. Source filtering proves values were available at the cutoff; the allowlist proves only declared keys leave the transform. Neither check replaces offline/online parity.
Build one trustworthy row
Now assemble those policies. The source record may contain post-job-SLA fields because labels need them later. The returned model row may contain only allowlisted feature keys.
1from datetime import datetime
2from math import isfinite
3
4allowed_feature_keys = {
5 "runner_pool",
6 "queued_minutes",
7 "hours_since_last_heartbeat",
8 "queue_backlog",
9 "queue_backlog_missing",
10 "priority_job",
11}
12known_runner_pools = {"a100-pool", "h100-pool"}
13future_only_keys = {"finished_at", "post_sla_escalation"}
14
15def make_features(row, prediction_time):
16 required_keys = {"queued_minutes", "last_heartbeat_at", "priority_job"}
17 missing_keys = sorted(required_keys - row.keys())
18 if missing_keys:
19 raise ValueError(f"missing required fields: {missing_keys}")
20
21 queued_minutes = float(row["queued_minutes"])
22 if not isfinite(queued_minutes) or queued_minutes <= 0:
23 raise ValueError("queued_minutes must be finite and positive")
24
25 runner_pool = row.get("runner_pool")
26 if not isinstance(runner_pool, str) or not runner_pool.strip():
27 raise ValueError("runner_pool is missing")
28 runner_pool = runner_pool.strip().casefold()
29
30 if not isinstance(row["priority_job"], bool):
31 raise ValueError("priority_job must be boolean")
32
33 last_heartbeat = datetime.fromisoformat(row["last_heartbeat_at"])
34 if last_heartbeat > prediction_time:
35 raise ValueError("last_heartbeat_at is after prediction_time")
36 heartbeat_age = min((prediction_time - last_heartbeat).total_seconds() / 3600, 180.0)
37
38 queue_backlog = row.get("queue_backlog")
39 if queue_backlog is not None:
40 queue_backlog = float(queue_backlog)
41 if not isfinite(queue_backlog) or queue_backlog < 0:
42 raise ValueError("queue_backlog must be finite and nonnegative")
43 features = {
44 "runner_pool": runner_pool if runner_pool in known_runner_pools else "other",
45 "queued_minutes": queued_minutes,
46 "hours_since_last_heartbeat": heartbeat_age,
47 "queue_backlog": 0.0 if queue_backlog is None else queue_backlog,
48 "queue_backlog_missing": int(queue_backlog is None),
49 "priority_job": int(row["priority_job"]),
50 }
51
52 blocked_fields = sorted(features.keys() - allowed_feature_keys)
53 if blocked_fields:
54 raise ValueError(f"unexpected feature fields: {blocked_fields}")
55 return features
56
57prediction_time = datetime.fromisoformat("2026-05-01T09:00:00")
58job = {
59 "job_id": "J-204",
60 "runner_pool": "a100-pool",
61 "queued_minutes": 42,
62 "queue_backlog": 18,
63 "priority_job": False,
64 "last_heartbeat_at": "2026-05-01T08:30:00",
65 "finished_at": "2026-05-03T14:00:00",
66 "post_sla_escalation": True,
67}
68
69features = make_features(job, prediction_time)
70print(features)
71print("future fields leaked:", sorted(features.keys() & future_only_keys))1{'runner_pool': 'a100-pool', 'queued_minutes': 42.0, 'hours_since_last_heartbeat': 0.5, 'queue_backlog': 18.0, 'queue_backlog_missing': 0, 'priority_job': 0}
2future fields leaked: []Probe the contract's highest-risk branches directly:
1missing_backlog = make_features({**job, "queue_backlog": None}, prediction_time)
2assert missing_backlog["queue_backlog"] == 0.0
3assert missing_backlog["queue_backlog_missing"] == 11unseen_pool = make_features({**job, "runner_pool": "l4-pool"}, prediction_time)
2assert unseen_pool["runner_pool"] == "other"1try:
2 make_features({**job, "runner_pool": None}, prediction_time)
3except ValueError as error:
4 assert str(error) == "runner_pool is missing"
5else:
6 raise AssertionError("missing runner_pool was accepted")1try:
2 make_features({**job, "last_heartbeat_at": "2026-05-01T12:00:00"}, prediction_time)
3except ValueError as error:
4 assert str(error) == "last_heartbeat_at is after prediction_time"
5else:
6 raise AssertionError("future heartbeat was accepted")1try:
2 make_features({**job, "queued_minutes": 0}, prediction_time)
3except ValueError as error:
4 assert str(error) == "queued_minutes must be finite and positive"
5else:
6 raise AssertionError("zero queued_minutes was accepted")A model still needs a fixed vector order and fitted categorical mapping. Version both beside the model artifact.
1runner_pool_code = {"a100-pool": 0, "h100-pool": 1, "other": 2}
2feature_order = (
3 "runner_pool_code",
4 "queued_minutes",
5 "hours_since_last_heartbeat",
6 "queue_backlog",
7 "queue_backlog_missing",
8 "priority_job",
9)
10features = {
11 "runner_pool": "other",
12 "queued_minutes": 42.0,
13 "hours_since_last_heartbeat": 8.0,
14 "queue_backlog": 0.0,
15 "queue_backlog_missing": 1,
16 "priority_job": 0,
17}
18
19vector = [
20 runner_pool_code[features["runner_pool"]],
21 features["queued_minutes"],
22 features["hours_since_last_heartbeat"],
23 features["queue_backlog"],
24 features["queue_backlog_missing"],
25 features["priority_job"],
26]
27print("order:", feature_order)
28print("vector:", vector)1order: ('runner_pool_code', 'queued_minutes', 'hours_since_last_heartbeat', 'queue_backlog', 'queue_backlog_missing', 'priority_job')
2vector: [2, 42.0, 8.0, 0.0, 1, 0]Why keep both queue_backlog=0 and queue_backlog_missing=1 when the source value is absent?
Answer
Measured zero and unknown congestion aren't the same fact. The indicator preserves that distinction instead of teaching the model that a broken feed means an empty queue.
Test parity before release
An offline notebook might compute queue backlog by scanning a completed daily table. The service might read an hourly cache. Even when both columns are named queue_backlog, differences in freshness or aggregation can change predictions. This failure is training-serving skew.
Feast documents point-in-time joins that reproduce feature state at each historical entity timestamp, scanning backward only within the configured TTL.[2] It also documents online stores for low-latency serving, where only the latest feature values for each entity key are stored.[3] A model release must identify the feature definition and snapshot that produced its score.
Sample offline and online paths on the same entities before promoting a model. Compare both a matching row and a stale-cache failure.
1def mismatches(offline, online):
2 keys = offline.keys() | online.keys()
3 return {key: (offline.get(key), online.get(key)) for key in keys if offline.get(key) != online.get(key)}
4
5offline = {"runner_pool": "a100-pool", "queue_backlog": 18.0, "priority_job": 0}
6online_rows = {
7 "matching": {"runner_pool": "a100-pool", "queue_backlog": 18.0, "priority_job": 0},
8 "stale cache": {"runner_pool": "a100-pool", "queue_backlog": 0.0, "priority_job": 0},
9}
10
11for name, online in online_rows.items():
12 differences = mismatches(offline, online)
13 print(name, "mismatches:", differences, "release allowed:", not differences)1matching mismatches: {} release allowed: True
2stale cache mismatches: {'queue_backlog': (18.0, 0.0)} release allowed: FalseMonitor the contract before monitoring accuracy:
| Production check | Failure it catches | Action |
|---|---|---|
| null rate by feature | upstream feed disappeared | fail closed or fallback |
| unseen-category rate | runner-pool catalog changed | collect labels and retrain |
| freshness lag | online values are stale | pause promotions |
| offline/online parity sample | transformations disagree | repair feature path |
Parity strategy also depends on where the feature is computed:
| Feature kind | Production path | Parity strategy |
|---|---|---|
| request-time | computed on the serving path from the live request and request-local context | log raw request fields and recompute the training transform offline |
| precomputed store | written to an online feature store ahead of the request | dual-write or dual-read the store path, plus point-in-time replay for training rows |
Request-path fields need log-and-recompute. Precomputed store fields need dual-write (or dual-read) plus PIT replay. Both still require a shared transformation contract; only the evidence path differs.
Freshness failures need an explicit product policy. This policy tries normal scoring, fallback, then abstention. The 15 / 45 minute cutoffs here are a local product policy, not a universal constant; other lessons may choose different numbers for the same idea. Thresholds depend on product tolerance and must be versioned with the serving path.
1def scoring_policy(freshness_lag_minutes):
2 if freshness_lag_minutes <= 15:
3 return "normal scoring"
4 if freshness_lag_minutes <= 45:
5 return "fallback model"
6 return "abstain and alert"
7
8for lag in [8, 30, 90]:
9 print(f"{lag} minutes -> {scoring_policy(lag)}")18 minutes -> normal scoring
230 minutes -> fallback model
390 minutes -> abstain and alertOffline and online transforms return different queue_backlog values for the same entity and cutoff. What should the release check do?
Answer
Block release and inspect freshness, aggregation, and feature-definition versions. Matching column names don't establish training-serving parity.