Feature Engineering for Production ML

At 09:00 UTC on May 1, 2026, the cluster scheduler evaluates whether training job J-204 will finish after its May 2, 09:00 UTC deadline. In this authored scenario, that deadline is the job's service-level agreement (SLA). A notebook reports 0.98 ROC-AUC, but it queried the complete event log: it used a noon heartbeat, imputed values with statistics fitted across all partitions, and read escalation flags generated after the deadline. That score measures access to future information as well as predictive ability. It is an invented warning example, not a production result.

The dataset-pipeline lesson showed how malformed rows and related records contaminate evaluation. Now we move from static records to event streams. The challenge is turning raw system telemetry into identical prediction inputs during both offline model training and live online serving.

Those inputs are features. hours_since_last_heartbeat compresses many scheduler pings into one number. Its definition must fix the time boundary, measurement units, and missing-data behavior, not just the column name.

The temporal decision barrier and point-in-time correctness

At that 09:00 evaluation, only some fields on the job record are eligible inputs. The table assumes the listed measurements were already available to the scorer. Before querying, distinguish those from fields generated after the decision. A source column existing in today's export doesn't make it historically available.

The job eventually finishes on May 3, so its later label is missed_sla = 1. Define how cancellations and still-running jobs are labeled too; a missing finish timestamp isn't automatically an on-time completion. A feature must respect the prediction timestamp. Post-SLA escalation is feature leakage when it was unavailable at this decision.^{[1]Reference 1Common Pitfalls and Recommended Practices: Data Leakagehttps://scikit-learn.org/stable/common_pitfalls.html#data-leakage}

The tempting shortcut is easy to spot in an exported table that already has the post-SLA escalation bit populated. Before running the snippet, predict which rule looks infallible and whether it could run at 09:00:

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)))

escalation rule: 1.0
backlog > 10: 0.5

On this four-row fixture, the escalation rule gets every label right because the escalation flag copies the final outcome. The backlog rule gets two right. Neither number reflects production quality, but the contrast exposes leakage: the perfect rule requires information unavailable at 09:00.

