Tutorial 13: Apache Kafka for Real-Time ML Pipelines

Introduction

Prerequisites

Understanding Apache Kafka

Setting Up Kafka Locally

Producers and Consumers with Python

Building a Streaming Feature Pipeline

Real-Time ML Inference with Kafka

Schema Registry for Data Contracts

Kafka Connect for Data Integration

Production Best Practices

Conclusion

Introduction

In modern machine learning systems, the ability to process data and generate predictions in real time is a critical differentiator. Batch inference pipelines that run on hourly or daily schedules simply cannot meet the requirements of fraud detection, recommendation engines, dynamic pricing, or anomaly detection systems that demand sub-second response times.

Apache Kafka is a distributed event streaming platform that enables you to build robust, scalable, real-time data pipelines. When integrated with ML models, Kafka allows you to ingest streaming data, compute features on the fly, serve predictions in real time, and feed results back into downstream systems — all with high throughput and fault tolerance.

This tutorial walks you through the complete journey: from Kafka fundamentals to building a production-grade real-time ML inference pipeline using Python.

Prerequisites

Python 3.9 or higher
Docker and Docker Compose installed
Basic understanding of machine learning concepts
Familiarity with REST APIs and JSON
Install required Python packages:

pip install confluent-kafka fastavro requests scikit-learn numpy pandas fastapi uvicorn

Understanding Apache Kafka

Core Concepts

Apache Kafka operates around several fundamental abstractions:

Topics are named feeds or categories to which records are published. Think of a topic as a database table or a folder in a filesystem. Each topic is split into partitions, which are ordered, immutable sequences of records. Partitions enable parallelism — multiple consumers can read from different partitions simultaneously. Producers are client applications that publish (write) events to Kafka topics. Consumers are applications that subscribe to (read and process) events from topics. Consumers are organized into consumer groups, where each partition is consumed by exactly one consumer within a group, enabling load balancing. Brokers are the Kafka servers that store data and serve clients. A Kafka cluster consists of multiple brokers for redundancy and scalability. Offsets are unique sequential IDs assigned to each record within a partition. Consumers track their position using offsets, enabling exactly-once or at-least-once processing guarantees.

Why Kafka for ML?

| Feature | Benefit for ML |

|---------|---------------|

| High throughput | Handle millions of prediction requests per second |

| Durability | Never lose input data even if ML service goes down |

| Decoupling | Separate data ingestion from model inference |

| Replayability | Reprocess historical data with updated models |

| Scalability | Add more consumers/partitions as load increases |

Setting Up Kafka Locally

Create a docker-compose.yml file for a complete Kafka stack:

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPERCLIENTPORT: 2181
      ZOOKEEPERTICKTIME: 2000
    ports:
"2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    dependson:

zookeeper
    ports:
"9092:9092"
    environment:
      KAFKABROKERID: 1

      KAFKAZOOKEEPERCONNECT: zookeeper:2181

      KAFKAADVERTISEDLISTENERS: PLAINTEXT://localhost:9092

      KAFKAOFFSETSTOPICREPLICATIONFACTOR: 1

      KAFKAAUTOCREATETOPICSENABLE: "true"


  schema-registry:
    image: confluentinc/cp-schema-registry:7.5.0
    dependson:
kafka
    ports:
"8081:8081"
    environment:
      SCHEMAREGISTRYHOSTNAME: schema-registry

      SCHEMAREGISTRYKAFKASTOREBOOTSTRAPSERVERS: kafka:9092


  kafka-connect:
    image: confluentinc/cp-kafka-connect:7.5.0
    dependson:
kafka
schema-registry
    ports:
"8083:8083"
    environment:
      CONNECTBOOTSTRAPSERVERS: kafka:9092
      CONNECTRESTPORT: 8083
      CONNECTGROUPID: "connect-cluster"
      CONNECTCONFIGSTORAGETOPIC: "connect-configs"

      CONNECTOFFSETSTORAGETOPIC: "connect-offsets"
      CONNECTSTATUSSTORAGETOPIC: "connect-status"

      CONNECTCONFIGSTORAGEREPLICATIONFACTOR: 1

      CONNECTOFFSETSTORAGEREPLICATIONFACTOR: 1

      CONNECTSTATUSSTORAGEREPLICATIONFACTOR: 1

      CONNECTKEYCONVERTER: "org.apache.kafka.connect.json.JsonConverter"

      CONNECTVALUECONVERTER: "org.apache.kafka.connect.json.JsonConverter"

Start the stack:

docker-compose up -d

Verify all services are running:

docker-compose ps

Producers and Consumers with Python

Building a Kafka Producer

