LearnProduction ML SystemsBatch and Streaming Feature Pipelines

⚙️MediumMLOps & Deployment

Batch and Streaming Feature Pipelines

Replay job-SLA features at decision time from events, then keep that same meaning in online serving.

14 min read

Learning path

Step 43 of 178 in the full curriculum

Feature Engineering for Production ML Gradient Boosted Trees in Production

Personalize this lesson

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

Job J-204 already has a cutoff. At 09:00 the model may use heartbeats from 01:00 and 08:30, and it must ignore the noon heartbeat. That filter treated arrival as instantaneous, so hours_since_last_heartbeat came out to 0.5.

Event buses don't deliver on that schedule. The 08:30 heartbeat can land at 11:00. Live scoring at 09:00 never saw it. If training still uses 0.5 hours, the fit is cheating on a number serving couldn't have computed.

One event has two relevant times

Take that late 08:30 heartbeat:

Field	Example	Meaning
`event_time`	May 1, 08:30	when the scheduler recorded the heartbeat
`ingested_at`	May 1, 11:00	when this pipeline learned about the heartbeat
`prediction_time`	May 1, 09:00	when scoring asked for J-204's feature row

The heartbeat happened before the scoring request, but it arrived two hours afterward. A live prediction at 09:00 couldn't use it. A faithful replay excludes facts that occurred too late or arrived too late:

t_{\mathrm{event}} \le t_p \quad\text{and}\quad t_{\mathrm{ingested}} \le t_p

$t_{\mathrm{event}}$ is when the source says the heartbeat happened. $t_{\mathrm{ingested}}$ is when this pipeline first observed it. $t_p$ is the prediction timestamp. The first inequality blocks future-event leakage. The second reproduces what the live system knew. Keep both timestamps even if your first source rarely arrives late.

Two-clock plot for job J-204 at 09:00: H1 at event time 01:00 and ingest 01:02 sits in the known quadrant, H2 at 08:30 ingested at 11:00 is a late arrival, and H3 at 12:00 is a future event. The side panel shows as-of heartbeat age advancing from 8.0 hours at 09:00 to 3.5 hours at 12:00 to 4.0 hours at 16:00. — At `09:00`, only H1 lies inside both time boundaries, so heartbeat age is `8.0h`. H2 is the subtle failure: it occurred at `08:30` but arrived at `11:00`, so an event-time-only replay would leak the previous chapter's `0.5h` number.

Rebuild what serving knew

Build the replay in small steps. The first cell defines one on-time heartbeat, one late arrival, and one future heartbeat for job J-204. Run the cells in order.

01-inspect-event-clocks.py

from datetime import datetime
from datetime import timedelta

def dt(value):
    return datetime.fromisoformat(value)

events = [
    {
        "id": "H1",
        "job": "J-204",
        "event_time": dt("2026-05-01T01:00:00"),
        "ingested_at": dt("2026-05-01T01:02:00"),
        "source_version": 1,
    },
    {
        "id": "H2",
        "job": "J-204",
        "event_time": dt("2026-05-01T08:30:00"),
        "ingested_at": dt("2026-05-01T11:00:00"),
        "source_version": 1,
    },
    {
        "id": "H3",
        "job": "J-204",
        "event_time": dt("2026-05-01T12:00:00"),
        "ingested_at": dt("2026-05-01T12:01:00"),
        "source_version": 1,
    },
]
prediction_time = dt("2026-05-01T09:00:00")

for event in events:
    lag_minutes = int((event["ingested_at"] - event["event_time"]).total_seconds() / 60)
    print(
        f'{event["id"]} occurred={event["event_time"]:%H:%M} '
        f'arrived={event["ingested_at"]:%H:%M} lag={lag_minutes}m'
    )

Output

H1 occurred=01:00 arrived=01:02 lag=2m
H2 occurred=08:30 arrived=11:00 lag=150m
H3 occurred=12:00 arrived=12:01 lag=1m

A filter on event_time alone looks reasonable, but it leaks H2 into a replay of the 09:00 decision and recreates the dishonest 0.5h age.

