Building a Real-Time Classification Pipeline: Streaming Inference, Feature Extraction, and Low-Latency Predictions in Production

You deployed a model. It scores 0.94 AUC on held-out data. Then you try to serve it in real time and discover that your P99 latency is 340ms, your feature pipeline has a 5-second lag, and when the model service goes down, there is no fallback. You are not serving ML. You are serving a liability.

Real-time classification at production scale is not a modeling problem. The model is usually the easy part. The hard part is everything around it: extracting features from live event streams without introducing stale data, storing those features for point-in-time retrieval, serving the model with consistent latency under load, handling the model service being unavailable, and knowing when the model has started degrading silently.

This article covers all of it, with TypeScript code for the classification API and feature pipeline integration.

---

The Architecture in One Paragraph

Events come in over Kafka or Kinesis. A feature extraction service consumes those events, computes features, and writes them to an online feature store (Redis or a purpose-built store like Feast or Hopsworks). When a prediction request arrives, the classification API fetches the latest features from the store, constructs the feature vector, calls the model serving layer (ONNX Runtime, TensorRT, or TorchServe), and returns a prediction. The model serving layer handles batching for GPU utilization. A/B traffic splitting routes some requests to a challenger model. Circuit breakers fall back to a cached or rule-based prediction when the model service is unavailable. Prediction logs feed a monitoring pipeline that detects data drift and prediction quality degradation.

Every piece of that paragraph is a design decision with real tradeoffs. Let’s go through them.

---

Feature Extraction from Streaming Events

The standard mistake: compute features directly in the prediction path. Someone fires a request, your API reaches back to the database, aggregates the last 30 days of user behavior, builds a feature vector, and calls the model. This works in a notebook. It does not work at 5,000 requests per second with a 50ms SLA.

The correct model: separate feature computation from prediction serving entirely. A Kafka consumer runs continuously, processes incoming events, and materializes features to a low-latency store. The prediction API reads pre-computed features. Computation cost is paid offline (or near-real-time) rather than on the critical path.

// feature-extractor.ts
import { Kafka } from "kafkajs";
import { Redis } from "ioredis";

interface UserEvent {
  userId: string;
  eventType: "click" | "purchase" | "view" | "search";
  itemId: string;
  timestamp: number;
  metadata: Record<string, unknown>;
}

interface UserFeatures {
  clickCount7d: number;
  purchaseCount30d: number;
  avgSessionDuration: number;
  lastActiveTs: number;
  categoryAffinities: Record<string, number>;
  updatedAt: number;
}

const kafka = new Kafka({ clientId: "feature-extractor", brokers: ["kafka:9092"] });
const consumer = kafka.consumer({ groupId: "feature-extractor-group" });
const redis = new Redis({ host: "redis", port: 6379 });

async function processEvent(event: UserEvent): Promise<void> {
  const key = `features:user:${event.userId}`;
  const existing = await redis.get(key);
  const features: UserFeatures = existing
    ? JSON.parse(existing)
    : {
        clickCount7d: 0,
        purchaseCount30d: 0,
        avgSessionDuration: 0,
        lastActiveTs: 0,
        categoryAffinities: {},
        updatedAt: 0,
      };

  // Increment rolling counters
  if (event.eventType === "click") {
    features.clickCount7d += 1;
  }
  if (event.eventType === "purchase") {
    features.purchaseCount30d += 1;
  }

  features.lastActiveTs = event.timestamp;
  features.updatedAt = Date.now();

  // TTL of 7 days keeps the store from growing unbounded
  await redis.set(key, JSON.stringify(features), "EX", 60 * 60 * 24 * 7);
}

async function runExtractor(): Promise<void> {
  await consumer.connect();
  await consumer.subscribe({ topic: "user-events", fromBeginning: false });

  await consumer.run({
    eachMessage: async ({ message }) => {
      if (!message.value) return;
      const event: UserEvent = JSON.parse(message.value.toString());
      await processEvent(event);
    },
  });
}

runExtractor().catch(console.error);