The confluent-kafka library is the officially supported Python client for Kafka. Here is a producer that sends structured ML feature data:

import json
import time
import random
from confluentkafka import Producer

def deliverycallback(err, msg):

    """Called once for each message produced to indicate delivery result."""
    if err is not None:
        print(f"Message delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")

def createproducer():
    """Create and configure a Kafka producer."""
    config = {
        'bootstrap.servers': 'localhost:9092',
        'client.id': 'ml-feature-producer',
        'acks': 'all',                    # Wait for all replicas to acknowledge
        'retries': 3,                     # Retry on transient failures
        'retry.backoff.ms': 100,          # Wait between retries
        'linger.ms': 5,                   # Batch messages for efficiency
        'batch.size': 16384,              # Batch size in bytes
        'compression.type': 'snappy',     # Compress for throughput
    }
    return Producer(config)

def generatetransactionevent():
    """Simulate a financial transaction event."""
    return {
        'transactionid': f"txn{random.randint(100000, 999999)}",
        'userid': f"user{random.randint(1, 1000)}",
        'amount': round(random.uniform(1.0, 10000.0), 2),
        'merchantcategory': random.choice(['grocery', 'electronics', 'restaurant', 'travel', 'atm']),

        'timestamp': time.time(),
        'locationlat': round(random.uniform(-90, 90), 6),
        'locationlon': round(random.uniform(-180, 180), 6),

        'devicetype': random.choice(['mobile', 'web', 'pos']),
    }

def runproducer(topic='raw-transactions', numevents=100):
    """Produce simulated transaction events to Kafka."""
    producer = createproducer()


    for i in range(numevents):
        event = generatetransactionevent()
        key = event['userid']

        value = json.dumps(event)

        producer.produce(
            topic=topic,
            key=key,
            value=value,
            callback=deliverycallback
        )

        # Trigger delivery callbacks
        producer.poll(0)
        time.sleep(0.01)

    # Wait for all messages to be delivered
    producer.flush(timeout=30)
    print(f"Produced {numevents} events to topic '{topic}'")


if name == 'main':
    runproducer()

Building a Kafka Consumer

import json
from confluentkafka import Consumer, KafkaError, KafkaException


def createconsumer(groupid='ml-inference-group'):

    """Create and configure a Kafka consumer."""
    config = {
        'bootstrap.servers': 'localhost:9092',
        'group.id': groupid,
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False,       # Manual commit for control
        'max.poll.interval.ms': 300000,
        'session.timeout.ms': 10000,
        'fetch.min.bytes': 1,
        'fetch.max.wait.ms': 500,
    }
    return Consumer(config)

def processmessage(message):

    """Process a single Kafka message."""
    event = json.loads(message.value().decode('utf-8'))
    print(f"Processing transaction {event['transactionid']} "
          f"for user {event['userid']}, amount: ${event['amount']}")

    return event

def runconsumer(topics=['raw-transactions']):
    """Consume and process messages from Kafka."""
    consumer = createconsumer()

    consumer.subscribe(topics)

    try:
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError.PARTITIONEOF:

                    print(f"Reached end of partition {msg.partition()}")
                    continue
                else:
                    raise KafkaException(msg.error())

            event = processmessage(msg)
            consumer.commit(asynchronous=False)

    except KeyboardInterrupt:
        print("Consumer interrupted")
    finally:
        consumer.close()

if name == 'main':
    runconsumer()

Building a Streaming Feature Pipeline

A streaming feature pipeline transforms raw events into ML-ready features in real time. This is the critical bridge between raw data and model inference.

import json
import time
import numpy as np
from collections import defaultdict
from confluentkafka import Consumer, Producer