02-expose-event-time-only-bug.py

def event_time_only(events, job_id, at):
    return [
        event
        for event in events
        if event["job"] == job_id and event["event_time"] <= at
    ]

def heartbeat_age_hours(visible, at):
    latest = max(visible, key=lambda event: event["event_time"])
    return (at - latest["event_time"]).total_seconds() / 3600

naive_replay = event_time_only(events, "J-204", prediction_time)
print("event-time-only ids:", [event["id"] for event in naive_replay])
print(f"event-time-only age: {heartbeat_age_hours(naive_replay, prediction_time):.1f}h")
print("leaked late heartbeat:", any(event["id"] == "H2" for event in naive_replay))

Output

event-time-only ids: ['H1', 'H2']
event-time-only age: 0.5h
leaked late heartbeat: True

Add the availability boundary. Now replay sees the same facts live serving saw at 09:00, and age jumps back to 8.0h.

03-filter-by-decision-time.py

def known_by(events, job_id, at):
    return [
        event
        for event in events
        if (
            event["job"] == job_id
            and event["event_time"] <= at
            and event["ingested_at"] <= at
        )
    ]

faithful_replay = known_by(events, "J-204", prediction_time)
print("decision-time ids:", [event["id"] for event in faithful_replay])
print(f"decision-time age: {heartbeat_age_hours(faithful_replay, prediction_time):.1f}h")
print("late heartbeat excluded:", all(event["id"] != "H2" for event in faithful_replay))

Output

decision-time ids: ['H1']
decision-time age: 8.0h
late heartbeat excluded: True

A historical training snapshot needs the latest visible heartbeat for each scoring request. H2 becomes usable after it arrives at 11:00; H3 becomes usable after 12:01. Age is computed at the cutoff from that latest timestamp, not cached from the last stream write.

04-build-as-of-rows.py

def state_order(event):
    return (event["event_time"], event.get("source_version", 0))

def revision_value(event):
    return event.get("phase", event["event_time"])

def as_of_hours(events, job_id, at):
    visible = known_by(events, job_id, at)
    if not visible:
        return "missing"
    latest_order = max(state_order(event) for event in visible)
    latest = [event for event in visible if state_order(event) == latest_order]
    if len({revision_value(event) for event in latest}) > 1:
        return "quarantined"
    hours = (at - latest[0]["event_time"]).total_seconds() / 3600
    return hours

for requested_at in [
    dt("2026-05-01T09:00:00"),
    dt("2026-05-01T12:00:00"),
    dt("2026-05-01T16:00:00"),
]:
    age = as_of_hours(events, "J-204", requested_at)
    rendered = f"{age:.1f}h" if isinstance(age, float) else age
    print(f"{requested_at:%H:%M} -> {rendered}")

Output

00 -> 8.0h
00 -> 3.5h
00 -> 4.0h

Analysts may also need corrected history of what happened by 09:00, regardless of arrival time. Keep that artifact separate: corrected history may include H2, while the decision-time replay contains only H1.

Each label has its own cutoff

Two-clock filtering still isn't enough if you build every training row from one global table snapshot. Table time travel in a Delta Lake or Apache Iceberg table restores the whole table as of one snapshot or transaction time, such as 2026-05-01 09:30:00. Every row comes from that same clock.

In ML training, each label row is a separate prediction with its own $t_p$ . Job J-201 might be scored at 09:00 while J-202 is scored at 09:30. A point-in-time join searches backward from each row's timestamp for the latest feature state visible at that moment.

If you query a single snapshot at 09:30 to construct queue_backlog for both jobs:

J-202 (predicted at 09:30) gets the correct state.
J-201 (predicted at 09:00) picks up the 09:15 backlog update. That's leakage.

The next snippet implements both joins in ordinary Python. Each label keeps its own cutoff; the leaky path reuses one snapshot time for every job.

pit-vs-snapshot-join.py

from datetime import datetime

def ts(value):
    return datetime.fromisoformat(value)