The important detail is the TTL. Without it, users who churned years ago consume memory forever. Set it to the longest window any feature needs.

---

Online Feature Store: Point-in-Time Lookups

The naive feature extractor above has a subtle problem: it computes running totals by reading and writing a single Redis key per event. Under high event volume, this creates contention. Concurrent writes produce incorrect counts.

The production-grade approach uses append-only event logs in the feature store and computes aggregates at read time for small windows, or uses pre-aggregated snapshots for larger windows refreshed by a background job.

More importantly: you need point-in-time correctness for training data. When you generate training labels and join features, you must join features as they existed at prediction time, not as they exist today. This is the training-serving skew problem. A purpose-built feature store (Feast, Hopsworks, Tecton) handles this. If you roll your own with Redis, you need to snapshot feature state at prediction time and store it alongside the prediction log. Without this, your offline evaluation metrics will not match production performance.

---

Model Serving Infrastructure

Three common choices:

Option	Best For	Latency	Operational Cost
ONNX Runtime	CPU inference, any framework export	2-15ms	Low
TensorRT	NVIDIA GPU, max throughput	1-5ms	High (GPU ops)
TorchServe	PyTorch native, flexible handlers	5-30ms	Medium

For most classification workloads, ONNX Runtime on CPU is the right default. Export your sklearn, XGBoost, or PyTorch model to ONNX once, serve it with a thin HTTP wrapper, and optimize later if needed. TensorRT is worth the complexity only when you have GPU hardware and throughput requirements that ONNX cannot meet.

The classification API calls the model server over gRPC (preferred over HTTP for latency) or HTTP2. Keep the model server as a separate process from the API layer. Mixing them means a model OOM crash takes down your API.

// classification-api.ts
import Fastify from "fastify";
import { Redis } from "ioredis";

interface ClassificationRequest {
  userId: string;
  contextFeatures?: Record<string, number>;
}

interface ClassificationResponse {
  prediction: string;
  confidence: number;
  modelVersion: string;
  featureAge: number; // milliseconds since features were last updated
  source: "model" | "fallback";
}

interface UserFeatures {
  clickCount7d: number;
  purchaseCount30d: number;
  avgSessionDuration: number;
  lastActiveTs: number;
  categoryAffinities: Record<string, number>;
  updatedAt: number;
}

const app = Fastify({ logger: true });
const redis = new Redis({ host: "redis", port: 6379 });

const MODEL_SERVER_URL = process.env.MODEL_SERVER_URL ?? "http://model-server:8080";
const MODEL_VERSION = process.env.MODEL_VERSION ?? "v1";
const FEATURE_STALENESS_THRESHOLD_MS = 60_000; // 1 minute

async function fetchFeatures(userId: string): Promise<UserFeatures | null> {
  const raw = await redis.get(`features:user:${userId}`);
  if (!raw) return null;
  return JSON.parse(raw) as UserFeatures;
}

function buildFeatureVector(
  features: UserFeatures,
  contextFeatures: Record<string, number> = {}
): number[] {
  return [
    features.clickCount7d,
    features.purchaseCount30d,
    features.avgSessionDuration,
    Math.min((Date.now() - features.lastActiveTs) / 1000, 86400), // recency in seconds, capped
    ...Object.values(contextFeatures),
  ];
}

async function callModelServer(featureVector: number[]): Promise<{ label: string; probability: number }> {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), 40); // 40ms budget for model call

  try {
    const res = await fetch(`${MODEL_SERVER_URL}/predict`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ inputs: [featureVector], version: MODEL_VERSION }),
      signal: controller.signal,
    });

    if (!res.ok) throw new Error(`Model server returned ${res.status}`);
    const data = await res.json() as { label: string; probability: number };
    return data;
  } finally {
    clearTimeout(timeout);
  }
}

function fallbackPrediction(): { label: string; probability: number } {
  // Rule-based fallback: return the most common class with low confidence
  return { label: "class_0", probability: 0.5 };
}

