Real-time Machine Learning

Build ML systems that learn and predict in real-time using streaming inference, online learning, and real-time feature engineering.

Real-time ML Architecture

A real-time ML system combines three capabilities:

Real-time Feature Engineering

import faust

app = faust.App("feature-engine", broker="kafka://localhost:9092")

# Define event model
class UserEvent(faust.Record):
    user_id: str
    action: str
    timestamp: float
    item_id: str

# Windowed aggregation table
user_click_counts = app.Table(
    "user-click-counts",
    default=int
).tumbling(300.0)  # 5-minute windows

user_events_topic = app.topic("user-events", value_type=UserEvent)

@app.agent(user_events_topic)
async def compute_features(events):
    async for event in events:
        # Update windowed aggregation
        user_click_counts[event.user_id] += 1

        # Compute real-time features
        features = {
            "user_id": event.user_id,
            "clicks_5min": user_click_counts[event.user_id].current(),
            "time_of_day": hour_from_timestamp(event.timestamp),
            "session_depth": get_session_depth(event.user_id),
        }

        # Store in online feature store
        await feature_store.set_features(event.user_id, features)

Online Learning

Online learning updates models incrementally, one example at a time:

from river import linear_model, preprocessing, metrics

# Online learning pipeline
model = preprocessing.StandardScaler() | linear_model.LogisticRegression()
metric = metrics.Accuracy()

async def online_learn(event_stream):
    async for event in event_stream:
        features = extract_features(event)
        label = event.get("label")

        if label is not None:
            # Learn from labeled event
            prediction = model.predict_one(features)
            model.learn_one(features, label)
            metric.update(label, prediction)

            # Log metrics periodically
            if metric.n % 1000 == 0:
                log_metric("online_accuracy", metric.get())
        else:
            # Predict for unlabeled event
            score = model.predict_proba_one(features)
            await publish_prediction(event["id"], score)

Streaming vs Batch Inference

Latency Optimization

• Model warm-up: Pre-load models into GPU memory at service startup. Never load on first request.
• Feature pre-computation: Compute and cache features before they are needed. Use event-triggered pre-computation.
• Model distillation: Use smaller, faster models for real-time inference. Train them to mimic larger models.
• Prediction caching: Cache predictions for repeated inputs. Short TTL for dynamic features, longer for stable ones.
• Async post-processing: Return predictions immediately. Run logging, analytics, and feedback asynchronously.