class StreamingFeatureEngine:
    """Real-time feature engineering from Kafka streams."""

    def init(self):
        self.userhistory = defaultdict(list)

        self.categorystats = defaultdict(lambda: {'count': 0, 'total': 0.0})

        self.consumer = Consumer({
            'bootstrap.servers': 'localhost:9092',
            'group.id': 'feature-engine',
            'auto.offset.reset': 'earliest',
            'enable.auto.commit': False,
        })

        self.producer = Producer({
            'bootstrap.servers': 'localhost:9092',
            'acks': 'all',
        })

    def computefeatures(self, event):

        """Transform raw event into ML features."""
        userid = event['userid']

        amount = event['amount']
        category = event['merchantcategory']
        timestamp = event['timestamp']

        # Update user history (keep last 100 transactions)
        self.userhistory[userid].append({
            'amount': amount,
            'timestamp': timestamp,
            'category': category,
        })
        if len(self.userhistory[userid]) > 100:
            self.userhistory[userid] = self.userhistory[userid][-100:]

        history = self.userhistory[userid]
        amounts = [h['amount'] for h in history]

        # Update category stats
        self.categorystats[category]['count'] += 1

        self.categorystats[category]['total'] += amount

        # Compute features
        features = {
            'transactionid': event['transactionid'],
            'userid': userid,
            'amount': amount,
            'amountzscore': self.computezscore(amount, amounts),

            'useravgamount': np.mean(amounts),

            'userstdamount': np.std(amounts) if len(amounts) > 1 else 0.0,

            'usermaxamount': max(amounts),

            'userminamount': min(amounts),

            'usertxncount': len(amounts),

            'timesincelasttxn': self.timesincelast(history, timestamp),

            'categoryavgamount': (

                self.categorystats[category]['total'] /
                self.categorystats[category]['count']

            ),
            'amounttocategoryratio': (
                amount / (self.categorystats[category]['total'] /

                self.categorystats[category]['count'])
                if self.categorystats[category]['count'] > 0 else 1.0

            ),
            'isnewuser': 1 if len(amounts) <= 2 else 0,

            'hourofday': int((timestamp % 86400) / 3600),

            'devicetypeencoded': {'mobile': 0, 'web': 1, 'pos': 2}.get(

                event.get('devicetype', 'web'), 1
            ),
        }
        return features

    def computezscore(self, value, history):
        if len(history) < 2:
            return 0.0
        mean = np.mean(history)
        std = np.std(history)
        return (value - mean) / std if std > 0 else 0.0

    def timesincelast(self, history, currenttime):
        if len(history) < 2:
            return 0.0
        return currenttime - history[-2]['timestamp']


    def run(self, inputtopic='raw-transactions', outputtopic='ml-features'):

        """Main loop: consume raw events, compute features, produce to output."""
        self.consumer.subscribe([inputtopic])
        print(f"Feature engine started: {inputtopic} -> {outputtopic}")

        try:
            while True:
                msg = self.consumer.poll(timeout=1.0)
                if msg is None or msg.error():
                    continue

                event = json.loads(msg.value().decode('utf-8'))
                features = self.computefeatures(event)


                self.producer.produce(
                    topic=outputtopic,
                    key=event['userid'],

                    value=json.dumps(features),
                )
                self.producer.poll(0)
                self.consumer.commit(asynchronous=False)

        except KeyboardInterrupt:
            pass
        finally:
            self.consumer.close()
            self.producer.flush()

if name == 'main':
    engine = StreamingFeatureEngine()
    engine.run()

Real-Time ML Inference with Kafka

Now we build the inference service that consumes feature vectors and produces predictions:

import json
import pickle
import numpy as np
from confluentkafka import Consumer, Producer

class RealTimeMLInference:
    """Consume features from Kafka, run ML inference, produce predictions."""

    FEATURECOLUMNS = [

        'amount', 'amountzscore', 'useravgamount', 'userstdamount',
        'usermaxamount', 'userminamount', 'usertxncount',
        'timesincelasttxn', 'categoryavgamount',

        'amounttocategoryratio', 'isnewuser', 'hourofday',
        'devicetypeencoded',
    ]

    def init(self, modelpath='fraudmodel.pkl'):
        with open(modelpath, 'rb') as f:

            self.model = pickle.load(f)
        print(f"Model loaded from {modelpath}")

        self.consumer = Consumer({
            'bootstrap.servers': 'localhost:9092',
            'group.id': 'ml-inference-service',
            'auto.offset.reset': 'latest',
            'enable.auto.commit': False,
        })

        self.producer = Producer({
            'bootstrap.servers': 'localhost:9092',
            'acks': 'all',
        })

    def predict(self, features):
        """Run model inference on a feature vector."""
        featurevector = np.array(

            [features.get(col, 0.0) for col in self.FEATURECOLUMNS]
        ).reshape(1, -1)

        prediction = self.model.predict(featurevector)[0]

        probability = self.model.predictproba(featurevector)[0]


        return {
            'transactionid': features['transactionid'],

            'userid': features['userid'],

            'isfraud': int(prediction),
            'fraudprobability': float(max(probability)),

            'modelversion': 'v1.0.0',
        }

    def run(self, inputtopic='ml-features', outputtopic='predictions'):
        """Main inference loop."""
        self.consumer.subscribe([inputtopic])

        print(f"Inference service started: {inputtopic} -> {outputtopic}")


        try:
            while True:
                msg = self.consumer.poll(timeout=1.0)
                if msg is None or msg.error():
                    continue

                features = json.loads(msg.value().decode('utf-8'))
                result = self.predict(features)

                self.producer.produce(
                    topic=outputtopic,
                    key=features['userid'],

                    value=json.dumps(result),
                )
                self.producer.poll(0)
                self.consumer.commit(asynchronous=False)

                if result['isfraud']:
                    print(f"FRAUD ALERT: {result['transactionid']} "

                          f"(prob: {result['fraudprobability']:.4f})")

        except KeyboardInterrupt:
            pass
        finally:
            self.consumer.close()
            self.producer.flush()