labels = [
    {"job": "J-201", "prediction_time": ts("2026-05-01T09:00:00")},
    {"job": "J-202", "prediction_time": ts("2026-05-01T09:30:00")},
]
feature_updates = [
    {"job": "J-201", "time": ts("2026-05-01T08:50:00"), "queue_backlog": 5},
    {"job": "J-201", "time": ts("2026-05-01T09:15:00"), "queue_backlog": 12},
    {"job": "J-202", "time": ts("2026-05-01T09:20:00"), "queue_backlog": 8},
]

def asof_join(labels, updates):
    rows = []
    for label in labels:
        visible = [
            update
            for update in updates
            if update["job"] == label["job"] and update["time"] <= label["prediction_time"]
        ]
        latest = max(visible, key=lambda update: update["time"]) if visible else None
        rows.append({
            "job": label["job"],
            "prediction_time": label["prediction_time"],
            "queue_backlog": None if latest is None else latest["queue_backlog"],
        })
    return rows

def snapshot_join(labels, updates, snapshot_time):
    latest_by_job = {}
    for update in updates:
        if update["time"] <= snapshot_time:
            previous = latest_by_job.get(update["job"])
            if previous is None or update["time"] > previous["time"]:
                latest_by_job[update["job"]] = update
    return [
        {
            "job": label["job"],
            "prediction_time": label["prediction_time"],
            "queue_backlog": latest_by_job[label["job"]]["queue_backlog"],
        }
        for label in labels
    ]

pit_join = asof_join(labels, feature_updates)
leakage_join = snapshot_join(labels, feature_updates, ts("2026-05-01T09:30:00"))

print("ASOF (PIT) JOIN:")
for row in pit_join:
    print(f'{row["job"]} {row["prediction_time"]:%H:%M} backlog={row["queue_backlog"]}')
print("SNAPSHOT JOIN (LEAKAGE):")
for row in leakage_join:
    print(f'{row["job"]} {row["prediction_time"]:%H:%M} backlog={row["queue_backlog"]}')

assert next(row["queue_backlog"] for row in pit_join if row["job"] == "J-201") == 5
assert next(row["queue_backlog"] for row in leakage_join if row["job"] == "J-201") == 12

Output

ASOF (PIT) JOIN:
J-201 09:00 backlog=5
J-202 09:30 backlog=8
SNAPSHOT JOIN (LEAKAGE):
J-201 09:00 backlog=12
J-202 09:30 backlog=8

The same per-row rule in pandas is merge_asof. Sort both frames by the join clock, then look backward within each job. It has to return J-201 = 5, not the leaky snapshot 12.

pandas-merge-asof-pit.py

from datetime import datetime

import pandas as pd

def ts(value):
    return datetime.fromisoformat(value)

labels = pd.DataFrame(
    [
        {"job": "J-201", "prediction_time": ts("2026-05-01T09:00:00")},
        {"job": "J-202", "prediction_time": ts("2026-05-01T09:30:00")},
    ]
).sort_values("prediction_time")
updates = pd.DataFrame(
    [
        {"job": "J-201", "prediction_time": ts("2026-05-01T08:50:00"), "queue_backlog": 5},
        {"job": "J-201", "prediction_time": ts("2026-05-01T09:15:00"), "queue_backlog": 12},
        {"job": "J-202", "prediction_time": ts("2026-05-01T09:20:00"), "queue_backlog": 8},
    ]
).sort_values("prediction_time")

pit = pd.merge_asof(
    labels,
    updates,
    by="job",
    on="prediction_time",
    direction="backward",
)

for row in pit.itertuples(index=False):
    print(f"{row.job} {row.prediction_time:%H:%M} backlog={row.queue_backlog}")
assert int(pit.loc[pit["job"].eq("J-201"), "queue_backlog"].iloc[0]) == 5

Output

J-201 09:00 backlog=5
J-202 09:30 backlog=8

Feature stores commonly wrap that backward search as historical retrieval. Feast scans backward from each entity timestamp up to that row's own time-to-live (TTL) window, not backward from "now."^{[1]Reference 1Point-in-time Joinshttps://docs.feast.dev/getting-started/concepts/point-in-time-joins} Your pipeline still owns the exact replay contract, including whether ingestion time belongs in it. The pandas call only implements the join clock you already chose.

