Streaming Inference
Continuous events drive predictions: the model sits inside a stream processor, features are state kept per key, and ordering, late events and where the model sits in the topology decide correctness more than the weights do.
The problem, the obvious approach, and why it breaks
Every lesson starts where the work starts: someone has a problem, and the first model that comes to mind looks fine offline.
Events arrive continuously and every one may change a prediction. When does the model belong inside the stream rather than behind an endpoint or in a nightly job — and what does the stream's semantics do to the prediction?
A logistics company wants an updated delay estimate for every shipment each time a scan event arrives, pushed to customers within seconds. The first attempt called the online endpoint from a consumer; it worked in staging and fell behind by hours on the first busy day, then produced estimates that went backwards in time.
Consume the event stream and call the online model endpoint once per event. The endpoint already exists, the consumer is a loop, and the freshness is as good as the stream.
The endpoint fetches the shipment's features from a store on each call; on a busy day the consumer cannot keep up with the event rate, lag grows, and the "real-time" estimate is hours stale.
- The endpoint fetches the shipment's features from a store on each call; on a busy day the consumer cannot keep up with the event rate, lag grows, and the "real-time" estimate is hours stale.
- Two scans for one shipment are processed by different consumer instances in different orders; the second-to-last estimate overwrites the last, and the customer sees the estimate jump backwards.
- A scan arriving ten minutes late carries an event time earlier than the estimate already published; the endpoint has no notion of event time and treats it as new information from now.
- The lane aggregate — recent delays on this route — is computed in the feature store from processing time, while training computed it from event time; the same shipment gets a different feature in production (Train / Serve Skew).
What is being predicted, and from what data
This domain leads with these two. A target nobody defined precisely is a label nobody can trust, and a dataset nobody can describe is a model nobody can debug.
- Predict the remaining time to delivery for a shipment given its scan history so far; the label is the realised delivery time, known when the last scan arrives.
- The decision is the estimate shown to the customer and used to trigger a delay notification, updated on every event rather than on request or on a schedule.
- One example is one shipment at one point in its scan sequence: the events so far, time since the last scan, route segment, carrier, and aggregates over recent shipments on the same lane.
- Events arrive on a partitioned log keyed by shipment id, from scanners with unsynchronised clocks; some arrive minutes late and occasionally out of order.
- Training examples were built by replaying historical event logs and cutting each shipment's sequence at every scan, with the lane aggregates computed as of the event's time.
How it actually works
Precisely enough to predict its behaviour — not a framework API.
- In streaming inference the model runs inside a stream processor as an operator. Events are partitioned by key so every event for one shipment reaches the same operator instance in order, and the features are state that instance keeps per key — the scan history, the running aggregates — updated on each event and read locally, with no fetch.
- The processor distinguishes event time from processing time and tracks a watermark: its estimate of how far event time has advanced. A late event is one whose event time is behind the watermark; the processor's policy decides whether it updates state, triggers a new prediction, or is dropped. Ordering within a key is guaranteed by partitioning; ordering across keys is not needed.
- Where the model sits in the topology decides what it sees. Placed after the per-key state update, it scores a consistent view of the shipment; placed before a windowed aggregate, it scores partial windows; placed after a join with another stream, it inherits that join's late-data semantics (Stream Joins).
- Throughput comes from partition parallelism, not from the endpoint's concurrency; the model is loaded in every operator instance, and its per-event cost sets the ceiling on events per second per partition.
The model as an operator
The topology decides what the model sees. Events are partitioned by shipment id so each shipment's scans arrive at one operator instance in order; that instance keeps the shipment's history and running aggregates as local state, updates them on each event, and runs the model against the updated state at that event's time. The prediction is emitted with the event time, a per-key sequence and the model digest.
No fetch, no endpoint, no retry. The cost is that everything the endpoint would have fetched now lives in operator state, and the state has to be checkpointed, evicted and rebuilt on restart. The depth on state, watermarks and checkpoints belongs to the data platform; what this domain adds is that the model's correctness depends on them.
Late events: training and production must agree
A scan that arrives ten minutes late is information the shipment's history now contains. The stream processor has a policy: update state and re-predict, update state silently, or drop. Training replayed the full historical log, where every event is present and in order — so the training model saw late events as if they had been on time. Whatever the production policy is, the training replay has to apply the same one, or the model is trained on histories production never assembles.
This is the assumption a streaming deployment most often violates without noticing. The symptom is estimate error concentrated on shipments with late scans, which is also where the product's promise matters most.
For a shipment at a given scan, the state the operator scores from contains the same events, with the same late-event policy applied, as the training replay assembled for that point in that shipment's history.
holds when Training replays the event log through the same watermark and late-data rules the job runs with, and the per-key ordering guarantee holds.
breaks when The watermark is tightened in production for latency; a repartition breaks per-key ordering during a scale-out; the training replay reads a cleaned table where late events were already merged in order.
respond Align the policy on both sides first; only then consider whether the model needs retraining on histories assembled under the production policy.
When the stream is the right place
Streaming inference is justified when the prediction must change on every event, the features are the event history itself, and the event rate exceeds what per-event fetching can sustain. Absent any one of those, a cheaper mode exists: a table fed by the stream and scored in batch, or an endpoint that fetches from a store the stream keeps fresh.
The decision below is the local form of the mode choice; the general one, including hybrids, is the Choosing the Inference Mode lesson and the decision tool.
Should the model run inside the stream processor, or beside it?
when Every event changes the prediction, the features are per-key event history, and per-event fetches cannot keep up with the rate.
cost Stateful operators, checkpoints, watermark policy, partition balance, and a feature implementation tied to the stream framework.
when Predictions are needed on demand rather than on every event, and the freshness the stream provides to the store is enough.
cost A feature store with a freshness bound and the online mode's timeout and fallback problems.
when The decision is made on a rhythm — hourly, daily — and events inside the cycle do not change it.
cost A staleness window, and a consumer that must not read a half-published run.
How to build it
Most important first.
- Key the stream by the entity the prediction is about, keep the features as operator state updated by each event, and run the model in the same operator so the feature and the prediction are computed from the same state at the same event time.
- Define the late-event policy explicitly and make training match it: if the processor drops events behind the watermark by more than a bound, the training replay must drop them too, or the model learns from information production never has.
- Emit predictions with the event time and a monotonic sequence per key, and have the consumer keep the latest by sequence, not by arrival, so a reordered publish cannot move an estimate backwards.
- Checkpoint operator state with the input offsets so a restart replays from a consistent point (Checkpointing in the stream sense), and version the model in the state so a redeploy with a new artifact can be told apart in the output.
What to measure
Which number actually maps to the decision — and which numbers look relevant and are not.
- Consumer lag in event time — the gap between the watermark and now — which is the real freshness of the estimate; a growing lag is the first sign the operator cannot keep up.
- Error of the estimate against realised delivery, bucketed by how many events had arrived when it was made and by whether late events were involved; that is where a late-data policy mismatch between training and production shows.
- Not the endpoint's request latency, which no longer exists, and not events per second in isolation — throughput that is achieved by dropping late events is a quality cost.
What must stay true after deployment
The field this whole domain exists for. A model is a set of assumptions with weights attached; these are the ones a monitor or a test should be checking.
- Every event for one key reaches the same operator instance in event-time order, or the late-event policy handles the exceptions in the same way the training replay did.
- The features the operator computes from its state at event time match, to a tolerance, the features the training replay computed for the same point in the same shipment's history.
- Event-time lag stays within the freshness the product promises under peak event rates and during recovery from a restart.
- Offline: replay a recorded day of events through the streaming job and diff its emitted features and predictions against the training pipeline's replay of the same day, including the late events.
- Online: per-partition lag in event time, the late-event drop rate, and out-of-order publishes caught by the sequence check at the consumer, all alerted.
- Over time: estimate error by number-of-events-seen and by late-event involvement, on realised deliveries; a widening error on late-involved shipments points at the watermark policy.
What can go wrong
- A hot key — one lane with a burst of scans — saturates one partition while the others idle; lag on that partition alone grows, and the aggregate lag metric looks fine (Data Skew).
- The watermark is configured tight to keep latency low and drops a significant share of legitimate late scans; the model never sees them, but training did.
- Operator state grows without bound because completed shipments are never evicted; a restart takes an hour to restore state and the estimate is stale for that hour.
- The artifact is upgraded by a rolling restart, and for a few minutes different partitions score with different models; the output stream cannot tell them apart unless the model digest is in every record.
- Streaming inference is the most operationally demanding mode: stateful operators, checkpoints, watermarks and partition balance are all things that the batch and online modes do not have.
- Keeping features as operator state removes the fetch and ties the feature computation to the stream framework; the same feature cannot be reused by a batch job without a second implementation or a unified definition (Feature Stores).
- A tighter watermark lowers latency and raises the late-drop rate; there is no setting that is right for every lane, and the choice is a quality decision made in a configuration file.
- "Streaming is just online inference triggered by events." Online inference fetches features per request; streaming keeps them as keyed state and processes events in order. The failure modes — lag, ordering, late data — are the stream's, not the endpoint's.
- "We need streaming because the data is a stream." The data being a stream says nothing about the prediction. If the decision is made hourly, consume the stream into a table and run batch; the mode follows the decision (Choosing the Inference Mode).
- "Late events are rare, drop them." They are rare in aggregate and common for the shipments that matter — the delayed ones on congested lanes. The drop policy is a slice-level quality decision.
Where this applies
ML advice is stated as universal far more often than it is. These labels say what each claim is specific to — and where CONTESTED appears, the note gives the strongest form of the opposing view.
- GENERALThat the prediction inherits the stream's event-time and ordering semantics holds for any model placed in a stream processor; what varies is how much the model's features depend on windows and joins that make those semantics bite.
- SCALE-SPECIFICAt low event rates a consumer calling an online endpoint works and the streaming machinery is overhead; the mode earns its complexity when per-event feature fetches cannot keep up with the event rate, or when ordering within a key is what makes the prediction correct.
- FRAMEWORK-SPECIFICWatermark semantics, late-data handling and state checkpointing differ between stream processors; the design here is at the level of the concepts, and the exact guarantees — and which are defaults — must be checked against the framework in use.
Where the depth lives
This domain teaches the model and hands the rest off by name.