Kafka + AI: Real-time ML Pipelines

Build production real-time AI pipelines using Apache Kafka for event streaming, feature computation, and ML model inference.

Kafka Architecture for AI

A typical Kafka-based AI pipeline flows through these stages:

• Producers emit events (user actions, sensor data, transactions) to Kafka topics.
• Stream processors (Kafka Streams, Flink) compute real-time features from event streams.
• ML consumers read enriched events, run inference, and publish predictions to output topics.
• Downstream consumers act on predictions (trigger alerts, update UIs, store results).

Producing Events for ML

from confluent_kafka import Producer
import json

producer = Producer({
    "bootstrap.servers": "kafka:9092",
    "acks": "all",  # Ensure durability
    "enable.idempotence": True,  # Exactly-once semantics
})

def emit_user_event(user_id, action, item_id, metadata):
    event = {
        "event_id": str(uuid4()),
        "timestamp": datetime.utcnow().isoformat(),
        "user_id": user_id,
        "action": action,
        "item_id": item_id,
        "metadata": metadata,
    }

    producer.produce(
        topic="user-interactions",
        key=user_id.encode(),  # Partition by user for ordering
        value=json.dumps(event).encode(),
        callback=delivery_report,
    )
    producer.flush()

ML Consumer: Real-time Inference

from confluent_kafka import Consumer
import json

consumer = Consumer({
    "bootstrap.servers": "kafka:9092",
    "group.id": "fraud-detection-v2",
    "auto.offset.reset": "latest",
    "enable.auto.commit": False,  # Manual commit for reliability
})
consumer.subscribe(["transactions"])

# Load model once at startup
model = load_fraud_model("fraud-model-v2")
feature_store = FeatureStoreClient()

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

    event = json.loads(msg.value())

    # Fetch real-time features
    user_features = feature_store.get_online_features(
        entity_rows=[{"user_id": event["user_id"]}],
        features=["user:avg_transaction", "user:transaction_count_1h"]
    )

    # Run inference
    prediction = model.predict({
        **event,
        **user_features
    })

    # Publish prediction
    producer.produce(
        topic="fraud-predictions",
        key=event["transaction_id"].encode(),
        value=json.dumps({
            "transaction_id": event["transaction_id"],
            "fraud_score": float(prediction.score),
            "is_fraud": prediction.score > 0.85,
        }).encode()
    )

    consumer.commit(msg)  # Commit after successful processing

Kafka Streams for Feature Engineering

Kafka Streams enables real-time feature computation directly within Kafka:

// Java Kafka Streams for windowed feature computation
StreamsBuilder builder = new StreamsBuilder();

KStream<String, UserEvent> events = builder.stream("user-interactions");

// Compute features: clicks per user in last 5 minutes
KTable<Windowed<String>, Long> clickCounts = events
    .filter((key, event) -> event.getAction().equals("CLICK"))
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count(Materialized.as("click-counts-store"));

// Compute features: average session duration
KTable<String, Double> avgSessionDuration = events
    .groupByKey()
    .aggregate(
        SessionStats::new,
        (key, event, stats) -> stats.update(event),
        Materialized.as("session-stats-store")
    )
    .mapValues(SessionStats::getAvgDuration);

Kafka Connect for Data Integration