Keep online state current

Batch replay, stream updates, and online reads are related but distinct jobs:

Job	Reads	Produces	Typical latency
batch replay	bounded event history	versioned training snapshot	minutes or hours
stream updater	newly arrived events	current feature state	seconds
online read	current feature state	one scoring row	milliseconds

Diagram showing Event log event_time + ingested_at, Feature definition v1 time boundary + missing policy, Batch replay historical training rows, and Stream updater newly arrived events. — Event log event_time + ingested_at, Feature definition v1 time boundary + missing policy, Batch replay historical training rows, and Stream updater newly arrived events.

Different stores are acceptable. Different meanings aren't. If batch computes seven-day mean backlog while online returns a one-hour queue count under the same name, offline metrics can't predict live behavior. That gap is training-serving skew.

Feast's online store is built for low-latency lookups: for each entity key it keeps only the latest feature values, not full history.^{[2]Reference 2Online Storehttps://docs.feast.dev/getting-started/components/online-store} Google Cloud's MLOps guidance describes a feature store as an optional shared repository for definitions, storage, and access across high-throughput batch and low-latency serving workloads.^{[3]Reference 3MLOps: Continuous Delivery and Automation Pipelines in Machine Learning.https://docs.cloud.google.com/architecture/mlops-continuous-delivery-and-automation-pipelines-in-machine-learning}

Store timestamps, not precomputed ages. hours_since_last_heartbeat depends on the scoring request's clock. If the stream updater writes 0.03 at 01:02 and no later heartbeat arrives, a 09:00 read must still report 8.0, not 0.03.

A late, older event mustn't overwrite newer online state. Same-time corrections need a deterministic rule too. Use the same source-owned (event_time, source_version) order in batch replay and online updates: version 2 replaces version 1, including when a stale version-1 retry arrives later. In a real source contract, require a stable sequence or revision field rather than trusting arrival order.

06-protect-latest-online-state.py

online_state = {}

def apply_latest_state(state, event):
    previous = state.get(event["job"])
    if previous is not None:
        if event["id"] == previous.get("id"):
            return "ignored duplicate event"
        if state_order(event) < state_order(previous):
            return "ignored stale event"
        if state_order(event) == state_order(previous):
            if previous.get("quarantined"):
                return "quarantine persists"
            if revision_value(event) == revision_value(previous):
                return "ignored duplicate revision"
            state[event["job"]] = {
                "job": event["job"],
                "event_time": event["event_time"],
                "source_version": event.get("source_version", 0),
                "phase": "quarantined",
                "quarantined": True,
            }
            return "quarantined conflicting revision"
    state[event["job"]] = event
    return "stored"

def online_age_hours(state, job_id, requested_at):
    current = state.get(job_id)
    if current is None:
        return "missing"
    if current.get("quarantined"):
        return "quarantined"
    return (requested_at - current["event_time"]).total_seconds() / 3600

arrivals = [
    {
        "id": "H4",
        "job": "J-204",
        "event_time": dt("2026-05-01T10:30:00"),
        "ingested_at": dt("2026-05-01T10:31:00"),
        "source_version": 1,
        "phase": "worker_scaled",
    },
    {
        "id": "H4-correction",
        "job": "J-204",
        "event_time": dt("2026-05-01T10:30:00"),
        "ingested_at": dt("2026-05-01T10:34:00"),
        "source_version": 2,
        "phase": "worker_scaled_corrected",
    },
    {
        "id": "H4-v1-retry",
        "job": "J-204",
        "event_time": dt("2026-05-01T10:30:00"),
        "ingested_at": dt("2026-05-01T10:36:00"),
        "source_version": 1,
        "phase": "worker_scaled",
    },
    {
        "id": "H1-retry",
        "job": "J-204",
        "event_time": dt("2026-05-01T01:00:00"),
        "ingested_at": dt("2026-05-01T11:05:00"),
        "source_version": 1,
    },
]

for event in arrivals:
    print(f'{event["id"]}: {apply_latest_state(online_state, event)}')
age_at_1040 = online_age_hours(online_state, "J-204", dt("2026-05-01T10:40:00"))
print(f"age at 10:40: {age_at_1040:.2f}h")
print("stored version:", online_state["J-204"]["source_version"])

Output

H4: stored
H4-correction: stored
H4-v1-retry: ignored stale event
H1-retry: ignored stale event
age at 10:40: 0.17h
stored version: 2

Latest-value state isn't enough for aggregates. A duplicated queue update can silently inflate queue_backlog unless the updater is idempotent. Idempotent means retrying the same event doesn't change the result after its first successful application. The updater should also reject impossible aggregate states instead of publishing them.

07-deduplicate-aggregate-updates.py

backlog_state = {"seen": set(), "origin_backlog": 0}

def apply_queue_update(state, event):
    if event["id"] in state["seen"]:
        return "duplicate skipped"
    next_backlog = state["origin_backlog"] + event["delta"]
    if next_backlog < 0:
        return "blocked negative backlog"
    state["seen"].add(event["id"])
    state["origin_backlog"] = next_backlog
    return "applied"

queue_updates = [
    {"id": "Q1", "delta": 3},
    {"id": "Q1", "delta": 3},
    {"id": "Q2", "delta": -1},
    {"id": "Q3", "delta": -5},
]

for update in queue_updates:
    result = apply_queue_update(backlog_state, update)
    print(f'{update["id"]}: {result}; backlog={backlog_state["origin_backlog"]}')

Output

Q1: applied; backlog=3
Q1: duplicate skipped; backlog=3
Q2: applied; backlog=2
Q3: blocked negative backlog; backlog=2

The set makes sequential retries visible, but a live updater needs a durable atomic write: record the event ID and apply its aggregate change together. Otherwise a worker can crash between those steps and a retry can lose or repeat an update. Use a transaction, conditional write, or stream processor state primitive with the same contract.

Bound lateness and freshness

A stream processor can't wait forever for missing events. It tracks a watermark: a declaration that event time has reached time $t$ , so events with timestamps at or before $t$ should already have arrived. That declaration isn't a guarantee. An older timestamp can still show up later. A product policy then decides whether the late event may update a recent window or must wait for a controlled replay.

This miniature policy accepts updates no more than 45 minutes behind the watermark. An event whose timestamp is still ahead of the watermark stays on the on-time path. A real stream processor also defines windows, triggers, state cleanup, and monitoring.

08-classify-late-events.py

def late_event_action(event_time, watermark, allowed_lateness):
    if event_time > watermark:
        return "on-time path"
    lateness = watermark - event_time
    if lateness <= allowed_lateness:
        return "accept late update"
    return "quarantine for replay"

watermark = dt("2026-05-01T12:00:00")
allowed_lateness = timedelta(minutes=45)

for event_time in [
    dt("2026-05-01T12:05:00"),
    dt("2026-05-01T11:30:00"),
    dt("2026-05-01T10:00:00"),
]:
    print(f"{event_time:%H:%M} -> {late_event_action(event_time, watermark, allowed_lateness)}")

Output

05 -> on-time path
30 -> accept late update
00 -> quarantine for replay

Freshness is a serving policy on top of that stream. The feature contract from the previous chapter already used a 15-minute normal window and a 45-minute fallback. Encode it against the stored timestamp, not against a cached age.

09-apply-serving-freshness-policy.py

def serving_mode(updated_at, requested_at):
    lag_minutes = int((requested_at - updated_at).total_seconds() / 60)
    if lag_minutes < 0:
        raise ValueError("feature update is from the future")
    if lag_minutes <= 15:
        return "normal scoring"
    if lag_minutes <= 45:
        return "fallback and log degraded mode"
    return "abstain from narrow SLA estimate"

requested_at = dt("2026-05-01T12:00:00")
for updated_at in [
    dt("2026-05-01T11:52:00"),
    dt("2026-05-01T11:30:00"),
    dt("2026-05-01T10:30:00"),
]:
    print(f"{updated_at:%H:%M} -> {serving_mode(updated_at, requested_at)}")

Output

52 -> normal scoring
30 -> fallback and log degraded mode
30 -> abstain from narrow SLA estimate

No model fixes a missing or temporally invalid input. Record freshness lag with each prediction trace so operators can distinguish model error from stale state.

Compare offline and online rows

Before promotion, replay a sample of requests offline and compare those rows with rows captured from online serving. Start with exact equality for categorical and integer fields; define tolerances explicitly if floating-point aggregations need them. For heartbeat age, both paths must compute hours at the same $t_p$ from the same stored timestamp.

10-check-offline-online-parity.py

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

correction_offline_row = {
    "hours_since_last_heartbeat": as_of_hours(arrivals, "J-204", dt("2026-05-01T10:40:00")),
}
correction_online_row = {
    "hours_since_last_heartbeat": online_age_hours(
        online_state, "J-204", dt("2026-05-01T10:40:00")
    ),
}
correction_differences = mismatches(correction_offline_row, correction_online_row)
print("same-time retry mismatches:", correction_differences)

conflicting_revisions = [
    {
        "id": "H5-a",
        "job": "J-205",
        "event_time": dt("2026-05-01T10:30:00"),
        "ingested_at": dt("2026-05-01T10:31:00"),
        "source_version": 7,
        "phase": "worker_scaled",
    },
    {
        "id": "H5-b",
        "job": "J-205",
        "event_time": dt("2026-05-01T10:30:00"),
        "ingested_at": dt("2026-05-01T10:32:00"),
        "source_version": 7,
        "phase": "worker_scale_failed",
    },
]
conflict_online_state = {}
for event in conflicting_revisions:
    apply_latest_state(conflict_online_state, event)
conflict_offline = as_of_hours(conflicting_revisions, "J-205", dt("2026-05-01T10:40:00"))
conflict_online = online_age_hours(
    conflict_online_state, "J-205", dt("2026-05-01T10:40:00")
)
assert conflict_offline == conflict_online == "quarantined"
print("equal-order conflict parity:", conflict_offline, "==", conflict_online)

for event in events:
    if event["ingested_at"] <= dt("2026-05-01T16:00:00"):
        apply_latest_state(online_state, event)

offline_row = {
    "hours_since_last_heartbeat": as_of_hours(events, "J-204", dt("2026-05-01T16:00:00")),
    "origin_backlog": backlog_state["origin_backlog"],
}
online_row = {
    "hours_since_last_heartbeat": online_age_hours(
        online_state, "J-204", dt("2026-05-01T16:00:00")
    ),
    "origin_backlog": backlog_state["origin_backlog"],
}

print("offline row:", offline_row)
print("online row:", online_row)
print("mismatches:", mismatches(offline_row, online_row))

Output

same-time retry mismatches: {}
equal-order conflict parity: quarantined == quarantined
offline row: {'hours_since_last_heartbeat': 4.0, 'origin_backlog': 2}
online row: {'hours_since_last_heartbeat': 4.0, 'origin_backlog': 2}
mismatches: {}

A stale online row produces a visible failure instead of a mysterious model regression.

11-detect-stale-online-state.py

stale_online_row = {
    "hours_since_last_heartbeat": 0.03,
    "origin_backlog": backlog_state["origin_backlog"],
}
differences = mismatches(offline_row, stale_online_row)
print("mismatches:", differences)
print("release allowed:", not differences)

Output

mismatches: {'hours_since_last_heartbeat': (4.0, 0.03)}
release allowed: False

Publish a compact receipt beside every snapshot and candidate model. It makes replay semantics inspectable instead of leaving them hidden in job code.

12-publish-replay-receipt.py

receipt = {
    "feature_definition": "job_sla_features_v1",
    "replay_boundary": "event_time <= prediction_time and ingested_at <= prediction_time",
    "offline_online_ordering": "(event_time, source_version); equal-key conflicts quarantine",
    "age_rule": "store last heartbeat timestamp; compute hours at request time",
    "aggregate_updates": "atomic event-id dedup; reject negative backlog",
    "snapshot": "job-sla-train-2026-05-01",
    "parity_samples_checked": 2,
    "freshness_policy": "sla-freshness-v1",
}

assert as_of_hours(events, "J-204", dt("2026-05-01T09:00:00")) == 8.0
assert as_of_hours(events, "J-204", dt("2026-05-01T12:00:00")) == 3.5
assert not correction_differences
assert not mismatches(offline_row, online_row)

for key, value in receipt.items():
    print(f"{key}={value}")

Output

feature_definition=job_sla_features_v1
replay_boundary=event_time <= prediction_time and ingested_at <= prediction_time
offline_online_ordering=(event_time, source_version); equal-key conflicts quarantine
age_rule=store last heartbeat timestamp; compute hours at request time
aggregate_updates=atomic event-id dedup; reject negative backlog
snapshot=job-sla-train-2026-05-01
parity_samples_checked=2
freshness_policy=sla-freshness-v1

When the contract breaks

Symptom	Cause	Fix
Training gets better after a backfill but serving doesn't	replay used facts that arrived later	filter by event time and ingestion time
Online heartbeat age stays near zero while the job goes silent	age was cached at write time	store `last_heartbeat_at` and compute hours at the scoring request
Online status moves backward after retry or same-time corrections disagree	stale or ambiguous event overwrote latest state	compare event timestamp plus source revision; quarantine unresolved ties
Backlog inflates during redelivery	aggregate update ran twice	deduplicate by stable event ID in same atomic write as aggregate change
Reliable model emits bad SLA estimates during upstream lag	freshness wasn't part of serving policy	trace age and fail to a safer response

Next Step

Continue to Gradient Boosted Trees in Production

Honest rows now carry heartbeat age and backlog as serving would have seen them. Next you'll fit a boosted tree on those columns and decide whether the validation evidence is enough to deploy.

PreviousFeature Engineering for Production ML

Share this article

X Facebook LinkedIn Bluesky Reddit Hacker News Email

References

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

MLOps: Continuous Delivery and Automation Pipelines in Machine Learning.

Google Cloud. · 2026 · Official documentation

https://docs.cloud.google.com/architecture/mlops-continuous-delivery-and-automation-pipelines-in-machine-learning

Discussion

Questions and insights from fellow learners.

Discussion loads when you reach this section.

Batch and Streaming Feature Pipelines

One event has two relevant times

A heartbeat occurred at `08:30` but wasn't ingested until `11:00`. Can a prediction made at `09:00` use it?

Rebuild what serving knew

Each label has its own cutoff

Keep online state current

A version-2 correction and a later retry of version 1 share the same event time. Which value should remain online?

Bound lateness and freshness

Compare offline and online rows

Offline replay says heartbeat age `4.0`, while the captured online row says `0.03` for the same job and cutoff. Should the model release proceed?

When the contract breaks

Mastery Check

Discussion

One event has two relevant times

A heartbeat occurred at 08:30 but wasn't ingested until 11:00. Can a prediction made at 09:00 use it?

A heartbeat occurred at 08:30 but wasn't ingested until 11:00. Can a prediction made at 09:00 use it?

Rebuild what serving knew

Each label has its own cutoff

Keep online state current

A version-2 correction and a later retry of version 1 share the same event time. Which value should remain online?

A version-2 correction and a later retry of version 1 share the same event time. Which value should remain online?

Bound lateness and freshness

Compare offline and online rows

Offline replay says heartbeat age 4.0, while the captured online row says 0.03 for the same job and cutoff. Should the model release proceed?

Offline replay says heartbeat age 4.0, while the captured online row says 0.03 for the same job and cutoff. Should the model release proceed?

When the contract breaks

Mastery Check

Discussion

A heartbeat occurred at `08:30` but wasn't ingested until `11:00`. Can a prediction made at `09:00` use it?

Offline replay says heartbeat age `4.0`, while the captured online row says `0.03` for the same job and cutoff. Should the model release proceed?