Streaming ML Inference Pipelines: A Guide to Kafka, Flink, and ONNX

We build streaming ML pipelines that process events with latency under 50 ms using Kafka, Flink, and ONNX Runtime. For instance, we developed an anti-fraud system for a fintech company handling 50,000 transactions per second with p99 latency of 45 ms. Another example is a pipeline for a payment aggregator with throughput of 20K events/s and p99 latency of 55 ms, which reduced chargebacks by 35%. Switching from batch to streaming allowed that client to cut infrastructure costs by 40% — with a project payback period of 3 months. One mid-sized client saved $150,000 annually after moving to streaming.

We have implemented over 50 such solutions. Founded in 2018, we bring 6+ years of expertise in real-time ML. Our streaming ML pipeline combines Kafka, Flink, and ONNX to deliver low latency. Apache Flink is one of the key tools, ensuring exactly-once semantics and fault tolerance. Our certified engineers guarantee pipeline stability and scalability.

Problems we solve

Latency. Classic batch pipelines introduce delays from minutes to hours. For real-time scoring of transactions or updating recommendations, this is unacceptable. We use sliding window aggregations and an online feature store (Redis) with lookups under 5 ms, so features are available instantly. Compared to batch processing, our approach reduces latency by up to 5x.

State management. Stream processing requires consistent state during failures and restarts. Apache Flink provides exactly-once semantics and checkpointing, which is critical for financial applications.

Model versioning. When updating a model, the pipeline cannot be stopped. We implement A/B testing via shadow traffic and gradual rollout of new versions using feature flags.

Architecture of a streaming ML pipeline

[Kafka / Kinesis / Pulsar] ↓ [Feature Computation] ← Flink / Spark Streaming / Kafka Streams (aggregations, windows, joins) ↓ [Feature Store Online] ← Redis / DynamoDB (< 5ms lookup) ↓ [Model Inference] ← Triton / TorchServe / ONNX Runtime (< 20ms) ↓ [Decision Engine] ← business rules + ML score ↓ [Action / Output Kafka] ← downstream systems 

Why Kafka Streams and ONNX?

Kafka Streams embeds into any microservice and doesn't require a separate processing cluster. For high‑load scenarios we use Flink with parallelism up to 16. ONNX Runtime runs models on CPU with latency < 5 ms and supports INT8 quantization for further acceleration. In fact, ONNX Runtime is 3x faster than running plain Python inference.

How to ensure exactly-once semantics?

Apache Flink provides exactly-once through checkpointing and consistent state snapshots. The Apache Flink documentation describes the mechanisms we apply in production. This guarantees that during failures no event is lost or processed twice.

Implementing a streaming pipeline

Real-time feature computation

from confluent_kafka import Consumer, Producer import json import redis import numpy as np import time from collections import deque, defaultdict import threading class StreamFeatureComputer: """Real-time feature computation""" def __init__(self, kafka_config: dict, redis_url: str): self.consumer = Consumer(kafka_config) self.producer = Producer({'bootstrap.servers': kafka_config['bootstrap.servers']}) self.redis = redis.from_url(redis_url) self.window_store = defaultdict(lambda: deque(maxlen=1000)) def compute_user_features(self, user_id: str, event: dict) -> dict: """Online features for a user""" key_prefix = f"user:{user_id}" now = event['timestamp'] # Sliding window aggregations via Redis pipe = self.redis.pipeline() # Transaction features event_key = f"{key_prefix}:events" pipe.lpush(event_key, json.dumps({ 'amount': event.get('amount', 0), 'ts': now, 'type': event.get('type', 'unknown') })) pipe.ltrim(event_key, 0, 999) # Keep last 1000 events pipe.expire(event_key, 86400) # TTL 24 hours pipe.execute() # Aggregations over different windows raw_events = self.redis.lrange(event_key, 0, -1) events = [json.loads(e) for e in raw_events] # Sort by timestamp events.sort(key=lambda x: x['ts'], reverse=True) window_1h = [e for e in events if now - e['ts'] <= 3600] window_24h = [e for e in events if now - e['ts'] <= 86400] amounts_1h = [e['amount'] for e in window_1h] amounts_24h = [e['amount'] for e in window_24h] features = { 'user_id': user_id, 'tx_count_1h': len(window_1h), 'tx_count_24h': len(window_24h), 'tx_amount_sum_1h': sum(amounts_1h), 'tx_amount_sum_24h': sum(amounts_24h), 'tx_amount_avg_1h': np.mean(amounts_1h) if amounts_1h else 0, 'tx_amount_max_1h': max(amounts_1h) if amounts_1h else 0, 'tx_amount_std_1h': np.std(amounts_1h) if len(amounts_1h) > 1 else 0, 'unique_merchants_1h': len(set(e.get('merchant_id') for e in window_1h)), 'time_since_last_tx': now - events[0]['ts'] if events else 9999, } return features def compute_velocity_features(self, entity_id: str, event_type: str, windows: list[int] = [60, 300, 3600]) -> dict: """Velocity checks: event frequency over different windows""" features = {} now = int(time.time()) for window in windows: key = f"velocity:{entity_id}:{event_type}:{window}" # Increment and expire pipe = self.redis.pipeline() pipe.incr(key) pipe.expire(key, window) count, _ = pipe.execute() features[f"count_{window}s"] = count return features 

Streaming inference with ONNX