Schema Registry for Data Contracts

Schema Registry enforces a contract on the data flowing through Kafka topics, preventing schema mismatches from breaking your ML pipelines.

import json
from confluentkafka import Producer

from confluentkafka.serialization import (
    SerializationContext, MessageField
)
from confluentkafka.schemaregistry import SchemaRegistryClient
from confluentkafka.schemaregistry.avro import AvroSerializer

Define the Avro schema for transaction events
TRANSACTIONSCHEMA = json.dumps({

    "type": "record",
    "name": "Transaction",
    "namespace": "com.ml.fraud",
    "fields": [
        {"name": "transactionid", "type": "string"},
        {"name": "userid", "type": "string"},

        {"name": "amount", "type": "double"},
        {"name": "merchantcategory", "type": "string"},
        {"name": "timestamp", "type": "double"},
        {"name": "locationlat", "type": "double"},

        {"name": "locationlon", "type": "double"},
        {"name": "devicetype", "type": "string"},

    ]
})

def createavroproducer():

    """Create a producer with Avro serialization and Schema Registry."""
    schemaregistryclient = SchemaRegistryClient({

        'url': 'http://localhost:8081'
    })

    avroserializer = AvroSerializer(
        schemaregistryclient,
        TRANSACTIONSCHEMA,

        lambda obj, ctx: obj,  # Convert object to dict
    )

    producer = Producer({
        'bootstrap.servers': 'localhost:9092',
    })

    return producer, avroserializer

def producewithschema(producer, serializer, topic, event):
    """Produce a message with Avro serialization."""
    ctx = SerializationContext(topic, MessageField.VALUE)
    serializedvalue = serializer(event, ctx)


    producer.produce(
        topic=topic,
        key=event['userid'].encode('utf-8'),
        value=serializedvalue,

    )
    producer.poll(0)

Schema Evolution

Schema Registry supports schema evolution with compatibility modes:

BACKWARD: New schema can read data produced with the previous schema.

FORWARD: Previous schema can read data produced with the new schema.

FULL: Both backward and forward compatible.

# Set compatibility level for a subject curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \ --data '{"compatibility": "BACKWARD"}' \ http://localhost:8081/config/raw-transactions-value

Kafka Connect for Data Integration

Kafka Connect moves data between Kafka and external systems without writing code. This is useful for sinking predictions to a database or sourcing training data.

Sink Predictions to PostgreSQL

{ "name": "predictions-postgres-sink", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max": "2", "topics": "predictions","connection.url": "jdbc:postgresql://postgres:5432/mlresults", "connection.user": "mluser", "connection.password": "mlpassword", "auto.create": "true", "auto.evolve": "true", "insert.mode": "upsert", "pk.mode": "recordvalue", "pk.fields": "transactionid", "key.converter": "org.apache.kafka.connect.storage.StringConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter.schemas.enable": "false" } }

Deploy the connector:

curl -X POST -H "Content-Type: application/json" \ --data @predictions-sink.json \ http://localhost:8083/connectors

Source Training Data from a Database

{
  "name": "training-data-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:postgresql://postgres:5432/transactions",
    "connection.user": "mluser",
    "connection.password": "mlpassword",
    "mode": "timestamp+incrementing",
    "timestamp.column.name": "updatedat",

    "incrementing.column.name": "id",
    "topic.prefix": "db-",
    "poll.interval.ms": "5000"
  }
}

Production Best Practices

1. Partition Strategy

Choose partition keys that distribute load evenly while maintaining ordering guarantees where needed. For ML inference, partitioning by userid ensures all events for a user are processed in order.

# Produce with explicit partition key
producer.produce(
    topic='raw-transactions',
    key=event['userid'].encode('utf-8'),  # Partition by user

    value=json.dumps(event).encode('utf-8'),
)

2. Consumer Group Scaling

Scale consumers horizontally by adding more instances to the same consumer group. Each partition is assigned to exactly one consumer.

# Scale by running multiple instances with the same group.id If topic has 12 partitions, you can run up to 12 consumersconsumerconfig = { 'group.id': 'ml-inference-service', # Same group for all instances 'partition.assignment.strategy': 'cooperative-sticky', }