app.post<{ Body: ClassificationRequest }>("/classify", async (req, reply) => {
  const { userId, contextFeatures } = req.body;

  const features = await fetchFeatures(userId);

  if (!features) {
    // Cold start: no features exist for this user
    const fallback = fallbackPrediction();
    return reply.send({
      prediction: fallback.label,
      confidence: fallback.probability,
      modelVersion: MODEL_VERSION,
      featureAge: -1,
      source: "fallback",
    } satisfies ClassificationResponse);
  }

  const featureAge = Date.now() - features.updatedAt;
  const featureVector = buildFeatureVector(features, contextFeatures);

  let result: { label: string; probability: number };
  let source: "model" | "fallback" = "model";

  try {
    result = await callModelServer(featureVector);
  } catch (err) {
    // Circuit breaker / timeout: use rule-based fallback
    app.log.warn({ userId, err }, "model server call failed, using fallback");
    result = fallbackPrediction();
    source = "fallback";
  }

  return reply.send({
    prediction: result.label,
    confidence: result.probability,
    modelVersion: MODEL_VERSION,
    featureAge,
    source,
  } satisfies ClassificationResponse);
});

app.listen({ port: 3000, host: "0.0.0.0" });

The 40ms timeout on the model call is deliberate. If your total API budget is 100ms and feature fetch takes 10ms, you have 40ms for the model and 50ms for everything else. Size this timeout based on your actual SLA math, not a round number.

---

Batching Strategies for GPU Utilization

A GPU sitting idle between single-request calls is wasted hardware. The model server should batch incoming requests dynamically: collect requests for a short window (1-5ms), stack them into a single batch, and run one forward pass. The tradeoff is latency variance: a single request that arrives alone waits up to 5ms for the batch window to close.

The right tuning depends on your traffic pattern. At 500 RPS, a 2ms batch window collects roughly one request per millisecond, batches of 10 are common, and GPU utilization stays high. At 50 RPS, the same window mostly produces batches of 1, and the 2ms wait adds nothing.

Dynamic batching (what TorchServe and Triton Inference Server call it) adjusts batch size based on queue depth rather than a fixed time window. This handles traffic bursts well without adding unnecessary latency during quiet periods.

For CPU inference with ONNX Runtime, batching matters less because you do not have the GPU warm-up overhead. Focus on concurrency instead: run multiple ONNX sessions in parallel with a worker pool sized to your CPU count.

---

Latency Budgets and SLA Design

Before you write a line of serving code, define your latency budget as a constraint, not a goal. “We want it to be fast” is not a budget. “P99 < 100ms measured at the load balancer” is a budget.

Then break it down:

Step	P50 Budget	P99 Budget
Network (client to API)	5ms	20ms
Feature fetch (Redis)	2ms	8ms
Feature vector construction	<1ms	<1ms
Model server call	10ms	40ms
Response serialization	<1ms	<1ms
Network (API to client)	5ms	20ms
Total	~23ms	~90ms

If your P99 model call is 40ms but your Redis P99 is 15ms, fix Redis first. Chasing model optimization while your feature store has fat tails is a common mistake.

Instrument every step independently. Do not measure the request end-to-end and guess where the time went. Add timing spans per stage and export them to your APM tool.

---

Model Versioning and A/B Testing in Production

Never swap models in place. Always run the challenger model in parallel with the champion, split traffic, and measure both before promoting.

The clean pattern is a routing layer in front of your model server pool. The API sends the feature vector to the router. The router decides which model version to call based on a consistent hash of the user ID (so the same user always hits the same model version) and the current traffic split configuration.

// model-router.ts
interface ModelRoute {
  version: string;
  url: string;
  trafficFraction: number; // 0.0 to 1.0
}

const routes: ModelRoute[] = [
  { version: "v1", url: "http://model-v1:8080", trafficFraction: 0.9 },
  { version: "v2", url: "http://model-v2:8080", trafficFraction: 0.1 },
];