Sculley et al. identify undeclared data dependencies, hidden feedback loops, and configuration debt as ML-specific technical debt.^{[2]Reference 2Hidden Technical Debt in Machine Learning Systems.https://research.google/pubs/hidden-technical-debt-in-machine-learning-systems/} Feature definitions sit in that dependency graph. An implicit time boundary can conceal a problem that ordinary offline scores don't expose.

Work through the temporal cutoff by hand before writing a transform. Prediction time is 09:00, and heartbeats exist at 01:00, 08:30, and 12:00. Which timestamps belong in the row?

Assuming the first two had reached the scorer by 09:00, they are eligible. The latest is 08:30, so heartbeat age is 30 minutes, or 0.5 hours. The noon ping would produce $-3.0$ hours. Exclude it from this decision's inputs; clipping it to zero would still reveal a future event.

For a latest-heartbeat feature, a point-in-time join can use a backward ASOF lookup. Within the same job's eligible events, select:

$e^* = \operatorname*{arg\,max}_{e \in \mathcal{E}_{\text{job}}:\, t_{\text{event}}(e) \le t_{\text{pred}},\, t_{\text{available}}(e) \le t_{\text{pred}}} t_{\text{event}}(e)$

Define a missing policy for an empty eligible set and a deterministic revision rule for tied event times. Rolling counts or averages instead aggregate an eligible window; not every feature is the latest event's value. Databricks' time-series designation enables backward joins rather than ordinary exact-key matching.^{[3]Reference 3Work with Time Series Feature Tableshttps://docs.databricks.com/aws/en/machine-learning/feature-store/time-series}

Track two distinct clocks:

• Event time ($t_{\text{event}}$): when the telemetry event occurred.
• Availability time ($t_{\text{available}}$): when that event or revision became accessible to the scoring path. A lake's ingestion timestamp is a useful proxy only if it represents that availability.

An 08:50 heartbeat arriving at 09:15 was unavailable at 09:00. Corrections and backfills have the same problem. As checked October 10, 2026, Feast's default historical join constrains event time; a created timestamp normally resolves duplicate event times. Its filter_by_created_timestamp=True option also applies the availability cutoff when the configured column represents availability. That column must be non-null, and backend support varies.^{[4]Reference 4Point-in-time Joinshttps://github.com/feast-dev/feast/blob/master/docs/getting-started/concepts/point-in-time-joins.md}

Both offline training and online serving need identical feature definitions, code versions, and source availability rules. Labels join later and stay outside model inputs. Matching code alone won't guarantee matching source data.

Numerical transformations: scale, skewness, and missingness

Filtering out future timestamps still doesn't give you a clean model input. Raw numerical telemetry presents three operational challenges: scale differences, heavy-tailed skewness, and sensor missingness.

Standardization and robust scaling

Suppose queued_minutes ranges from 0 to 120 minutes, while transferred bytes reach trillions. Such scale differences can make gradient-based optimization poorly conditioned and change the effect of regularization. Scaling is model-dependent; an exact threshold tree can represent equivalent splits after an invertible monotone rescaling.

In standard $Z$-score normalization, you subtract the mean and divide by the standard deviation:

$z = \frac{x - \mu}{\sigma}$

Fit $\mu$ and $\sigma$ on the appropriate training partition, including inside each evaluation fold. Reusing those artifacts at inference preserves the fitted representation. A declared adaptive online scaler is another design, but must be evaluated in update order; silently fitting a new scaler on the test or serving batch changes the protocol. Define constant-feature handling when $\sigma=0$. Welford's algorithm can maintain running statistics without rescanning history:^{[5]Reference 5Note on a Method for Calculating Corrected Sums of Squares and Productshttps://doi.org/10.1080/00401706.1962.10490022}

$\mu_k = \mu_{k-1} + \frac{x_k - \mu_{k-1}}{k}, \quad M_{2,k} = M_{2,k-1} + (x_k - \mu_{k-1})(x_k - \mu_k), \quad \sigma_k^2 = \frac{M_{2,k}}{k}$

Here $M_{2,k}$ is the sum of squared deviations after $k$ observations; initialize the count and accumulator before the first update. Dividing by $k$ gives population variance for the observed values, while sample variance uses $k-1$ when $k>1$. Numerical stability doesn't make non-finite readings valid telemetry.

Telemetry can have heavy tails. In the eight-job fixture below, one 12,000-minute wait dominates the mean and standard deviation, shrinking differences among ordinary waits in standardized units.

Robust scaling uses the median and interquartile range (IQR):

$z_{\text{robust}} = \frac{x - Q_{50}}{Q_{75} - Q_{25}}$

These quantiles are less sensitive to a few extreme magnitudes, not immune to contamination or finite-sample changes. The transform neither clips outliers nor guarantees better predictions. Choose a quantile convention and a zero-IQR policy. For nonnegative values, $\log(1+x)$ compresses a right tail but still grows without bound.^{[6]Reference 6RobustScaler: median and quantile-range scalinghttps://scikit-learn.org/stable/modules/generated/sklearn.preprocessing.RobustScaler.html}

Before running the comparison, check how one 12,000-minute outlier impacts the normal job waiting 42 minutes under each transform:

import math
from statistics import quantiles

wait_times = [5.0, 12.0, 18.0, 24.0, 30.0, 42.0, 55.0, 12000.0]

mean = sum(wait_times) / len(wait_times)
std = math.sqrt(sum((x - mean) ** 2 for x in wait_times) / len(wait_times))

sorted_w = sorted(wait_times)
median = (sorted_w[3] + sorted_w[4]) / 2.0
q1, _, q3 = quantiles(wait_times, n=4, method="inclusive")
iqr = q3 - q1

normal_job = 42.0
outlier_job = 12000.0

z_normal = (normal_job - mean) / std
z_outlier = (outlier_job - mean) / std

robust_normal = (normal_job - median) / iqr
robust_outlier = (outlier_job - median) / iqr

log_normal = math.log1p(normal_job)
log_outlier = math.log1p(outlier_job)

print(f"mean: {mean:.1f}, std: {std:.1f}")
print(f"normal job 42m -> z: {z_normal:.3f}, robust: {robust_normal:.3f}, log1p: {log_normal:.3f}")
print(f"outlier job 12000m -> z: {z_outlier:.3f}, robust: {robust_outlier:.3f}, log1p: {log_outlier:.3f}")

mean: 1523.2, std: 3959.9
normal job 42m -> z: -0.374, robust: 0.522, log1p: 3.761
outlier job 12000m -> z: 2.646, robust: 416.452, log1p: 9.393

The mean is 1,523.2 and the population standard deviation is 3,959.9 minutes. Standardization places the 5-minute and 42-minute waits about 0.0093 units apart. Their information hasn't vanished: positive affine scaling is invertible in exact arithmetic. The issue is its interaction with optimization, precision, and penalties.

With inclusive interpolated quartiles, $Q_{25}=16.5$, median $=27.0$, and $Q_{75}=45.25$, so IQR $=28.75$. The 42-minute wait becomes $0.522$, but the outlier still becomes $416.452$. If your aim is to bound extreme inputs, robust scaling alone doesn't do that. The quartile method is part of the reproducible definition.^{[7]Reference 7statistics.quantiles: interpolation methodshttps://docs.python.org/3.14/library/statistics.html#statistics.quantiles}

Missingness indicators preserve the measurement gap

Missing values require an explicit policy. Filling missing queue_backlog with zero alone merges an unknown backlog with a measured empty queue. Missingness could reflect an outage, delayed collection, unsupported instrumentation, or another cause. It isn't itself proof of cluster stress.

Pairing an imputed default value with a binary missingness indicator preserves that distinction:

$x_{\text{imputed}} = 0.0, \quad x_{\text{missing}} = \begin{cases} 1 & \text{if } x \text{ is null} \\ 0 & \text{if } x \text{ is present} \end{cases}$

The extra bit lets the model distinguish measurement availability. Whether that signal predicts deadline misses must be tested, particularly if instrumentation changes after training.

Categorical encodings: one-hot pitfalls and smoothed target encoding

Numeric estimators need a representation for runner_pool. Feeding arbitrary codes 0, 1, and 2 as a scalar introduces an order and spacing into linear or distance-based models. Some estimators instead accept declared categorical features and treat their codes as category identities. Match the encoding to the estimator.

One-hot encoding and its limits

One-hot encoding resolves this by allocating an orthogonal binary dimension for each distinct category:

$\text{a100-pool} \to [1, 0, 0], \quad \text{h100-pool} \to [0, 1, 0], \quad \text{other} \to [0, 0, 1]$

Three choices matter as cardinality grows:

1. Storage and model size: Sparse one-hot matrices avoid allocating every zero, but column metadata and model parameters still grow with category count. Dense conversion can be expensive; wide encoding doesn't imply slow inference in every implementation.
2. Tree split structure: Grouping many categories via binary indicators may require multiple splits. Native categorical splitting is an alternative; neither encoding guarantees superior generalization.
3. Unseen categories: Choose rejection, a reserved bucket, an infrequent-category bucket, or an all-zero representation deliberately. Scikit-learn's handle_unknown="ignore" emits all zeros; it doesn't create an other bit. The example below explicitly maps unknown pools to a category included at fit time.^{[8]Reference 8OneHotEncoder: sparse output and unknown categorieshttps://scikit-learn.org/stable/modules/generated/sklearn.preprocessing.OneHotEncoder.html}

Out-of-fold target encoding with m-estimate smoothing

For a binary target, target encoding represents a category using an estimate of its conditional target mean:

$S_c = \mathbb{E}[y \mid X = c]$

Computing each training row's encoding from its own label introduces a target dependency. For a singleton category with missed_sla=1, the unsmoothed mean is exactly 1.0. This can encourage overfitting; it doesn't force every downstream model to memorize or achieve near-perfect accuracy.^{[9]Reference 9TargetEncoder: internal cross-fitting contracthttps://scikit-learn.org/stable/modules/generated/sklearn.preprocessing.TargetEncoder.html}

Separate two jobs:

1. Out-of-fold (K-fold) target encoding: Split the training data into $K$ partitions. For samples in fold $k$, calculate the category target means using only the remaining $K-1$ folds.
2. m-estimate smoothing: Shrink category estimates toward a prior target mean $\mu$:

$\hat{S}_c = \frac{n_c \bar{y}_c + m \mu}{n_c + m}$

Here $n_c$ is the category count and $\bar y_c$ its target mean among allowed fitting rows. For each held-out fold, compute both category statistics and the prior $\mu$ from those rows, not all partitions. The nonnegative weight $m$ controls shrinkage; it can be fixed or selected within training. Empirical Bayes estimates hyperparameters from data; a fixed $m$ isn't automatically that procedure. Binary Laplace add-one smoothing is a particular prior choice, not a synonym for every m-estimate.

Smoothing still uses an included row's label. Cross-fitting removes that direct dependency; smoothing stabilizes small-category estimates. Neither alone repairs future-label or related-entity leakage.

Notice how rare categories behave under naive encoding versus smoothed m-estimate encoding:

records = [
    {"pool": "a100-pool", "missed_sla": 0},
    {"pool": "a100-pool", "missed_sla": 0},
    {"pool": "a100-pool", "missed_sla": 1},
    {"pool": "a100-pool", "missed_sla": 0},
    {"pool": "h100-pool", "missed_sla": 1},
    {"pool": "h100-pool", "missed_sla": 1},
    {"pool": "h100-pool", "missed_sla": 1},
    {"pool": "l4-pool",   "missed_sla": 1},
]

global_mean = sum(r["missed_sla"] for r in records) / len(records)
smoothing_m = 5.0

pool_counts = {}
pool_sums = {}
for r in records:
    p = r["pool"]
    pool_counts[p] = pool_counts.get(p, 0) + 1
    pool_sums[p] = pool_sums.get(p, 0) + r["missed_sla"]

naive_encoding = {p: pool_sums[p] / pool_counts[p] for p in pool_counts}
smoothed_encoding = {
    p: (pool_sums[p] + smoothing_m * global_mean) / (pool_counts[p] + smoothing_m)
    for p in pool_counts
}

print(f"global prior SLA miss rate: {global_mean:.3f}")
print("naive encoding (l4-pool leaks target!):", {k: round(v, 3) for k, v in naive_encoding.items()})
print("smoothed m-estimate (l4-pool shrinks to prior):", {k: round(v, 3) for k, v in smoothed_encoding.items()})

global prior SLA miss rate: 0.625
naive encoding (l4-pool leaks target!): {'a100-pool': 0.25, 'h100-pool': 1.0, 'l4-pool': 1.0}
smoothed m-estimate (l4-pool shrinks to prior): {'a100-pool': 0.458, 'h100-pool': 0.766, 'l4-pool': 0.688}

For l4-pool, smoothing changes 1.0 to 0.688, but its own label contributes to both the category sum and the full-data prior. This cell demonstrates shrinkage, not a safe training representation.

Now exclude a row's entire fold from its category statistics and prior. These two explicit folds illustrate cross-fitting under an exchangeable-row assumption. Predict the encoding of l4-pool when its category is absent from the fitting fold:

folds = ([0, 2, 4, 6], [1, 3, 5, 7])

def encode_oof(data, folds, m):
    encoded = [None] * len(data)
    for held_out in folds:
        fitting = [row for i, row in enumerate(data) if i not in held_out]
        prior = sum(row["missed_sla"] for row in fitting) / len(fitting)
        for i in held_out:
            category = data[i]["pool"]
            labels = [row["missed_sla"] for row in fitting if row["pool"] == category]
            encoded[i] = (sum(labels) + m * prior) / (len(labels) + m)
    return encoded

oof = encode_oof(records, folds, smoothing_m)
changed = [dict(row) for row in records]
changed[-1]["missed_sla"] = 0

print("OOF values:", [round(value, 3) for value in oof])
print(f"l4-pool: naive={naive_encoding['l4-pool']:.3f}, smoothed={smoothed_encoding['l4-pool']:.3f}, OOF={oof[-1]:.3f}")
print("own encoding unchanged after label flip:", encode_oof(changed, folds, smoothing_m)[-1] == oof[-1])

OOF values: [0.357, 0.679, 0.357, 0.679, 0.583, 0.821, 0.583, 0.75]
l4-pool: naive=1.000, smoothed=0.688, OOF=0.750
own encoding unchanged after label flip: True

The fitting fold's prior is 0.75, not the full-data 0.625. Flipping this row's label leaves its own encoding unchanged because neither its category sum nor its prior uses that label. The model still gets different training values from the full-training mapping later used for validation or serving; that's intentional cross-fitting, not an accidental implementation mismatch.

For the scheduler's forward-time task, ordinary shuffled folds can still use future jobs' outcomes. Fit each historical encoding from earlier matured labels, respect related-job boundaries, and define a warm-up policy before any prior labels exist. Scikit-learn 1.9 accepts custom splitters in TargetEncoder, but requires each sample to appear in a validation fold exactly once; a plain TimeSeriesSplit doesn't meet that coverage rule. A dedicated forward encoder may be needed. Its fit_transform cross-fits; fit followed by transform on the same training rows doesn't.^{[9]Reference 9TargetEncoder: internal cross-fitting contracthttps://scikit-learn.org/stable/modules/generated/sklearn.preprocessing.TargetEncoder.html}

An earlier job isn't necessarily a usable labeled job. In this small replay, all jobs are unrelated and use the same runner pool. The first job's outcome is available at 08:00; the second job was scored earlier but its outcome arrives at 10:00. Predict the smoothed encoding at 09:00:

from datetime import datetime

def clock(hour):
    return datetime.fromisoformat(f"2026-05-01T{hour}:00:00+00:00")

history = [
    {"scored_at": clock("06"), "label_available_at": clock("08"), "missed_sla": 0},
    {"scored_at": clock("07"), "label_available_at": clock("10"), "missed_sla": 1},
]

def same_pool_encoding(cutoff):
    eligible = [
        row["missed_sla"] for row in history
        if row["scored_at"] < cutoff and row["label_available_at"] <= cutoff
    ]
    if not eligible:
        return None  # This fixture's warm-up policy: no target-encoded score.
    prior = sum(eligible) / len(eligible)
    m = 5.0
    return (sum(eligible) + m * prior) / (len(eligible) + m)

for hour in ("07", "09", "10"):
    print(f"{hour}:00 encoding:", same_pool_encoding(clock(hour)))
print("using every earlier job at 09:00:", sum(r["missed_sla"] for r in history) / len(history))

07:00 encoding: None
09:00 encoding: 0.0
10:00 encoding: 0.5
using every earlier job at 09:00: 0.5

At 09:00, the 1 label is future information even though its job was scored at 07:00. At 10:00, that label becomes eligible under the fixture's inclusive availability boundary. Because every job has the same category, its mean and the allowed global prior coincide, so smoothing doesn't change either value. With multiple categories, compute both statistics from the eligible rows as in the OOF cell. A frozen encoder uses its artifact's fitting cutoff; an encoder updated with newly matured labels needs an explicit update policy and replay of those versions. This cell demonstrates label eligibility, not a complete training or update pipeline.

Feature contracts and strict row validation

A mathematical definition still needs runtime enforcement. Suppose the cluster backlog feed disconnects or an unknown runner pool arrives. What should the row mean in each case?

Each surviving field requires a data type, an explicit missing-value rule, and a validated reason to exist. That's the feature contract for this job-SLA scorer:

The 180-hour cap represents an explicit engineering boundary, not an estimated statistical parameter. Learned parameters (such as a training-set mean or target encoding mappings) and fixed limits (such as this cap) both belong in the versioned preprocessing package.

These are candidate signals, not established causal explanations. Confirm that each measurement was available at cutoff. Before running the event filter, decide whether an 08:50 ping arriving at 09:15 should replace 08:30:

from datetime import datetime

prediction_time = datetime.fromisoformat("2026-05-01T09:00:00+00:00")
heartbeats = [
    ("01:00", "01:01"),
    ("08:30", "08:31"),
    ("08:50", "09:15"),
    ("12:00", "12:01"),
]

def timestamp(clock):
    return datetime.fromisoformat(f"2026-05-01T{clock}:00+00:00")

visible_heartbeats = [
    timestamp(event_time) for event_time, available_time in heartbeats
    if timestamp(event_time) <= prediction_time and timestamp(available_time) <= prediction_time
]
if not visible_heartbeats:
    raise ValueError("no heartbeat available by 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)
print("08:50 late arrival eligible:", timestamp("09:15") <= prediction_time)

visible heartbeats: 2
latest visible: 2026-05-01T08:30:00+00:00
heartbeat age hours: 0.5
08:50 late arrival eligible: False

The next builder validates a row already selected by the historical query. It doesn't inspect ingestion metadata or prove the entire source snapshot's availability. The query above and the row contract solve different problems. Extra source fields are ignored; the output contains only allowlisted keys. Check which fields survive from J-204 and which two are omitted:

from datetime import datetime, timezone
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"}
heartbeat_cap_hours = 180.0

def nonnegative_number(value, name):
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{name} must be numeric, not a string or boolean")
    value = float(value)
    if not isfinite(value) or value < 0:
        raise ValueError(f"{name} must be finite and nonnegative")
    return value

def parse_timestamp(value):
    if not isinstance(value, str):
        raise ValueError("last_heartbeat_at must be an ISO timestamp string")
    parsed = datetime.fromisoformat(value)
    if parsed.utcoffset() is None:
        raise ValueError("last_heartbeat_at must include a timezone")
    return parsed.astimezone(timezone.utc)

def make_features(row, prediction_time):
    if not isinstance(prediction_time, datetime) or prediction_time.utcoffset() is None:
        raise ValueError("prediction_time must be a timezone-aware datetime")
    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 = nonnegative_number(row["queued_minutes"], "queued_minutes")

    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 = parse_timestamp(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, heartbeat_cap_hours
    )

    queue_backlog = row.get("queue_backlog")
    if queue_backlog is not None:
        queue_backlog = nonnegative_number(queue_backlog, "queue_backlog")
    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+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+00:00",
    "finished_at": "2026-05-03T14:00:00+00:00",
    "post_sla_escalation": True,
}

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

{'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 remain outside features; their presence in the source payload isn't a rejection. The numeric validator does reject True, even though Python can cast it to 1.0. Its allowlist and types don't establish the correctness or timing of the upstream measurements.

The next cell tests boundary conditions. Predict the missing-backlog behavior, unseen runner pool handling, and how malformed inputs trigger validation errors:

missing_backlog = make_features({**job, "queue_backlog": None}, prediction_time)
assert missing_backlog["queue_backlog"] == 0.0
assert missing_backlog["queue_backlog_missing"] == 1
unseen_pool = make_features({**job, "runner_pool": "l4-pool"}, prediction_time)
assert unseen_pool["runner_pool"] == "other"
zero_queue = make_features({**job, "queued_minutes": 0}, prediction_time)
assert zero_queue["queued_minutes"] == 0.0
same_instant = make_features(
    {**job, "last_heartbeat_at": "2026-05-01T04:30:00-04:00"}, prediction_time
)
assert same_instant == features

print("missing backlog:", missing_backlog["queue_backlog"], missing_backlog["queue_backlog_missing"])
print("unseen pool:", unseen_pool["runner_pool"])
print("zero queue and equivalent timezone offset: accepted")

invalid_cases = {
    "absent pool": {"runner_pool": None},
    "future heartbeat": {"last_heartbeat_at": "2026-05-01T12:00:00+00:00"},
    "missing heartbeat": {"last_heartbeat_at": None},
    "timezone missing": {"last_heartbeat_at": "2026-05-01T08:30:00"},
    "negative queue": {"queued_minutes": -1},
    "boolean queue": {"queued_minutes": True},
    "nonfinite backlog": {"queue_backlog": float("nan")},
    "string priority": {"priority_job": "false"},
}
for name, change in invalid_cases.items():
    try:
        make_features({**job, **change}, prediction_time)
    except ValueError:
        print(name + ": rejected")
    else:
        raise AssertionError(name + " was accepted")

missing backlog: 0.0 1
unseen pool: other
zero queue and equivalent timezone offset: accepted
absent pool: rejected
future heartbeat: rejected
missing heartbeat: rejected
timezone missing: rejected
negative queue: rejected
boolean queue: rejected
nonfinite backlog: rejected
string priority: rejected

For a model consuming positional numeric vectors, fix the layout. Vectorization maps the named dictionary into ordered elements:

pool_categories = ("a100-pool", "h100-pool", "other")
numeric_order = (
    "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,
}

feature_order = tuple("pool=" + pool for pool in pool_categories) + numeric_order
vector = [int(features["runner_pool"] == pool) for pool in pool_categories]
vector += [features[name] for name in numeric_order]
print("order:", feature_order)
print("vector:", vector)

order: ('pool=a100-pool', 'pool=h100-pool', 'pool=other', 'queued_minutes', 'hours_since_last_heartbeat', 'queue_backlog', 'queue_backlog_missing', 'priority_job')
vector: [1, 0, 0, 42.0, 0.5, 18.0, 0, 0]

The final vector contains eight numerical entries. Reordering category columns can pair the active bit with a different learned weight even if the raw strings match. Package the category list, clipping thresholds, missing-data indicators, and column order with the model artifact.

Changing preprocessing changes what the model receives

Job J-188 last recorded a heartbeat 200 hours before scoring. During model training, the preprocessing pipeline clipped that age to the 180.0-hour ceiling. If an engineer later deploys a serving change that raises the cap to 360 hours without retraining the model, J-188 now enters the model as 200.0.

Sculley et al. call attention to such dependencies with the CACE principle: Changing Anything Changes Everything.^{[2]Reference 2Hidden Technical Debt in Machine Learning Systems.https://research.google/pubs/hidden-technical-debt-in-machine-learning-systems/} It is a warning about coupled systems, not a theorem that every edit changes every prediction or requires retraining.

The existing weights stay fixed after a serving-only cap edit. Inputs above 180 hours may change; inputs below both caps don't. A model that ignores heartbeat age could be unaffected. Models with interactions can respond differently across pools or backlog levels, and refitting can change other weights as well. Re-evaluate the changed preprocessing/model pair before deciding whether to retain it, restore the old cap, or retrain.

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)

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

Make the score consequence visible with a deliberately authored logistic rule. Its coefficients weren't fitted and its output isn't demonstrated to be calibrated. Predict which cap crosses the chosen alert threshold:

import math

def toy_score(raw_hours, cap, heartbeat_weight=0.02):
    age = heartbeat_age_hours(raw_hours, cap)
    logit = heartbeat_weight * age + 0.01 * 18.0 - 4.0
    return 1.0 / (1.0 + math.exp(-logit))

for cap in (180.0, 360.0):
    score = toy_score(200.0, cap)
    print(f"cap={cap:.0f}h -> score={score:.3f}, alert={score >= 0.5}")
print("fresh job unchanged:", toy_score(0.5, 180.0) == toy_score(0.5, 360.0))
print("zero heartbeat weight unchanged:", toy_score(200.0, 180.0, 0.0) == toy_score(200.0, 360.0, 0.0))

cap=180h -> score=0.445, alert=False
cap=360h -> score=0.545, alert=True
fresh job unchanged: True
zero heartbeat weight unchanged: True

An alert isn't an eviction. If a controller consumes the score and requeues jobs, it may alter future backlogs, outcomes, and training data. The direction and magnitude depend on that policy and the system; this example doesn't simulate a controller or establish a feedback effect.

For fixed fitted preprocessing, apply the training artifact at inference. Here is the explicit unknown-pool mapping in scikit-learn:

import pandas as pd
from sklearn.preprocessing import OneHotEncoder

pool_categories = ["a100-pool", "h100-pool", "other"]
encoder = OneHotEncoder(categories=[pool_categories], sparse_output=False)
training_pools = pd.DataFrame({"runner_pool": ["a100-pool", "h100-pool"]})
encoder.fit(training_pools)

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"
)
encoded = encoder.transform(mapped_pools.to_frame())

print("category order:", encoder.categories_[0].tolist())
for pool, values, age in zip(rows["runner_pool"], encoded, rows["hb_age_h"]):
    print(f"{pool}: pool bits={values.astype(int).tolist()}, hb_age_h={age}")
assert encoded[1].tolist() == [0.0, 0.0, 1.0]

category order: ['a100-pool', 'h100-pool', 'other']
a100-pool: pool bits=[1, 0, 0], hb_age_h=0.5
l4-pool: pool bits=[0, 0, 1], hb_age_h=180.0

l4-pool maps to [0, 0, 1], and its 200-hour heartbeat clips to 180.0. For this fixed preprocessing contract, reuse the fitted encoder rather than refitting on serving batches. Learned vocabularies and normalization parameters belong to the relevant training partition.

Compare the offline row with the live row

Suppose the offline table reports queue_backlog=18.0 while the captured live row contains 0.0 for the same job and cutoff. Both pass numeric validation, yet the inputs differ. This is training-serving skew; its prediction effect depends on the model.

Possible sources include:

1. Discrepancy in logic: Offline features computed using SQL/PySpark batch queries, while online features computed using Java/Go microservices with subtle differences in rounding or null handling.
2. Time-travel leakage in backfills: Offline pipelines joining the current state of a database table rather than replaying historical point-in-time state.
3. Freshness lag: Online key-value stores falling behind streaming message queues during traffic spikes.

The unified feature store pattern

A feature store can organize definitions and historical/online retrieval:

• Registry: records feature schemas, sources, and associated definitions or metadata.
• Offline path: builds training rows from retained history and configured time-aware joins.
• Online path: retrieves current entity features with a serving latency target. Feast's online stores retain the latest values, not historical versions.^{[10]Reference 10Online Storehttps://github.com/feast-dev/feast/blob/master/docs/getting-started/components/online-store.md}

A registry doesn't automatically execute equivalent transforms, synchronize updates, enforce availability-time replay, or guarantee millisecond latency. Check the product, backend, and deployment. A versioned shared preprocessing package and logged serving rows can be enough for a small application; a separate online store isn't needed for every batch model.

Product names also hide different generations. Google's current documentation marks Optimized online serving deprecated, with critical patches only from May 17, 2026 and a planned February 17, 2027 shutdown. It directs feature-serving migrations to Bigtable online serving, which supports scheduled and continuous synchronization. Check the selected generation rather than treating every Vertex AI Feature Store capability as interchangeable. Verified October 10, 2026.^{[11]Reference 11Feature Store online serving types and Optimized serving deprecationhttps://docs.cloud.google.com/gemini-enterprise-agent-platform/machine-learning/featurestore/latest/online-serving-types}

Sample both offline and online paths on identical entities before promoting a new model candidate:

def mismatches(offline, online):
    missing = object()
    differences = {}
    for key in sorted(offline.keys() | online.keys()):
        left, right = offline.get(key, missing), online.get(key, missing)
        if left != right:
            differences[key] = (
                "<missing>" if left is missing else left,
                "<missing>" if right is missing else right,
            )
    return differences

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},
    "missing field": {"runner_pool": "a100-pool", "queue_backlog": 18.0},
}

for name, online in online_rows.items():
    differences = mismatches(offline, online)
    print(name, "mismatches:", differences, "parity gate passed:", not differences)

assert mismatches({"value": None}, {}) == {"value": (None, "<missing>")}

matching mismatches: {} parity gate passed: True
stale cache mismatches: {'queue_backlog': (18.0, 0.0)} parity gate passed: False
missing field mismatches: {'priority_job': (0, '<missing>')} parity gate passed: False

The latter two rows fail this declared parity gate; the matching sample doesn't approve a whole release. Compare captured serving inputs with a replay of the same source availability, preprocessing version, and layout. A current latest-only lookup can't reconstruct an old request. This helper uses scalar equality, which doesn't distinguish False from 0 in Python; validate types separately. Cross-engine floats may need declared tolerances and explicit non-finite handling.

Monitor feature integrity continuously in production:

These checks expose symptoms, not unique causes. Freshness needs a definition per feed or feature: expected update schedule, processing watermark, or entity update age. A write elsewhere in the store doesn't prove this job's data is current. Heartbeat age is separate: an old heartbeat can be the correct value in a fully caught-up feed.

The next thresholds are authored policy choices, not industry standards. Its fallback must work with independently trustworthy inputs; switching models while reusing the same stale measurements may not help. Check the boundary cases:

import math

def scoring_policy(freshness_lag_minutes):
    if (
        isinstance(freshness_lag_minutes, bool)
        or not isinstance(freshness_lag_minutes, (int, float))
        or not math.isfinite(freshness_lag_minutes)
        or freshness_lag_minutes < 0
    ):
        return "abstain and alert"
    if freshness_lag_minutes <= 15:
        return "normal scoring"
    if freshness_lag_minutes <= 45:
        return "fallback model"
    return "abstain and alert"

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

8 minutes -> normal scoring
15 minutes -> normal scoring
30 minutes -> fallback model
45 minutes -> fallback model
90 minutes -> abstain and alert
-1 minutes -> abstain and alert
nan minutes -> abstain and alert

You have separate checks for event eligibility, row shape, layout, cap effects, parity, and freshness. They are not a complete trained pipeline: the builder emits raw numeric values, the scaling cell is separate, and neither trains an SLA classifier. To extend the lab, retain a model/preprocessor version and availability metadata, then replay a captured serving row using those exact artifacts. Ask which check would detect a wrong-but-well-typed backlog value.