Online Learning Pipeline for Real-Time Predictions

Production-ready guide covering online learning pipeline for real-time predictions with implementation patterns, code examples, and anti-patterns for enterprise engineering teams.

An online learning pipeline for real-time predictions is a system that continuously ingests data, updates models on the fly, and serves predictions with low latency. It is essential when data distributions shift rapidly—such as in ad bidding, fraud detection, or dynamic pricing—where static models quickly become stale and batch retraining cannot keep pace.

Data Ingestion: Streaming Features from Event Sources

Real-time predictions depend on feature data arriving continuously. The pipeline must collect and preprocess events from sources like Kafka, AWS Kinesis, or Pub/Sub, then deliver them to the prediction service.

Each event is a JSON object with a consistent schema. The pipeline consumes events using a Kafka consumer group with enable.auto.commit=true and auto.commit.interval.ms=500. A typical event:

{
  "user_id": 12345,
  "timestamp": "2023-10-05T14:32:15.123Z",
  "action": "click",
  "item_id": 67890,
  "session_id": "sess-abc123"
}

The ingestion layer uses a kafka-consumer process with the following configuration:

kafka-consumer:
  bootstrap_servers: "kafka1:9092,kafka2:9092"
  group_id: "feature-ingest"
  topic: "user-events"
  auto_offset_reset: "latest"
  fetch_max_wait_ms: 100
  fetch_min_bytes: 1024
  enable_auto_commit: true
  auto_commit_interval_ms: 500

The consumer reads events in batches and forwards them to a feature computation service via gRPC. A failed message is logged with:

ERROR: kafka-consumer: failed to process message: key=12345, offset=45678, error=InvalidSessionId, details={"session_id":"sess-xyz789"}

The pipeline silently drops events when the gRPC service is unreachable, but logs a retries_exceeded warning after three consecutive failures.

Feature Engineering: Computation and Storage

Features must be computed and stored for low-latency access during prediction. The pipeline computes features on demand and caches them in a Redis cluster.

A feature is defined by a feature_spec file:

feature_spec:
  name: "user_recent_clicks"
  version: "v2"
  source: "user_events"
  aggregation: "count"
  window: "PT1H"
  filters:
    - action: "click"
    - user_id: "not_null"
  output: "redis://redis1:6379/0"
  ttl: 3600

This spec is used by a feature-processor service that runs as a systemd unit:

[Unit]
Description=Feature Processor
After=network.target

[Service]
ExecStart=/usr/bin/feature-processor --config /etc/feature-processor.yaml
Restart=on-failure
User=feature
Group=feature
Environment=REDIS_URL=redis://redis1:6379

[Install]
WantedBy=multi-user.target

The processor consumes events from Kafka, computes features using a sliding window, and writes results to Redis using HSET commands:

HSET user:12345:clicks:1h count 42 timestamp 1696520123

Features are updated every 30 seconds via a cron job:


*/30 * * * * feature /usr/bin/feature-refresh --batch-size 1000 --timeout 60s

When a feature computation fails, the pipeline logs:

WARN: feature-processor: window 1h for user 12345 missing 12345 events, total=1234, expected=1500

The pipeline silently drops late-arriving events (more than 5 minutes after timestamp) and logs them in a late-events topic.

Model Training: Incremental Updates on Streaming Data

The core of the online learning pipeline is continuous model training. Models are updated incrementally using incoming data, avoiding the need for full batch retraining.

Training is driven by a model-trainer service that listens to a training-data Kafka topic. The trainer runs in a Docker container with the following docker-compose.yml:

version: '3.8'
services:
  model-trainer:
    image: scikit-learn:1.3
    command: >
      python -m trainer
      --input-topic training-data
      --output-model model.pkl
      --learning-rate 0.01
      --batch-size 256
      --window-size PT10M
      --checkpoint-interval 300
    volumes:
      - ./models:/models
    environment:
      - MODEL_VERSION=v3
      - REDIS_URL=redis://redis1:6379
    depends_on:
      - redis

The training loop processes batches of 256 samples, computes gradients, and updates the model weights using stochastic gradient descent. The model is saved every 300 seconds with:

import joblib
joblib.dump(model, "/models/model-v3.pkl")

A model update is published to a model-updates topic with:

{
  "model_id": "user_click_model",
  "version": "v3",
  "timestamp": "2023-10-05T14:45:30Z",
  "metrics": {
    "accuracy": 0.87,
    "f1": 0.84,
    "loss": 0.21
  },
  "size_bytes": 1048576
}

When a training job fails, the service logs:

ERROR: model-trainer: failed to train model: error=DataCorruption, expected=256 samples, got=241, details={"batch_id": "b12345", "start_time": "2023-10-05T14:45:00Z"}

The pipeline uses a model-registry service that maintains a list of active models. Each model is registered via a REST API:

POST /models
{
  "id": "user_click_model",
  "version": "v3",
  "source": "s3://models/user_click_model/v3.pkl",
  "created_at": "2023-10-05T14:45:30Z",
  "status": "active"
}