function selectRoute(userId: string): ModelRoute {
  // Deterministic hash so the same user always hits the same model
  let hash = 0;
  for (let i = 0; i < userId.length; i++) {
    hash = (hash * 31 + userId.charCodeAt(i)) >>> 0;
  }
  const bucket = (hash % 100) / 100;

  let cumulative = 0;
  for (const route of routes) {
    cumulative += route.trafficFraction;
    if (bucket < cumulative) return route;
  }
  return routes[0]; // fallback to champion
}

Log the model version alongside every prediction. When you compute metrics downstream, group by model version. Only promote v2 when its metrics are statistically better, not just numerically better. A 0.2% lift with a sample size of 500 is noise.

---

Fallback Strategies

Your model service will go down. Plan for it now, not during an incident.

Three fallback layers in order of preference:

1. Cached prediction: For users you have already scored recently, return the cached result with a staleness flag. Acceptable for classification tasks where the underlying state changes slowly (e.g., fraud risk, churn propensity).

2. Rule-based classifier: A hard-coded decision tree or threshold-based rule that covers the most common cases. This is not a performance compromise. It is an explicit degradation contract: “During model service outage, precision drops from 0.91 to 0.74. We accept this.”

3. Most-common-class prediction: Return the majority class with low confidence. The worst outcome. Use only when you have no other option and explicitly log it so you can measure the impact.

Implement these with a circuit breaker, not just a timeout. If the model server returns errors for 10 consecutive requests, open the circuit and stop calling it for 30 seconds. During that window, serve from fallback immediately rather than waiting for each individual request to time out.

---

Monitoring for Data Drift and Prediction Quality

A model that was accurate six months ago may be silently degrading today. Users change. Products change. The feature distributions shift. Your model does not adapt.

Two distinct problems:

Data drift: The distribution of input features has changed. Compute a rolling distribution of each feature and compare it to the training distribution using PSI (Population Stability Index) or KS test. Alert when PSI > 0.25 for any feature. This does not mean the model is wrong yet, but it means you should pay attention.

Prediction quality degradation: The model is making worse predictions. This requires labels. For some domains (fraud, click-through) you get labels within hours. For others (churn, lifetime value) you wait months. Log every prediction with a unique ID. When the label eventually arrives, join it back to the prediction and compute precision/recall/AUC over a rolling window. Compare to the offline benchmark. A 3-5% drop from baseline is worth investigating. A 10% drop is an incident.

Log at minimum:

• prediction ID
• user ID
• model version
• feature vector hash (not the full vector; this is a privacy and storage tradeoff)
• predicted class and probability
• feature staleness at prediction time
• whether the fallback was used

---

The Cold Start Problem

A new user has no features. You cannot serve a meaningful prediction. Your options:

1. Default prior: Return the population-level most common class. Works if you can tolerate low-quality predictions for new users.

2. Context-only features: Use only features derivable from the current request (device type, location, referrer, time of day) with no user history. Train a separate “cold start” model on these features.

3. Warm-up request: On first event ingestion, trigger a batch feature computation job that initializes the user’s feature vector from available signals (registration data, A/B variant assignment, etc.).

4. Hybrid: Route requests with no feature history to the cold start model. Route requests with fresh features to the main model. Gradually shift users from cold start to main model as feature history accumulates.

The worst approach is to use zero vectors for missing features and serve the main model anyway. The model was not trained on zero vectors. The predictions will be unreliable and you will not know it because there is no signal distinguishing cold-start predictions from confident ones.

---

Production Checklist

Before going live with a real-time classification pipeline:

• Feature freshness SLA defined and monitored (max acceptable staleness per feature group)
• Model server and classification API decoupled, circuit breaker in place
• Fallback path tested under load, not just in unit tests
• Latency budget documented and each step instrumented independently
• A/B test harness in place before the first model update ships
• Prediction log schema finalized and includes model version, feature staleness, fallback flag
• Cold start path explicitly handled, not silently falling through to the main model
• Data drift monitoring running from day one, not added after the first incident

Real-time ML inference is a distributed systems problem with a model in the middle. Get the systems right first. The model will improve over time. The infrastructure needs to be correct from the start.