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.

Data Engineering for ML: Pipelines, Labeling, and Data Quality

“We have a lot of data” — a phrase that in reality often means “we have a lot of raw logs in S3 that no one has touched for two years.” Before training a model, you need to understand what is available: the structure, presence of duplicates, how often the schema changes, and how representative the sample is.

Data Engineering for ML is not just ETL. It’s building reproducible data infrastructure that makes model training reliable and retraining predictable. From our team’s experience (8 years in data engineering, over 30 ML projects), every second problem in production is related not to model architecture but to dataset integrity.

How Are ETL Pipelines for ML Different from BI?

ETL for analytics and ETL for ML are different tasks. Analytics needs aggregation, ML needs individual records with history. Analytics doesn’t require train/val/test split, ML does. Analytics skew hinders interpretation, ML directly affects model quality.

Tools. Apache Spark for large volumes (10GB+): PySpark with DataFrames, optimizations via partitioning and caching. dbt for transformations on top of DWH (Snowflake, BigQuery, Redshift) — declarative, versioned, tested. Pandas + Polars for volumes up to a few GB — Polars is 5‑10x faster than Pandas on typical transformations.

Temporal splits. For ML it’s important that the split is by time, not random. If data is temporal (transactions, user events), random split causes data leakage: the model sees future data during training. Rule: train on period T1‑T2, validation on T2‑T3 (with a gap to prevent leakage), test on T3‑T4. An incorrect split can cost 10–15% of model quality on validation.

Incremental pipelines. The model is retrained weekly on new data. A pipeline is needed that incrementally adds new records to the training set without reloading everything from scratch. Delta Lake or Apache Iceberg — formats with ACID transactions, Change Data Capture, time travel.

What Causes Training‑Serving Skew and How to Avoid It?

Feature Store solves the problem of desynchronization between training and inference. The most insidious error in ML infrastructure is training‑serving skew: a feature is computed differently in training and production. The model learns on correct data, but inference gets different values.

Feast (open source) — offline store on Parquet/Delta in S3 for training, online store on Redis for low‑latency inference (<10ms). Feature definitions as Python code:

from feast import FeatureView, Field
from feast.types import Float32, Int64

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    schema=[
        Field(name="purchase_count_7d", dtype=Int64),
        Field(name="avg_session_duration", dtype=Float32),
    ],
    ttl=timedelta(days=7),
    source=user_features_source,
)

One definition, used everywhere. No discrepancies. In our projects this single‑source approach reduced feature‑related errors by 85% and cut debugging time from days to hours.

Streaming features. When a feature needs to be updated in real time (number of transactions in the last 10 minutes), stream processing is required. Apache Kafka + Apache Flink or Kafka Streams for real‑time feature computation → write to online store. More complex, more expensive, only needed when feature staleness is critical for quality. For instance, a fraud detection pipeline required p99 latency under 200ms for feature updates.

Data Labeling: How Not to Waste Budget

Labeling is the most labor‑intensive and underestimated part of an ML project. Poorly labeled data cannot be fixed by any architecture.

Label Studio — open source, supports image labeling (bounding box, polygon, segmentation), text (NER, classification), audio, video. Deploys in 10 minutes via Docker. For small teams — first choice.

Labeling quality assessment. Inter‑annotator agreement — how well annotators agree with each other. Cohen’s Kappa > 0.8 — good, 0.6‑0.8 — acceptable, < 0.6 — task ambiguous or instructions poor. Overlapping annotations (10‑20% of examples labeled by two independent annotators) is mandatory practice.

Active learning prevents budget waste. Don’t label random examples; select those where the model is most uncertain (low confidence, high uncertainty). Allows achieving the same quality with 50‑70% of the labeling volume. Modals, Prodigy, Label Studio support active learning workflows. In one NLP project, we reduced the labeling budget by 2.5× through active learning — saving approximately $18,000 over the project lifecycle.

Synthetic data. When real data is scarce or expensive to obtain. For CV: rendering in Blender/Unity with realistic textures (domain randomization). For NLP: paraphrase via LLM, backtranslation. Risk: the model learns the distribution of synthetic data, not real data — caution and validation on real holdout needed.

Data Quality: Validation and Monitoring

Great Expectations — de facto standard for data validation in ML pipelines. Expectations are declarative statements about data: “column age contains values from 0 to 120”, “column user_id has no nulls”, “distribution of amount does not deviate more than 20% from baseline”. Runs in the pipeline, on failure blocks progression. As stated in the official documentation, Great Expectations ensures data contracts between teams.

Pandera — Pythonic alternative for pandas/polars DataFrames. Schema‑based validation with type hints:

import pandera as pa

schema = pa.DataFrameSchema({
    "user_id": pa.Column(int, nullable=False),
    "score": pa.Column(float, pa.Check.between(0, 1)),
    "label": pa.Column(str, pa.Check.isin(["positive", "negative", "neutral"])),
})

Data freshness. The model expects data from the last N days. ETL fails, data is not updated — the model uses stale features. Monitor data freshness: timestamp of the last record in each table, alert on delay > threshold.

Deduplication. Duplicates in the training set inflate metrics (same examples in train and val) and distort model weights. MinHash LSH for approximate deduplication of large datasets. For exact — hash by normalized content.

Validation Tools Comparison

What Does a Data Engineering Project for ML Include?

We provide the full cycle:

• Audit of existing data and pipelines (1 week).
• Architecture design: selection of tools, formats, labeling methods.
• Implementation of ETL/ELT pipeline with validation and monitoring.
• Documentation of code and processes (model card, data card).
• Training your team on pipeline operation.
• Post‑deployment support for 3 months.
• Access to code repository and all pipeline definitions.

How We Build a Pipeline: Step by Step

1. Audit existing data. Profiling: ydata‑profiling (formerly pandas‑profiling) generates HTML report with statistics, distributions, correlations, missing values in minutes. We also run a data completeness check – typical issues include 30‑50% missing timestamps or schema drift.
2. Pipeline design. Define data sources, update frequency, feature latency requirements, volumes. Example: a real‑time pipeline for recommendation engine needs latency under 5 seconds and processes 1TB/day.
3. Implementation and testing. Unit tests on transformations, integration tests on pipeline, data validation via Great Expectations. We target 95% test coverage for transformation logic.
4. Deployment and monitoring. Alerts on freshness, quality checks, anomalies in data volumes. Typical alert threshold: no new data for 2 hours.

Storage and Formats

For most ML projects at start: Parquet in S3 + DVC for versioning. Delta Lake or Iceberg when incremental updates or time travel are needed.

Why Trust Us

We have been working in data engineering and ML for over 8 years. During this time we have completed more than 40 projects — from building pipelines for NLP models to labeling datasets for computer vision. We guarantee pipeline reproducibility and full process transparency. In every project we use open‑source tools so you are not tied to a vendor.

Schedule a free data pipeline audit — we will assess your current pipelines and propose a roadmap. Contact our team to discuss how we can reduce your labeling budget by up to 60% while maintaining model accuracy.