LearnProduction ML SystemsFeature Engineering for Production ML

⚙️MediumMLOps & Deployment

Feature Engineering for Production ML

Turn training-job events into stable prediction inputs while preventing leakage and training-serving mismatch.

14 min read

Learning path

Step 42 of 178 in the full curriculum

Dataset Pipelines and Data Quality Batch and Streaming Feature Pipelines

Personalize this lesson

Adapt explanations and teaching visuals to your background and preferred voice.

At 09:00 on May 1, 2026 the scheduler asks a blunt question: will training job J-204 miss its SLA? The scoring service doesn't read the raw event log. It needs one frozen row of measurements that already existed at that moment.

A versioned dataset with honest splits still stops one step short of that row. Schema checks keep columns typed. Grouped splits keep related records from leaking into evaluation. Those checks don't freeze what each number meant at prediction time.

Those numbers are features. hours_since_last_heartbeat compresses many scheduler pings into one input. Inventing a column is cheap. Making the same column mean the same thing in a training snapshot and in a live request is the actual work.

Start with a prediction timestamp

At that 09:00 ask, only some fields on the job record are legal inputs. Heartbeats after the timestamp aren't available to the prediction service, so they can't appear in the training row either.

Candidate field	Known at prediction time?	Use as feature?
runner pool	yes	yes, categorical
queued minutes	yes	yes, numeric
hours 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.

That leak isn't subtle in a warehouse dump that already has the later escalation bit:

leaky-escalation-rule.py

rows = [
    {"missed_sla": 1, "escalation": True, "queue_backlog": 18},
    {"missed_sla": 0, "escalation": False, "queue_backlog": 3},
    {"missed_sla": 1, "escalation": True, "queue_backlog": 9},
    {"missed_sla": 0, "escalation": False, "queue_backlog": 22},
]

def accuracy(predict):
    hits = sum(predict(row) == row["missed_sla"] for row in rows)
    return hits / len(rows)

print("escalation rule:", accuracy(lambda row: int(row["escalation"])))
print("backlog > 10:", accuracy(lambda row: int(row["queue_backlog"] > 10)))

Output

escalation rule: 1.0
backlog > 10: 0.5

On this four-row dump, predicting a miss exactly when the row was escalated is perfect. queue_backlog > 10 is a coin flip. At 09:00 the escalation field doesn't exist yet, so the perfect rule can't run. The backlog number can.