Models are promoted to active status only after validation against a held-out dataset.

Prediction Serving: Low-Latency Inference

Predictions are served via a gRPC API with a prediction-service daemon. The service loads the current active model and responds to incoming requests with a predict endpoint.

A prediction request includes user and item features:

{
  "user_id": 12345,
  "item_id": 67890,
  "context": {
    "timestamp": "2023-10-05T14:33:10Z",
    "session_id": "sess-abc123"
  }
}

The response is a JSON object:

{
  "prediction": 0.73,
  "features": {
    "user_recent_clicks": 42,
    "item_popularity": 0.81,
    "user_session_duration": 245.3
  },
  "model_version": "v3",
  "timestamp": "2023-10-05T14:33:10.123Z"
}

The service runs as a systemd unit with:

[Unit]
Description=Prediction Service
After=network.target

[Service]
ExecStart=/usr/bin/prediction-service --model-dir /models --port 8080
Restart=on-failure
User=prediction
Group=prediction
Environment=REDIS_URL=redis://redis1:6379

[Install]
WantedBy=multi-user.target

When a feature is missing, the service logs:

WARN: prediction-service: missing feature user_recent_clicks for user 12345, using default=0

The pipeline supports model versioning: if a new model is available, the service begins serving predictions with it after a warm-up period of 30 seconds.

Model Lifecycle: Versioning, Validation, and Rollout

Model versions are managed through a versioned model registry. Each model has a model.yaml file:

model:
  id: user_click_model
  version: v3
  description: "Click probability model using real-time features"
  created_at: 2023-10-05T14:45:30Z
  input_schema:
    user_id: int
    item_id: int
    session_id: string
    timestamp: datetime
  output_schema:
    prediction: float
    features: object
  dependencies:
    - feature:user_recent_clicks:v2
    - feature:item_popularity:v1
  metrics:
    accuracy: 0.87
    f1: 0.84
    loss: 0.21

A model is validated by running a model-validator service that ingests a test dataset and computes metrics. The validator sends results to a model-validation topic:

{
  "model_id": "user_click_model",
  "version": "v3",
  "test_set": "test-2023-10-05",
  "metrics": {
    "accuracy": 0.86,
    "auc": 0.91,
    "precision": 0.83
  },
  "status": "passed"
}

Validation is triggered by a cron job:

0 3 * * * root /usr/bin/model-validator --test-set test-$(date +'%Y-%m-%d')

A model is promoted to active status only after passing validation and being served for at least 15 minutes.

Rollout is done via canary deployment: the new model is served alongside the current one, with traffic split by canary-split header:

GET /predict
Header: X-Canary-Split: 0.3

The pipeline monitors the split and logs:

INFO: model-rollout: canary v3 serving 30% of requests, 12345 predictions processed, avg_latency=18.4ms

Failure Modes and Silent Pitfalls

The pipeline fails silently in subtle ways:

The pipeline detects these via health checks:

GET /health
{
  "status": "ok",
  "services": {
    "ingest": { "healthy": true, "latency_ms": 12 },
    "features": { "healthy": true, "queue_length": 456 },
    "training": { "healthy": false, "last_update": "2023-10-05T14:45:30Z", "error": "MissingFeature" },
    "prediction": { "healthy": true, "requests_per_second": 245 }
  }
}

When a service is unhealthy, the pipeline logs:

CRITICAL: pipeline: feature-processor lagging, current_offset=123456, latest_offset=130000, lag=6544

Monitoring and Alerting

The pipeline uses Prometheus to scrape metrics and Grafana to visualize them. Key metrics include:

Alerts are defined in alerting.yaml:

alerting:
  rules:
    - name: "FeatureProcessorLag"
      expr: "pipeline_events_in{job='feature-processor'} < 100"
      for: "5m"
      severity: "warning"
      annotations:
        summary: "Feature processor lagging"
        description: "Events in rate below 100 per second for 5 minutes"
    - name: "ModelUpdateFailed"
      expr: "model_training_loss{status='failed'} > 0.3"
      for: "10m"
      severity: "critical"
      annotations:
        summary: "Model update failed"
        description: "Training loss exceeds threshold for 10 minutes"

The pipeline sends alerts via Slack and PagerDuty using a webhook endpoint.

Configuration and Deployment

The entire pipeline is configured in a pipeline.yaml file:

pipeline:
  ingestion:
    source: "kafka"
    topic: "user-events"
    consumer_group: "feature-ingest"
  features:
    cache: "redis"
    ttl: 3600
    refresh_interval: "30m"
  models:
    trainer:
      input_topic: "training-data"
      output_model: "/models/model.pkl"
      learning_rate: 0.01
      batch_size: 256
    registry:
      endpoint: "http://model-registry:8080"
      validation_interval: "1h"
  prediction:
    service: "gRPC"
    port: 8080
    model_version: "v3"
    canary_split: 0.3

This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.