3. Error Handling and Dead Letter Queues

Never lose messages — route failed processing to a dead letter queue:

def processwithdlq(consumer, producer, msg):
    try:
        event = json.loads(msg.value().decode('utf-8'))
        result = processevent(event)

        producer.produce(topic='predictions', value=json.dumps(result))
    except Exception as e:
        errorrecord = {
            'originalmessage': msg.value().decode('utf-8'),

            'error': str(e),
            'topic': msg.topic(),
            'partition': msg.partition(),
            'offset': msg.offset(),
        }
        producer.produce(
            topic='ml-inference-dlq',
            value=json.dumps(errorrecord),
        )

4. Monitoring and Observability

Track key metrics for your ML Kafka pipeline:

Consumer lag: How far behind the consumer is from the latest message.
Throughput: Messages processed per second.
Latency: End-to-end time from event production to prediction output.
Error rate: Percentage of messages routed to the DLQ.

from prometheusclient import Counter, Histogram, starthttpserver

predictionscounter = Counter('mlpredictionstotal', 'Total predictions', ['modelversion'])

inferencelatency = Histogram('mlinferenceseconds', 'Inference latency')
errorscounter = Counter('mlerrorstotal', 'Processing errors')


starthttpserver(8000)  # Expose metrics on port 8000

5. Idempotent Processing

Ensure your inference service can safely reprocess the same message without side effects:

import redis

redisclient = redis.Redis(host='localhost', port=6379, db=0)

def processidempotent(event):

    txnid = event['transactionid']

    if redisclient.exists(f"processed:{txnid}"):

        return None  # Already processed, skip

    result = runinference(event)
    redisclient.setex(f"processed:{txnid}", 3600, "1")  # TTL 1 hour
    return result

Conclusion

Apache Kafka provides the backbone for real-time ML systems by decoupling data producers from consumers, ensuring durability, and enabling horizontal scalability. In this tutorial, you learned how to:

Set up a complete Kafka stack with Schema Registry and Kafka Connect
Build Python producers and consumers using the confluent-kafka library
Implement a streaming feature engineering pipeline
Deploy a real-time ML inference service
Enforce data contracts with Avro schemas
Apply production best practices including error handling, monitoring, and idempotent processing

The architecture presented here — raw events flowing through feature engineering into model inference with predictions sinking to downstream systems — is the standard pattern used by companies like Netflix, Uber, and LinkedIn for their real-time ML platforms. Start with this foundation and extend it to meet your specific requirements.

Apache Kafka for Real-Time ML Tutorial: Streaming Data Pipeline

Tutorial 13: Apache Kafka for Real-Time ML Pipelines

Table of Contents

Introduction

Prerequisites

Understanding Apache Kafka

Core Concepts

Why Kafka for ML?

Setting Up Kafka Locally

Producers and Consumers with Python

Building a Kafka Producer

Building a Kafka Consumer

Building a Streaming Feature Pipeline

Real-Time ML Inference with Kafka

Schema Registry for Data Contracts

Define the Avro schema for transaction events

Schema Evolution

Kafka Connect for Data Integration

Sink Predictions to PostgreSQL

Source Training Data from a Database

Production Best Practices

1. Partition Strategy

2. Consumer Group Scaling

If topic has 12 partitions, you can run up to 12 consumers

3. Error Handling and Dead Letter Queues

4. Monitoring and Observability

5. Idempotent Processing

Conclusion

Related Articles

Complete Prefect Tutorial: Modern Workflow Orchestration for ML

Complete Apache Airflow Tutorial: Workflow Orchestration for Data Pipelines

Complete Comet ML Tutorial: MLOps Platform for Experiment Tracking and Model Management

Flyte Tutorial: Workflow Orchestration for Machine Learning and Data Engineering

Related Articles

Complete Prefect Tutorial: Modern Workflow Orchestration for ML

Tutorial Lengkap Prefect: Modern Workflow Orchestration untuk ML Prefect adalah platform workflow orchestration modern y...

Complete Apache Airflow Tutorial: Workflow Orchestration for Data Pipelines

Tutorial Lengkap Apache Airflow: Workflow Orchestration untuk Data Pipelines Apache Airflow adalah platform open-source ...

Complete Comet ML Tutorial: MLOps Platform for Experiment Tracking and Model Management

Tutorial Lengkap Comet ML: Platform MLOps untuk Experiment Tracking dan Model Management Dalam dunia machine learning mo...

Flyte Tutorial: Workflow Orchestration for Machine Learning and Data Engineering

Tutorial Flyte: Workflow Orchestration untuk Machine Learning dan Data Engineering Flyte adalah platform workflow orches...