Sculley et al. treat data dependencies, hidden feedback loops, undeclared consumers, and configuration as ML-specific technical debt.^{[1]Reference 1Hidden Technical Debt in Machine Learning Systems.https://research.google/pubs/hidden-technical-debt-in-machine-learning-systems/} Feature definitions sit in that graph. When the time boundary is implicit, an impressive offline score doesn't survive serving.

Work the cutoff by hand before writing a transform. Prediction time is 09:00. Heartbeats exist at 01:00, 08:30, and 12:00. Only the first two are visible. The latest visible ping is 08:30, so heartbeat age is 30 minutes, which is 0.5 hours. The noon ping would make the age negative. Don't cap a future event to zero. Drop it.

Job J-204 cutoff at 09:00: heartbeats at 01:00 and 08:30 stay in the known panel and compress to a 0.5-hour heartbeat age, while the 12:00 heartbeat, May 3 finished_at, and post-SLA escalation sit in the future panel outside the emitted row [a100-pool, 42.0, 0.5, 18.0, 0, 0]. — The 09:00 divider is the contract. Heartbeats at 01:00 and 08:30 stay in; `hb_age_h` is hours since the 08:30 ping (`0.5`). Short headers map to `queued_minutes`, `hours_since_last_heartbeat`, and `queue_backlog_missing`. The noon heartbeat, May 3 finish, and post-SLA escalation stay out of the emitted row.

Diagram showing Job events through 09:00, Feature contract types + missing policy, Training snapshot label joins later, and Online request same computation. — Job events through 09:00, Feature contract types + missing policy, Training snapshot label joins later, and Online request same computation.

The shared contract is the only reason the training snapshot and the live request can claim to mean the same thing. Labels join later. They never travel with the model input.

Write the missing-value and category rules

A cutoff still isn't a model input. Each surviving field needs a type, a missing rule, and a reason to exist. That's the feature contract for this job-SLA scorer:

Feature	Type	Missing rule	Why it can help
`queued_minutes`	numeric	reject if absent; allow `0`	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 extra 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 stays stable 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.

The 180-hour heartbeat cap is part of the fitted contract, not a live knob. Sculley et al. call this CACE, Changing Anything Changes Everything: raising the cap at deploy time without retraining means every weight that reads heartbeat age is reading a different number.^{[1]Reference 1Hidden Technical Debt in Machine Learning Systems.https://research.google/pubs/hidden-technical-debt-in-machine-learning-systems/}

Keep only events at or before 09:00

Make the prediction timestamp explicit, then drop later events before aggregating. Keep heartbeats at or before 09:00. Treat a heartbeat as visible as soon as it occurs. The next lesson adds a separate ingestion timestamp so a replay can't use an earlier heartbeat that arrived late.

keep-visible-events.py

from datetime import datetime

prediction_time = datetime.fromisoformat("2026-05-01T09:00:00")
heartbeats = [
    datetime.fromisoformat("2026-05-01T01:00:00"),
    datetime.fromisoformat("2026-05-01T08:30:00"),
    datetime.fromisoformat("2026-05-01T12:00:00"),
]

visible_heartbeats = [heartbeat for heartbeat in heartbeats if heartbeat <= prediction_time]
print("visible heartbeats:", len(visible_heartbeats))
print("latest visible:", max(visible_heartbeats).isoformat())
print("heartbeat age hours:", (prediction_time - max(visible_heartbeats)).total_seconds() / 3600)

Output

visible heartbeats: 2
latest visible: 2026-05-01T08:30:00
heartbeat age hours: 0.5

Filtering source history isn't enough by itself. The row contract must also enforce output keys, types, missing-value meaning, fitted category mappings, and numeric bounds:

queue_backlog=0 means a measured empty queue; missing backlog emits queue_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.

Source filtering proves values were available at the cutoff. The allowlist proves only declared keys leave the transform. Neither check replaces offline/online parity.

Assemble the contract in code

The source record may contain post-job-SLA fields because labels need them later. The returned model row may contain only allowlisted feature keys.

build-feature-row.py

from datetime import datetime
from math import isfinite

allowed_feature_keys = {
    "runner_pool",
    "queued_minutes",
    "hours_since_last_heartbeat",
    "queue_backlog",
    "queue_backlog_missing",
    "priority_job",
}
known_runner_pools = {"a100-pool", "h100-pool"}
future_only_keys = {"finished_at", "post_sla_escalation"}

def make_features(row, prediction_time):
    required_keys = {"queued_minutes", "last_heartbeat_at", "priority_job"}
    missing_keys = sorted(required_keys - row.keys())
    if missing_keys:
        raise ValueError(f"missing required fields: {missing_keys}")

    queued_minutes = float(row["queued_minutes"])
    if not isfinite(queued_minutes) or queued_minutes < 0:
        raise ValueError("queued_minutes must be finite and nonnegative")

    runner_pool = row.get("runner_pool")
    if not isinstance(runner_pool, str) or not runner_pool.strip():
        raise ValueError("runner_pool is missing")
    runner_pool = runner_pool.strip().casefold()

    if not isinstance(row["priority_job"], bool):
        raise ValueError("priority_job must be boolean")

    last_heartbeat = datetime.fromisoformat(row["last_heartbeat_at"])
    if last_heartbeat > prediction_time:
        raise ValueError("last_heartbeat_at is after prediction_time")
    heartbeat_age = min((prediction_time - last_heartbeat).total_seconds() / 3600, 180.0)

    queue_backlog = row.get("queue_backlog")
    if queue_backlog is not None:
        queue_backlog = float(queue_backlog)
        if not isfinite(queue_backlog) or queue_backlog < 0:
            raise ValueError("queue_backlog must be finite and nonnegative")
    features = {
        "runner_pool": runner_pool if runner_pool in known_runner_pools else "other",
        "queued_minutes": queued_minutes,
        "hours_since_last_heartbeat": heartbeat_age,
        "queue_backlog": 0.0 if queue_backlog is None else queue_backlog,
        "queue_backlog_missing": int(queue_backlog is None),
        "priority_job": int(row["priority_job"]),
    }

    blocked_fields = sorted(features.keys() - allowed_feature_keys)
    if blocked_fields:
        raise ValueError(f"unexpected feature fields: {blocked_fields}")
    return features

prediction_time = datetime.fromisoformat("2026-05-01T09:00:00")
job = {
    "job_id": "J-204",
    "runner_pool": "a100-pool",
    "queued_minutes": 42,
    "queue_backlog": 18,
    "priority_job": False,
    "last_heartbeat_at": "2026-05-01T08:30:00",
    "finished_at": "2026-05-03T14:00:00",
    "post_sla_escalation": True,
}

features = make_features(job, prediction_time)
print(features)
print("future fields leaked:", sorted(features.keys() & future_only_keys))

Output

{'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}
future fields leaked: []

finished_at and post_sla_escalation stay on the source record for a later label join. They never appear in features. Probe the highest-risk branches next: missing backlog, unseen pool, absent pool, future heartbeat, and a negative queue time.

preserve-missing-backlog.py

missing_backlog = make_features({**job, "queue_backlog": None}, prediction_time)
assert missing_backlog["queue_backlog"] == 0.0
assert missing_backlog["queue_backlog_missing"] == 1

map-unseen-runner-pool.py

unseen_pool = make_features({**job, "runner_pool": "l4-pool"}, prediction_time)
assert unseen_pool["runner_pool"] == "other"

reject-missing-runner-pool.py

try:
    make_features({**job, "runner_pool": None}, prediction_time)
except ValueError as error:
    assert str(error) == "runner_pool is missing"
else:
    raise AssertionError("missing runner_pool was accepted")

reject-future-heartbeat.py

try:
    make_features({**job, "last_heartbeat_at": "2026-05-01T12:00:00"}, prediction_time)
except ValueError as error:
    assert str(error) == "last_heartbeat_at is after prediction_time"
else:
    raise AssertionError("future heartbeat was accepted")

accept-zero-queue-time.py

zero_queue = make_features({**job, "queued_minutes": 0}, prediction_time)
assert zero_queue["queued_minutes"] == 0.0

reject-negative-queue-time.py

try:
    make_features({**job, "queued_minutes": -1}, prediction_time)
except ValueError as error:
    assert str(error) == "queued_minutes must be finite and nonnegative"
else:
    raise AssertionError("negative queued_minutes was accepted")

A model still needs a fixed vector order and fitted categorical mapping. Version both beside the model artifact. For J-204 at 09:00 that vector is the row the figure emitted:

vectorize-features.py

runner_pool_code = {"a100-pool": 0, "h100-pool": 1, "other": 2}
feature_order = (
    "runner_pool_code",
    "queued_minutes",
    "hours_since_last_heartbeat",
    "queue_backlog",
    "queue_backlog_missing",
    "priority_job",
)
features = {
    "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,
}

vector = [
    runner_pool_code[features["runner_pool"]],
    features["queued_minutes"],
    features["hours_since_last_heartbeat"],
    features["queue_backlog"],
    features["queue_backlog_missing"],
    features["priority_job"],
]
print("order:", feature_order)
print("vector:", vector)

Output

order: ('runner_pool_code', 'queued_minutes', 'hours_since_last_heartbeat', 'queue_backlog', 'queue_backlog_missing', 'priority_job')
vector: [0, 42.0, 0.5, 18.0, 0, 0]

Same names in a different order, or a regenerated runner_pool_code map, would feed the trained weights a different story. The package has to carry the fitted mapping, the 180-hour cap, and this order with the weights.

Changing one fitted number changes the vector

The 180-hour heartbeat cap is part of the trained package. Sculley et al. call the failure CACE: Changing Anything Changes Everything.^{[1]Reference 1Hidden Technical Debt in Machine Learning Systems.https://research.google/pubs/hidden-technical-debt-in-machine-learning-systems/} Job J-188 last heartbeated 200 hours before scoring. Training clipped that age to 180.0. A later deploy that "just" raises the cap to 360 hours writes 200.0 into the same slot. Every weight that reads heartbeat age now sees a number the fit never produced.

Same 200-hour heartbeat age under two caps. The training package used a 180-hour cap, so the emitted hb_age_h is 180.0. A live service that raises the cap to 360 without retraining emits 200.0 in the same vector slot, so every weight that reads heartbeat age sees a different number. — The raw age is the same 200 hours. Training stored `180.0` because that was the fitted cap. A live raise to 360 writes `200.0` into the same slot, so the model is no longer reading the number it was trained on.

changing-fitted-cap.py

def heartbeat_age_hours(raw_hours, cap_hours):
    return min(raw_hours, cap_hours)

raw_age = 200.0
training_vector = heartbeat_age_hours(raw_age, 180.0)
live_vector = heartbeat_age_hours(raw_age, 360.0)
print("raw hours:", raw_age)
print("training hb_age_h:", training_vector)
print("live hb_age_h after cap change:", live_vector)
print("same slot, different meaning:", training_vector != live_vector)

Output

raw hours: 200.0
training hb_age_h: 180.0
live hb_age_h after cap change: 200.0
same slot, different meaning: True

pandas and scikit-learn still have to ship that same cap. OrdinalEncoder can own the runner-pool codes only if other is a fitted category, not something you invent at serve time.

library-feature-contract.py

import pandas as pd
from sklearn.preprocessing import OrdinalEncoder

rows = pd.DataFrame(
    [
        {"runner_pool": "a100-pool", "queued_minutes": 42.0, "raw_heartbeat_hours": 0.5},
        {"runner_pool": "l4-pool", "queued_minutes": 18.0, "raw_heartbeat_hours": 200.0},
    ]
)
rows["hb_age_h"] = rows["raw_heartbeat_hours"].map(lambda hours: heartbeat_age_hours(hours, 180.0))

mapped_pools = rows["runner_pool"].map(
    lambda pool: pool if pool in {"a100-pool", "h100-pool"} else "other"
)
encoder = OrdinalEncoder(categories=[["a100-pool", "h100-pool", "other"]])
rows["runner_pool_code"] = encoder.fit_transform(mapped_pools.to_frame())

print(rows[["runner_pool", "runner_pool_code", "hb_age_h"]].to_string(index=False))
print("unknown pools share the fitted other code:", set(rows.loc[rows["runner_pool"].eq("l4-pool"), "runner_pool_code"]) == {2.0})

Output

runner_pool  runner_pool_code  hb_age_h
  a100-pool               0.0       0.5
    l4-pool               2.0     180.0
unknown pools share the fitted other code: True

The pandas table didn't invent a new meaning. l4-pool still becomes fitted other (2.0), and the 200-hour heartbeat still clips to 180.0. Ship the encoder categories and the cap with the weights. Don't regenerate them from today's runner catalog.

Compare the offline row with the live row

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. That failure is training-serving skew.

A feature store is one way to ship the same contract to both paths: historical rows for training, low-latency lookups for scoring. Feast's historical retrieval does a point-in-time join: it scans backward from each entity timestamp up to the feature view's TTL, then joins the latest eligible feature row.^{[2]Reference 2Point-in-time Joinshttps://docs.feast.dev/getting-started/concepts/point-in-time-joins} Its online store keeps only the latest values per entity key, not the historical series.^{[3]Reference 3Online Storehttps://docs.feast.dev/getting-started/components/online-store} A model release must name the feature definition and snapshot that produced its score. The store doesn't invent the cutoff, missing policy, or vector order. It executes them.

Sample offline and online paths on the same entities before promoting a model. Compare both a matching row and a stale-cache failure.

check-parity.py

def mismatches(offline, online):
    keys = offline.keys() | online.keys()
    return {key: (offline.get(key), online.get(key)) for key in keys if offline.get(key) != online.get(key)}

offline = {"runner_pool": "a100-pool", "queue_backlog": 18.0, "priority_job": 0}
online_rows = {
    "matching": {"runner_pool": "a100-pool", "queue_backlog": 18.0, "priority_job": 0},
    "stale cache": {"runner_pool": "a100-pool", "queue_backlog": 0.0, "priority_job": 0},
}

for name, online in online_rows.items():
    differences = mismatches(offline, online)
    print(name, "mismatches:", differences, "release allowed:", not differences)

Output

matching mismatches: {} release allowed: True
stale cache mismatches: {'queue_backlog': (18.0, 0.0)} release allowed: False

Matching names and types aren't enough. The 18.0 vs 0.0 backlog pair is a release blocker even though both dictionaries look well-formed.

Watch the contract before watching 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 need a shared transformation contract. Only the evidence path differs.

Freshness failures need an explicit product policy. This one tries normal scoring, fallback, then abstention. The 15 / 45 minute cutoffs are a local product policy, not a universal constant. Other lessons may choose different numbers for the same idea. Version the thresholds with the serving path.

apply-freshness-policy.py

def scoring_policy(freshness_lag_minutes):
    if freshness_lag_minutes <= 15:
        return "normal scoring"
    if freshness_lag_minutes <= 45:
        return "fallback model"
    return "abstain and alert"

for lag in [8, 30, 90]:
    print(f"{lag} minutes -> {scoring_policy(lag)}")

Output

minutes -> normal scoring
minutes -> fallback model
minutes -> abstain and alert

The 09:00 row is now a contract: cutoff, types, missing rules, fitted maps, vector order, and a parity sample. It still assumes every heartbeat is visible as soon as it occurs. Real pipelines have a second clock.

Next Step

Continue to Batch and Streaming Feature Pipelines

You can now freeze one prediction-time row and reject future fields. Next you'll add a second clock, ingestion time, so a heartbeat that occurred before cutoff but arrived late can't leak into training or serving.

PreviousDataset Pipelines and Data Quality

Share this article

X Facebook LinkedIn Bluesky Reddit Hacker News Email

References

Hidden Technical Debt in Machine Learning Systems.

Sculley et al. · 2015

https://research.google/pubs/hidden-technical-debt-in-machine-learning-systems/

Point-in-time Joins

Feast Contributors · 2026 · Official documentation

https://docs.feast.dev/getting-started/concepts/point-in-time-joins

Online Store

Feast Contributors · 2026 · Official documentation

https://docs.feast.dev/getting-started/components/online-store

Discussion

Questions and insights from fellow learners.

Discussion loads when you reach this section.

Back to Topics

LearnProduction ML SystemsFeature Engineering for Production ML

⚙️MediumMLOps & Deployment

Feature Engineering for Production ML

Turn training-job events into stable prediction inputs while preventing leakage and training-serving mismatch.

14 min read

Learning path

Step 42 of 178 in the full curriculum

Dataset Pipelines and Data Quality Batch and Streaming Feature Pipelines

Personalize this lesson

Adapt explanations and teaching visuals to your background and preferred voice.

Start with a prediction timestamp

At that 09:00 ask, only some fields on the job record are legal inputs. Heartbeats after the timestamp aren't available to the prediction service, so they can't appear in the training row either.

Candidate field	Known at prediction time?	Use as feature?
runner pool	yes	yes, categorical
queued minutes	yes	yes, numeric
hours 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

That leak isn't subtle in a warehouse dump that already has the later escalation bit:

leaky-escalation-rule.py

rows = [
    {"missed_sla": 1, "escalation": True, "queue_backlog": 18},
    {"missed_sla": 0, "escalation": False, "queue_backlog": 3},
    {"missed_sla": 1, "escalation": True, "queue_backlog": 9},
    {"missed_sla": 0, "escalation": False, "queue_backlog": 22},
]

def accuracy(predict):
    hits = sum(predict(row) == row["missed_sla"] for row in rows)
    return hits / len(rows)

print("escalation rule:", accuracy(lambda row: int(row["escalation"])))
print("backlog > 10:", accuracy(lambda row: int(row["queue_backlog"] > 10)))

Output

escalation rule: 1.0
backlog > 10: 0.5

The shared contract is the only reason the training snapshot and the live request can claim to mean the same thing. Labels join later. They never travel with the model input.

Write the missing-value and category rules

A cutoff still isn't a model input. Each surviving field needs a type, a missing rule, and a reason to exist. That's the feature contract for this job-SLA scorer:

Feature	Type	Missing rule	Why it can help
`queued_minutes`	numeric	reject if absent; allow `0`	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

Keep only events at or before 09:00

keep-visible-events.py

from datetime import datetime

prediction_time = datetime.fromisoformat("2026-05-01T09:00:00")
heartbeats = [
    datetime.fromisoformat("2026-05-01T01:00:00"),
    datetime.fromisoformat("2026-05-01T08:30:00"),
    datetime.fromisoformat("2026-05-01T12:00:00"),
]

visible_heartbeats = [heartbeat for heartbeat in heartbeats if heartbeat <= prediction_time]
print("visible heartbeats:", len(visible_heartbeats))
print("latest visible:", max(visible_heartbeats).isoformat())
print("heartbeat age hours:", (prediction_time - max(visible_heartbeats)).total_seconds() / 3600)

Output

visible heartbeats: 2
latest visible: 2026-05-01T08:30:00
heartbeat age hours: 0.5

Filtering source history isn't enough by itself. The row contract must also enforce output keys, types, missing-value meaning, fitted category mappings, and numeric bounds:

queue_backlog=0 means a measured empty queue; missing backlog emits queue_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.

Source filtering proves values were available at the cutoff. The allowlist proves only declared keys leave the transform. Neither check replaces offline/online parity.

Assemble the contract in code

The source record may contain post-job-SLA fields because labels need them later. The returned model row may contain only allowlisted feature keys.

build-feature-row.py

from datetime import datetime
from math import isfinite

allowed_feature_keys = {
    "runner_pool",
    "queued_minutes",
    "hours_since_last_heartbeat",
    "queue_backlog",
    "queue_backlog_missing",
    "priority_job",
}
known_runner_pools = {"a100-pool", "h100-pool"}
future_only_keys = {"finished_at", "post_sla_escalation"}

def make_features(row, prediction_time):
    required_keys = {"queued_minutes", "last_heartbeat_at", "priority_job"}
    missing_keys = sorted(required_keys - row.keys())
    if missing_keys:
        raise ValueError(f"missing required fields: {missing_keys}")

    queued_minutes = float(row["queued_minutes"])
    if not isfinite(queued_minutes) or queued_minutes < 0:
        raise ValueError("queued_minutes must be finite and nonnegative")

    runner_pool = row.get("runner_pool")
    if not isinstance(runner_pool, str) or not runner_pool.strip():
        raise ValueError("runner_pool is missing")
    runner_pool = runner_pool.strip().casefold()

    if not isinstance(row["priority_job"], bool):
        raise ValueError("priority_job must be boolean")

    last_heartbeat = datetime.fromisoformat(row["last_heartbeat_at"])
    if last_heartbeat > prediction_time:
        raise ValueError("last_heartbeat_at is after prediction_time")
    heartbeat_age = min((prediction_time - last_heartbeat).total_seconds() / 3600, 180.0)

    queue_backlog = row.get("queue_backlog")
    if queue_backlog is not None:
        queue_backlog = float(queue_backlog)
        if not isfinite(queue_backlog) or queue_backlog < 0:
            raise ValueError("queue_backlog must be finite and nonnegative")
    features = {
        "runner_pool": runner_pool if runner_pool in known_runner_pools else "other",
        "queued_minutes": queued_minutes,
        "hours_since_last_heartbeat": heartbeat_age,
        "queue_backlog": 0.0 if queue_backlog is None else queue_backlog,
        "queue_backlog_missing": int(queue_backlog is None),
        "priority_job": int(row["priority_job"]),
    }

    blocked_fields = sorted(features.keys() - allowed_feature_keys)
    if blocked_fields:
        raise ValueError(f"unexpected feature fields: {blocked_fields}")
    return features

prediction_time = datetime.fromisoformat("2026-05-01T09:00:00")
job = {
    "job_id": "J-204",
    "runner_pool": "a100-pool",
    "queued_minutes": 42,
    "queue_backlog": 18,
    "priority_job": False,
    "last_heartbeat_at": "2026-05-01T08:30:00",
    "finished_at": "2026-05-03T14:00:00",
    "post_sla_escalation": True,
}

features = make_features(job, prediction_time)
print(features)
print("future fields leaked:", sorted(features.keys() & future_only_keys))

Output

{'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}
future fields leaked: []

preserve-missing-backlog.py

missing_backlog = make_features({**job, "queue_backlog": None}, prediction_time)
assert missing_backlog["queue_backlog"] == 0.0
assert missing_backlog["queue_backlog_missing"] == 1

map-unseen-runner-pool.py

unseen_pool = make_features({**job, "runner_pool": "l4-pool"}, prediction_time)
assert unseen_pool["runner_pool"] == "other"

reject-missing-runner-pool.py

try:
    make_features({**job, "runner_pool": None}, prediction_time)
except ValueError as error:
    assert str(error) == "runner_pool is missing"
else:
    raise AssertionError("missing runner_pool was accepted")

reject-future-heartbeat.py

try:
    make_features({**job, "last_heartbeat_at": "2026-05-01T12:00:00"}, prediction_time)
except ValueError as error:
    assert str(error) == "last_heartbeat_at is after prediction_time"
else:
    raise AssertionError("future heartbeat was accepted")

accept-zero-queue-time.py

zero_queue = make_features({**job, "queued_minutes": 0}, prediction_time)
assert zero_queue["queued_minutes"] == 0.0

reject-negative-queue-time.py

try:
    make_features({**job, "queued_minutes": -1}, prediction_time)
except ValueError as error:
    assert str(error) == "queued_minutes must be finite and nonnegative"
else:
    raise AssertionError("negative queued_minutes was accepted")

A model still needs a fixed vector order and fitted categorical mapping. Version both beside the model artifact. For J-204 at 09:00 that vector is the row the figure emitted:

vectorize-features.py

runner_pool_code = {"a100-pool": 0, "h100-pool": 1, "other": 2}
feature_order = (
    "runner_pool_code",
    "queued_minutes",
    "hours_since_last_heartbeat",
    "queue_backlog",
    "queue_backlog_missing",
    "priority_job",
)
features = {
    "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,
}

vector = [
    runner_pool_code[features["runner_pool"]],
    features["queued_minutes"],
    features["hours_since_last_heartbeat"],
    features["queue_backlog"],
    features["queue_backlog_missing"],
    features["priority_job"],
]
print("order:", feature_order)
print("vector:", vector)

Output

order: ('runner_pool_code', 'queued_minutes', 'hours_since_last_heartbeat', 'queue_backlog', 'queue_backlog_missing', 'priority_job')
vector: [0, 42.0, 0.5, 18.0, 0, 0]

Changing one fitted number changes the vector

changing-fitted-cap.py

def heartbeat_age_hours(raw_hours, cap_hours):
    return min(raw_hours, cap_hours)

raw_age = 200.0
training_vector = heartbeat_age_hours(raw_age, 180.0)
live_vector = heartbeat_age_hours(raw_age, 360.0)
print("raw hours:", raw_age)
print("training hb_age_h:", training_vector)
print("live hb_age_h after cap change:", live_vector)
print("same slot, different meaning:", training_vector != live_vector)

Output

raw hours: 200.0
training hb_age_h: 180.0
live hb_age_h after cap change: 200.0
same slot, different meaning: True

pandas and scikit-learn still have to ship that same cap. OrdinalEncoder can own the runner-pool codes only if other is a fitted category, not something you invent at serve time.

library-feature-contract.py

import pandas as pd
from sklearn.preprocessing import OrdinalEncoder

rows = pd.DataFrame(
    [
        {"runner_pool": "a100-pool", "queued_minutes": 42.0, "raw_heartbeat_hours": 0.5},
        {"runner_pool": "l4-pool", "queued_minutes": 18.0, "raw_heartbeat_hours": 200.0},
    ]
)
rows["hb_age_h"] = rows["raw_heartbeat_hours"].map(lambda hours: heartbeat_age_hours(hours, 180.0))

mapped_pools = rows["runner_pool"].map(
    lambda pool: pool if pool in {"a100-pool", "h100-pool"} else "other"
)
encoder = OrdinalEncoder(categories=[["a100-pool", "h100-pool", "other"]])
rows["runner_pool_code"] = encoder.fit_transform(mapped_pools.to_frame())

print(rows[["runner_pool", "runner_pool_code", "hb_age_h"]].to_string(index=False))
print("unknown pools share the fitted other code:", set(rows.loc[rows["runner_pool"].eq("l4-pool"), "runner_pool_code"]) == {2.0})

Output

runner_pool  runner_pool_code  hb_age_h
  a100-pool               0.0       0.5
    l4-pool               2.0     180.0
unknown pools share the fitted other code: True

Compare the offline row with the live row

Sample offline and online paths on the same entities before promoting a model. Compare both a matching row and a stale-cache failure.

check-parity.py

def mismatches(offline, online):
    keys = offline.keys() | online.keys()
    return {key: (offline.get(key), online.get(key)) for key in keys if offline.get(key) != online.get(key)}

offline = {"runner_pool": "a100-pool", "queue_backlog": 18.0, "priority_job": 0}
online_rows = {
    "matching": {"runner_pool": "a100-pool", "queue_backlog": 18.0, "priority_job": 0},
    "stale cache": {"runner_pool": "a100-pool", "queue_backlog": 0.0, "priority_job": 0},
}

for name, online in online_rows.items():
    differences = mismatches(offline, online)
    print(name, "mismatches:", differences, "release allowed:", not differences)

Output

matching mismatches: {} release allowed: True
stale cache mismatches: {'queue_backlog': (18.0, 0.0)} release allowed: False

Matching names and types aren't enough. The 18.0 vs 0.0 backlog pair is a release blocker even though both dictionaries look well-formed.

Watch the contract before watching 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 need a shared transformation contract. Only the evidence path differs.

apply-freshness-policy.py

def scoring_policy(freshness_lag_minutes):
    if freshness_lag_minutes <= 15:
        return "normal scoring"
    if freshness_lag_minutes <= 45:
        return "fallback model"
    return "abstain and alert"

for lag in [8, 30, 90]:
    print(f"{lag} minutes -> {scoring_policy(lag)}")

Output

minutes -> normal scoring
minutes -> fallback model
minutes -> abstain and alert

Next Step

Continue to Batch and Streaming Feature Pipelines

PreviousDataset Pipelines and Data Quality

Share this article

X Facebook LinkedIn Bluesky Reddit Hacker News Email

References

Hidden Technical Debt in Machine Learning Systems.

Sculley et al. · 2015

https://research.google/pubs/hidden-technical-debt-in-machine-learning-systems/

Point-in-time Joins

Feast Contributors · 2026 · Official documentation

https://docs.feast.dev/getting-started/concepts/point-in-time-joins

Online Store

Feast Contributors · 2026 · Official documentation

https://docs.feast.dev/getting-started/components/online-store

Discussion

Questions and insights from fellow learners.

Discussion loads when you reach this section.

Feature Engineering for Production ML

Start with a prediction timestamp

Why must a feature definition begin with a prediction timestamp?

Write the missing-value and category rules

Keep only events at or before 09:00

Assemble the contract in code

Why keep both `queue_backlog=0` and `queue_backlog_missing=1` when the source value is absent?

Changing one fitted number changes the vector

A serving change raises the heartbeat-age cap from 180 hours to 360 hours and keeps the same model file. What broke?

Compare the offline row with the live row

Offline and online transforms return different `queue_backlog` values for the same entity and cutoff. What should the release check do?

Mastery Check

Discussion

Feature Engineering for Production ML

Start with a prediction timestamp

Why must a feature definition begin with a prediction timestamp?

Write the missing-value and category rules

Keep only events at or before 09:00

Assemble the contract in code

Why keep both `queue_backlog=0` and `queue_backlog_missing=1` when the source value is absent?

Changing one fitted number changes the vector

A serving change raises the heartbeat-age cap from 180 hours to 360 hours and keeps the same model file. What broke?

Compare the offline row with the live row

Offline and online transforms return different `queue_backlog` values for the same entity and cutoff. What should the release check do?

Mastery Check

Discussion

Why must a feature definition begin with a prediction timestamp?

Why must a feature definition begin with a prediction timestamp?

Why keep both queue_backlog=0 and queue_backlog_missing=1 when the source value is absent?

Why keep both queue_backlog=0 and queue_backlog_missing=1 when the source value is absent?

A serving change raises the heartbeat-age cap from 180 hours to 360 hours and keeps the same model file. What broke?

A serving change raises the heartbeat-age cap from 180 hours to 360 hours and keeps the same model file. What broke?

Offline and online transforms return different queue_backlog values for the same entity and cutoff. What should the release check do?

Offline and online transforms return different queue_backlog values for the same entity and cutoff. What should the release check do?

Mastery Check

Discussion

Why must a feature definition begin with a prediction timestamp?

Why must a feature definition begin with a prediction timestamp?

Why keep both queue_backlog=0 and queue_backlog_missing=1 when the source value is absent?

Why keep both queue_backlog=0 and queue_backlog_missing=1 when the source value is absent?

A serving change raises the heartbeat-age cap from 180 hours to 360 hours and keeps the same model file. What broke?

A serving change raises the heartbeat-age cap from 180 hours to 360 hours and keeps the same model file. What broke?

Offline and online transforms return different queue_backlog values for the same entity and cutoff. What should the release check do?

Offline and online transforms return different queue_backlog values for the same entity and cutoff. What should the release check do?

Mastery Check

Discussion

Why keep both `queue_backlog=0` and `queue_backlog_missing=1` when the source value is absent?

Offline and online transforms return different `queue_backlog` values for the same entity and cutoff. What should the release check do?

Why keep both `queue_backlog=0` and `queue_backlog_missing=1` when the source value is absent?

Offline and online transforms return different `queue_backlog` values for the same entity and cutoff. What should the release check do?