import onnxruntime as ort import asyncio from aiohttp import ClientSession class StreamMLInference: """Low-latency inference in a stream""" def __init__(self, model_path: str, feature_store: redis.Redis): # ONNX for maximum speed opts = ort.SessionOptions() opts.inter_op_num_threads = 2 opts.intra_op_num_threads = 2 opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL self.session = ort.InferenceSession( model_path, sess_options=opts, providers=['CPUExecutionProvider'] ) self.feature_store = feature_store self.input_names = [inp.name for inp in self.session.get_inputs()] def predict(self, features: dict) -> dict: """Inference < 5ms for tabular model""" # Form input tensor feature_vector = np.array([[features.get(name, 0.0) for name in self.input_names]], dtype=np.float32) start = time.perf_counter() outputs = self.session.run(None, {self.input_names[0]: feature_vector}) latency_ms = (time.perf_counter() - start) * 1000 score = float(outputs[0][0][1]) # Probability of positive class return { 'score': score, 'decision': 'block' if score > 0.8 else 'review' if score > 0.5 else 'allow', 'latency_ms': latency_ms } def batch_predict(self, features_list: list[dict]) -> list[dict]: """Batch inference for microbatches""" if not features_list: return [] feature_matrix = np.array([[f.get(name, 0.0) for name in self.input_names] for f in features_list], dtype=np.float32) outputs = self.session.run(None, {self.input_names[0]: feature_matrix}) scores = outputs[0][:, 1].tolist() return [ {'score': s, 'decision': 'block' if s > 0.8 else 'review' if s > 0.5 else 'allow'} for s in scores ] 

Apache Flink pipeline (Python API)

from pyflink.datastream import StreamExecutionEnvironment from pyflink.datastream.connectors.kafka import KafkaSource, KafkaSink from pyflink.common import WatermarkStrategy, Types from pyflink.datastream.window import TumblingEventTimeWindows, SlidingEventTimeWindows from pyflink.common.time import Time def build_flink_ml_pipeline(): env = StreamExecutionEnvironment.get_execution_environment() env.set_parallelism(4) # Kafka source source = KafkaSource.builder() \ .set_bootstrap_servers("kafka:9092") \ .set_topics("transactions") \ .set_group_id("ml-pipeline") \ .set_value_only_deserializer(JsonRowDeserializationSchema()) \ .build() stream = env.from_source( source, WatermarkStrategy.for_monotonous_timestamps(), "Kafka Source" ) # Compute aggregations over a 5-minute sliding window windowed = stream \ .key_by(lambda event: event['user_id']) \ .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.seconds(30))) \ .aggregate(TransactionAggregator()) # Join with static features from database enriched = windowed.map(EnrichWithStaticFeatures()) # ML inference scored = enriched.map(MLScoringFunction()) # Sink: real-time actions sink = KafkaSink.builder() \ .set_bootstrap_servers("kafka:9092") \ .set_record_serializer(JsonRowSerializationSchema("ml-decisions")) \ .build() scored.sink_to(sink) env.execute("ML Streaming Pipeline") 

Monitoring and metrics

class StreamPipelineMonitor: """Metrics for real-time ML pipeline""" def __init__(self, prometheus_port: int = 8000): from prometheus_client import Counter, Histogram, Gauge, start_http_server self.events_processed = Counter('ml_events_total', 'Total events processed', ['decision']) self.inference_latency = Histogram('ml_inference_latency_ms', 'Inference latency in milliseconds', buckets=[1, 5, 10, 20, 50, 100, 500]) self.feature_lag = Gauge('feature_store_lag_ms', 'Time between event and feature availability') self.model_score_dist = Histogram('ml_model_score', 'Distribution of model scores', buckets=[0.1*i for i in range(11)]) start_http_server(prometheus_port) def record_inference(self, result: dict): self.events_processed.labels(decision=result['decision']).inc() self.inference_latency.observe(result.get('latency_ms', 0)) self.model_score_dist.observe(result['score']) 

Comparison of stream inference approaches

Online feature store comparison

Timeline and cost

A basic pipeline implementation (Kafka + Flink + Feature Store + ONNX) takes 3–4 weeks. Complex scenarios with custom aggregations and A/B testing take up to 6 weeks. We select the optimal stack for your use case. Contact us to get a project estimate within 2 days.

What's included

• Architecture diagram
• Pipeline code with comments
• Monitoring dashboards (Prometheus/Grafana)
• Documentation
• Team training
• One month post-launch support

Results and economics

Economic efficiency

Switching from a batch to a streaming pipeline reduces infrastructure costs by 30–40% by eliminating intermediate data storage. The average project payback period is 3 months. For example, one client reduced chargebacks by 35% and cut fraud detection time from 5 minutes to 50 ms after implementation. Another client saved $150,000 annually.

Step-by-step implementation plan

1. Analysis and audit (2–3 days): study current data, latency and throughput requirements, choose the tech stack.
2. Architecture design (3–5 days): design the pipeline scheme, define contracts.
3. Implementation (1–2 weeks): code the stream processing, feature engineering, integration with the ML model.
4. Testing (3–5 days): load testing, fault tolerance verification, optimization.
5. Deployment and monitoring (2–3 days): deploy to production, set up dashboards and alerts.

Schedule an audit of your current system and receive a commercial proposal. Contact us to discuss your case.