Real-Time Anomaly Detection Pipeline

Introduction

30 Second Summary

When thousands of fans rush to buy concert tickets at the same moment, server logs explode with activity. Buried in that flood of data are warning signs of real problems. Catching those signals in real-time is what separates a smooth on-sale from a full site outage.

In this project, you will build a real-time anomaly detection pipeline that ingests simulated server logs through Apache Kafka, detects anomalies using machine learning, and produces benchmark charts proving how the system performs under increasing load.

What You'll Build

A complete proof-of-concept data pipeline that simulates Tickster's real-world traffic patterns (normal browsing, on-sale spikes, error bursts) and detects anomalies in real-time using an Isolation Forest model running on a local Docker stack.

By the end of this project, you'll have:

• A Docker Compose stack running a single-node Kafka broker in KRaft mode.
• A Kafka producer that generates realistic server log events simulating Tickster traffic patterns.
• A stream consumer with sliding-window feature engineering and Isolation Forest anomaly detection that publishes alerts to a dedicated topic.
• A benchmarking harness that measures P50/P95/P99 latency, throughput, and resource usage, with matplotlib charts saved as PNG files.
• Secret Mission: Add a statistical Z-score detector and compare its detection accuracy and latency against Isolation Forest.

Are there any prerequisites?

You should be comfortable writing basic Python scripts. You also need Docker Desktop and VS Code installed on your machine. No prior experience with Kafka or machine learning is required.

Set Up the Project and Kafka Infrastructure

Every real-time pipeline starts with infrastructure. Before you can detect anomalies in streaming data, you need a message broker to move events from producers to consumers at high throughput.

In this step, you will create the project structure, install all Python dependencies, spin up an Apache Kafka broker in Docker, and verify end-to-end connectivity by round-tripping a test message through a Kafka topic.

In this step, get ready to:

• Create the project directory, virtual environment, and install all Python packages.
• Launch a single-node Kafka broker in KRaft mode using Docker Compose.
• Confirm Kafka connectivity by producing and consuming a test message from Python.

Create the project directory and Python virtual environment

Your project needs a clean directory structure with separate folders for source code and benchmark output.

• Check your Python version by running this command:

python3 --version

✔️ I see version 3.13 or higher

Your Python version is ready. Continue to the next instruction below.

ⓧ I see a version lower than 3.13

You need Python 3.13 or higher for this project. On macOS, the system Python is often outdated.

• Install the latest Python from python.org/downloads.
• After installation, run python3.13 --version to confirm the new version is available.
• Use python3.13 instead of python3 for all commands in this project if your system default remains older.

ⓧ Command not found

Python is not installed on your system.

• Download and install Python 3.13+ from python.org/downloads.
• On Windows, check the box that says Add python.exe to PATH during installation.
• After installation, close and reopen your terminal, then run python3 --version again.

• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder with subdirectories and move into it by running these commands:

mkdir -p tickster-anomaly-pipeline/src tickster-anomaly-pipeline/results
cd tickster-anomaly-pipeline

• Confirm the directories exist by running:

ls

You should see src and results listed in the output.

• Create your Python virtual environment by running:

python3 -m venv venv

• Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt.

• Create the requirements.txt file that lists all project dependencies. Open VS Code in the current folder by running:

code .

What if code . doesn't work?

If your terminal doesn't recognize the code command, open VS Code manually. Press Cmd+Space (macOS) or the Windows key (Windows), type VS Code, and press Enter. Then click File > Open Folder and select the tickster-anomaly-pipeline folder on your Desktop.

• In VS Code, create a new file called requirements.txt in the project root by clicking File > New File.
• Paste the following content into requirements.txt:

confluent-kafka==2.14.2
scikit-learn==1.9.0
numpy==2.2.6
matplotlib==3.10.3
psutil==7.0.0

What are these packages for?

Each package plays a specific role in your pipeline. confluent-kafka is the high-performance Python client for Kafka, backed by the C library librdkafka. scikit-learn provides the Isolation Forest model for anomaly detection.

The remaining packages handle numerics (numpy), chart generation (matplotlib), and system resource measurement (psutil).

• Save requirements.txt with Cmd+S (macOS) or Ctrl+S (Windows).
• Install all the packages by running this command in your terminal (make sure (venv) is still active):

pip install -r requirements.txt

The installation takes a minute or two. You should see lines ending with Successfully installed followed by all five package names.

Installation failing?

If you see errors about missing build tools or compilation failures, make sure you are running inside the (venv) virtual environment. Check that your terminal prompt starts with (venv). If it does not, re-run the activate command above.

On Windows, if confluent-kafka fails to install, make sure you are using Python 3.13 from python.org (not the Microsoft Store version). Help me fix my pip install errors.

Create the Docker Compose file for Kafka in KRaft mode

Apache Kafka is the message broker that sits at the center of your pipeline. It receives log events from producers and delivers them to consumers. You will run it locally inside a Docker container.

What is KRaft mode?

KRaft is Kafka's built-in consensus protocol that replaced the older ZooKeeper dependency. In KRaft mode, a single Kafka node handles both broker and controller roles. This makes local development much simpler because you only need one container instead of two.

• Make sure Docker Desktop is running. Press Cmd+Space (macOS) or the Windows key (Windows), type Docker Desktop, and press Enter to open it.
• Wait for the Docker engine status (bottom-left corner) to show a green indicator before continuing.

• Verify Docker is accessible from your terminal by running:

docker --version

✔️ I see a version number

Docker is ready. Continue to the next instruction.

ⓧ Command not found

Docker CLI is not on your PATH. Make sure Docker Desktop is fully started (green indicator in the bottom-left). If you just installed Docker Desktop, close and reopen your terminal.

• In VS Code, create a new file called docker-compose.yml in the project root.
• Paste the following configuration into docker-compose.yml:

services:
  broker:
    image: apache/kafka:latest
    container_name: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://localhost:9092,CONTROLLER://localhost:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_NUM_PARTITIONS: 3

What does this configuration do?

This spins up a single Kafka broker using the official apache/kafka Docker image in KRaft mode. The broker handles both message storage (broker role) and cluster coordination (controller role) in one process.

Port 9092 is mapped so your Python code can connect from the host machine. The replication factors are set to 1 because this is a single-node development setup. KAFKA_NUM_PARTITIONS: 3 means new topics are created with 3 partitions by default, enabling parallel consumption later.

• Save docker-compose.yml.
• Start the Kafka broker by running this command in your terminal:

docker compose up -d

Docker will pull the apache/kafka:latest image (about 228 MB) and start the container in the background. This may take a minute on the first run.

• Verify the container is running by checking:

docker ps

You should see a container named broker with status Up in the output.

Container not showing as Up?

If docker ps shows no running containers, check the logs with docker compose logs broker. The most common cause is port 9092 already being used by another process. Stop any other Kafka or service using that port, then run docker compose up -d again.

Help me get my Kafka Docker container running.

Verify Kafka connectivity with a test message

The final check is to confirm your Python code can talk to Kafka. You will create a test topic, send a message to it, and read that message back.

• Create a test topic inside the Kafka container by running:

docker exec --workdir /opt/kafka/bin/ broker ./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test-topic

You should see a message confirming the topic was created.

• In VS Code, create a new file called test_kafka.py in the project root (not inside src/).
• Paste the following test script into test_kafka.py:

from confluent_kafka import Producer, Consumer, KafkaError
import time

# Produce a test message
producer = Producer({"bootstrap.servers": "localhost:9092"})
producer.produce("test-topic", value="hello from python")
producer.flush()
print("Produced: 'hello from python'")

# Consume the test message
consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "test-group",
    "auto.offset.reset": "earliest",
})
consumer.subscribe(["test-topic"])

print("Waiting for message...")
start = time.time()
while time.time() - start < 10:
    msg = consumer.poll(timeout=1.0)
    if msg is None:
        continue
    if msg.error():
        print(f"Error: {msg.error()}")
    else:
        print(f"Consumed: '{msg.value().decode('utf-8')}'")
        break

consumer.close()
print("Kafka connectivity verified!")

What does this script do?

This script exercises the full produce-consume cycle. It creates a Producer that sends one message to the test-topic topic, then creates a Consumer that subscribes to the same topic and reads the message back.

If you see the consumed message printed, your entire stack is working: Docker is running Kafka, your Python environment has the confluent-kafka client installed correctly, and network connectivity between host and container is established.

• Save test_kafka.py and run it from your terminal:

python3 test_kafka.py

You should see output like this:

Produced: 'hello from python'
Waiting for message...
Consumed: 'hello from python'
Kafka connectivity verified!

Not seeing the consumed message?

If the script times out without consuming, the broker may still be initializing. Wait 10-15 seconds after docker compose up -d and try running the script again. Kafka needs a moment to finish its startup sequence.

If you see a connection refused error, confirm the broker container is running with docker ps. The container name should be broker and its status should show Up. Help me debug my Kafka connectivity issue.

✔️ Awesome, I've got everything!

Great. Double check you have saved both requirements.txt and docker-compose.yml.

ⓧ I'd like to double check the full code

Here is the full requirements.txt:

confluent-kafka==2.14.2
scikit-learn==1.9.0
numpy==2.2.6
matplotlib==3.10.3
psutil==7.0.0

And the full docker-compose.yml:

services:
  broker:
    image: apache/kafka:latest
    container_name: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://localhost:9092,CONTROLLER://localhost:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_NUM_PARTITIONS: 3

Your infrastructure is live. You have a Kafka broker accepting messages and a Python environment with all the tools you need. Next up, you will build the traffic simulator that generates realistic Tickster server log events.

Build the Tickster Traffic Simulator

Your Kafka broker is running and you confirmed a test message can round-trip through it.

Now you need something that generates realistic traffic. In this step, you will build a Kafka producer that simulates Tickster server logs. The producer generates normal browsing events and can inject three types of anomalies: traffic spikes, error bursts, and latency degradation.

In this step, get ready to:

• Create a log event data model with realistic traffic distributions and anomaly injection modes.
• Build a Kafka producer loop with configurable rate, duration, and CLI arguments.
• Run the simulator at 100 events/second and verify messages arrive in the topic.

Create the event data model and normal traffic generator

Before you can detect anomalies, you need a baseline of what normal traffic looks like. Tickster's server logs contain fields like the HTTP method, the endpoint hit, the status code returned, and the response time.

• Create a new file src/producer.py by adding this code:

import json
import time
import random
import argparse
from confluent_kafka import Producer

ENDPOINTS = [
    "/api/events", "/api/tickets/purchase", "/api/cart",
    "/api/user/profile", "/api/search", "/api/venues"
]
METHODS = ["GET", "GET", "GET", "POST", "GET", "GET"]
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0)",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0)",
]


def generate_normal_event():
    endpoint_idx = random.randint(0, len(ENDPOINTS) - 1)
    return {
        "timestamp": time.time(),
        "method": METHODS[endpoint_idx],
        "endpoint": ENDPOINTS[endpoint_idx],
        "status_code": random.choices([200, 201, 301, 404, 500], weights=[85, 5, 3, 5, 2])[0],
        "response_time_ms": max(1, random.gauss(120, 40)),
        "user_agent": random.choice(USER_AGENTS),
        "session_id": f"sess_{random.randint(1000, 9999)}",
    }

What does this code do?

ENDPOINTS and METHODS define the six API routes that Tickster's servers handle. Each endpoint is paired with its typical HTTP method (most are GET, ticket purchases are POST).

generate_normal_event() builds a single log entry with a realistic distribution. Status codes are weighted so 200 responses appear 85% of the time. Response times follow a Gaussian curve centered at 120ms with 40ms standard deviation.

• Save src/producer.py.
• Confirm the file exists by running:

ls src/producer.py

You should see src/producer.py in the output.

Don't see the file?

Make sure you are in the project root directory (the folder containing src/). If you navigated elsewhere, cd back to it.

help me find my project directory

• Add the three anomaly injection functions below generate_normal_event() in src/producer.py:

def inject_spike_anomaly():
    event = generate_normal_event()
    event["endpoint"] = "/api/tickets/purchase"
    event["method"] = "POST"
    event["response_time_ms"] = max(1, random.gauss(120, 40))
    return event


def inject_error_burst():
    event = generate_normal_event()
    event["status_code"] = random.choice([500, 502, 503, 504])
    event["response_time_ms"] = max(1, random.gauss(300, 100))
    return event


def inject_latency_degradation():
    event = generate_normal_event()
    event["response_time_ms"] = random.gauss(800, 200)
    return event

What do these anomaly modes simulate?

• Spike anomaly: Floods the /api/tickets/purchase endpoint with POST requests. This mimics what happens when a popular concert goes on sale and thousands of users hit "Buy" simultaneously.
• Error burst: Forces 5xx server errors with elevated response times. This simulates a backend service crashing under load.
• Latency degradation: Pushes response times to 800ms (mean) instead of the normal 120ms. This simulates a slow database query or network congestion.

• Save src/producer.py. You will not see a visible change yet. The producer loop in the next chunk uses these functions to generate events.

Build the producer loop and CLI interface

The data model defines what events look like. Now you need the engine that sends them into Kafka at a controlled rate. The producer loop will throttle itself to match a target events-per-second rate while stamping each event with a producer timestamp for latency tracking later.

• Add the producer logic and CLI entry point below the anomaly functions in src/producer.py:

def delivery_report(err, msg):
    if err is not None:
        print(f"Delivery failed: {err}")


def run_producer(rate=100, duration=60, anomaly_probability=0.0, anomaly_type="spike"):
    producer = Producer({"bootstrap.servers": "localhost:9092"})

    anomaly_generators = {
        "spike": inject_spike_anomaly,
        "error": inject_error_burst,
        "latency": inject_latency_degradation,
    }

    start_time = time.time()
    event_count = 0
    interval = 1.0 / rate

    print(f"Starting producer: {rate} events/sec, duration={duration}s, anomaly_prob={anomaly_probability}")

    while time.time() - start_time < duration:
        loop_start = time.time()

        if random.random() < anomaly_probability:
            event = anomaly_generators[anomaly_type]()
            event["_anomaly_injected"] = True
        else:
            event = generate_normal_event()
            event["_anomaly_injected"] = False

        event["_producer_timestamp"] = time.time()

        producer.produce(
            "server-logs",
            key=event["endpoint"],
            value=json.dumps(event),
            callback=delivery_report,
        )
        producer.poll(0)
        event_count += 1

        elapsed = time.time() - loop_start
        sleep_time = interval - elapsed
        if sleep_time > 0:
            time.sleep(sleep_time)

    producer.flush()
    print(f"Produced {event_count} events in {time.time() - start_time:.1f}s")

What does the producer loop do?

• interval = 1.0 / rate calculates the sleep time between events to achieve the target throughput (e.g. 10ms gap for 100 events/sec).
• _producer_timestamp stamps the exact send time on each event. Later in the pipeline, you will compare this against the detection timestamp to measure end-to-end latency.
• _anomaly_injected flags whether the event was injected as an anomaly. This is a ground-truth label you will use during benchmarking to calculate detection accuracy.
• producer.poll(0) triggers delivery callbacks without blocking. producer.flush() at the end ensures all buffered messages are sent before the script exits.

• Now add the CLI entry point at the bottom of src/producer.py:

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Tickster traffic simulator")
    parser.add_argument("--rate", type=int, default=100, help="Events per second")
    parser.add_argument("--duration", type=int, default=60, help="Duration in seconds")
    parser.add_argument("--anomaly-prob", type=float, default=0.05, help="Anomaly injection probability")
    parser.add_argument("--anomaly-type", choices=["spike", "error", "latency"], default="spike")
    args = parser.parse_args()
    run_producer(args.rate, args.duration, args.anomaly_prob, args.anomaly_type)

What does the CLI interface give you?

The argparse block lets you control the simulator from the command line without editing code. You can change the event rate, how long it runs, how often anomalies appear, and which type of anomaly to inject.

• Save src/producer.py. You will run the script in the next substep to verify everything works end-to-end.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/producer.py with all functions in place.

ⓧ I'd like to double check the full code

import json
import time
import random
import argparse
from confluent_kafka import Producer

ENDPOINTS = [
    "/api/events", "/api/tickets/purchase", "/api/cart",
    "/api/user/profile", "/api/search", "/api/venues"
]
METHODS = ["GET", "GET", "GET", "POST", "GET", "GET"]
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0)",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0)",
]


def generate_normal_event():
    endpoint_idx = random.randint(0, len(ENDPOINTS) - 1)
    return {
        "timestamp": time.time(),
        "method": METHODS[endpoint_idx],
        "endpoint": ENDPOINTS[endpoint_idx],
        "status_code": random.choices([200, 201, 301, 404, 500], weights=[85, 5, 3, 5, 2])[0],
        "response_time_ms": max(1, random.gauss(120, 40)),
        "user_agent": random.choice(USER_AGENTS),
        "session_id": f"sess_{random.randint(1000, 9999)}",
    }


def inject_spike_anomaly():
    event = generate_normal_event()
    event["endpoint"] = "/api/tickets/purchase"
    event["method"] = "POST"
    event["response_time_ms"] = max(1, random.gauss(120, 40))
    return event


def inject_error_burst():
    event = generate_normal_event()
    event["status_code"] = random.choice([500, 502, 503, 504])
    event["response_time_ms"] = max(1, random.gauss(300, 100))
    return event


def inject_latency_degradation():
    event = generate_normal_event()
    event["response_time_ms"] = random.gauss(800, 200)
    return event


def delivery_report(err, msg):
    if err is not None:
        print(f"Delivery failed: {err}")


def run_producer(rate=100, duration=60, anomaly_probability=0.0, anomaly_type="spike"):
    producer = Producer({"bootstrap.servers": "localhost:9092"})

    anomaly_generators = {
        "spike": inject_spike_anomaly,
        "error": inject_error_burst,
        "latency": inject_latency_degradation,
    }

    start_time = time.time()
    event_count = 0
    interval = 1.0 / rate

    print(f"Starting producer: {rate} events/sec, duration={duration}s, anomaly_prob={anomaly_probability}")

    while time.time() - start_time < duration:
        loop_start = time.time()

        if random.random() < anomaly_probability:
            event = anomaly_generators[anomaly_type]()
            event["_anomaly_injected"] = True
        else:
            event = generate_normal_event()
            event["_anomaly_injected"] = False

        event["_producer_timestamp"] = time.time()

        producer.produce(
            "server-logs",
            key=event["endpoint"],
            value=json.dumps(event),
            callback=delivery_report,
        )
        producer.poll(0)
        event_count += 1

        elapsed = time.time() - loop_start
        sleep_time = interval - elapsed
        if sleep_time > 0:
            time.sleep(sleep_time)

    producer.flush()
    print(f"Produced {event_count} events in {time.time() - start_time:.1f}s")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Tickster traffic simulator")
    parser.add_argument("--rate", type=int, default=100, help="Events per second")
    parser.add_argument("--duration", type=int, default=60, help="Duration in seconds")
    parser.add_argument("--anomaly-prob", type=float, default=0.05, help="Anomaly injection probability")
    parser.add_argument("--anomaly-type", choices=["spike", "error", "latency"], default="spike")
    args = parser.parse_args()
    run_producer(args.rate, args.duration, args.anomaly_prob, args.anomaly_type)

Run the simulator and verify messages arrive

The producer is complete. Time to fire it up and confirm that events flow into Kafka's server-logs topic.

• Start the producer for a short 10-second burst by running:

python src/producer.py --rate 100 --duration 10 --anomaly-prob 0.05 --anomaly-type spike

You should see output like:

Starting producer: 100 events/sec, duration=10s, anomaly_prob=0.05
Produced 1000 events in 10.0s

Why 10 seconds for the test run?

A short burst is enough to confirm the producer works without flooding your terminal. At 100 events/sec for 10 seconds, you get 1000 events. That is plenty to verify the topic is receiving data.

• Now verify messages actually landed in the server-logs topic by running a quick consumer check:

python -c "
from confluent_kafka import Consumer
import json

c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'verify', 'auto.offset.reset': 'earliest'})
c.subscribe(['server-logs'])
msg = c.poll(timeout=5.0)
if msg and not msg.error():
    event = json.loads(msg.value().decode('utf-8'))
    print('Topic has messages. Sample event:')
    print(json.dumps(event, indent=2))
else:
    print('No messages found')
c.close()
"

You should see a sample JSON event printed with all the expected fields: timestamp, method, endpoint, status_code, response_time_ms, user_agent, session_id, _anomaly_injected, and _producer_timestamp.

Seeing 'No messages found'?

• Make sure the Kafka broker is still running. Run docker compose ps and confirm the broker container shows a healthy status.
• Check that the producer did not print any "Delivery failed" errors. If it did, the broker may have restarted. Run docker compose up -d to restart it, wait a few seconds, then re-run the producer.
• If you see a ModuleNotFoundError, make sure your virtual environment is activated before running the script.

help me debug why my producer messages aren't showing up in Kafka

Your traffic simulator is firing realistic events into Kafka. Next up, you will build the consumer that reads these events, computes sliding-window features, and prepares them for anomaly detection.

Build the Stream Consumer with Feature Engineering

Your producer is live, pumping realistic Tickster traffic into Kafka. The next challenge is making sense of that raw event stream in real time.

In this step, you'll build a consumer that reads log events and transforms them into meaningful statistical features using a sliding window. These features become the input your anomaly detection model will classify in the next step.

In this step, get ready to:

• Build a sliding-window buffer that keeps the last 60 seconds of events.
• Extract statistical features from the window: request rate, response time, and error rate.
• Wire the consumer to evaluate features every 5 seconds and print them to the console.

Build the sliding-window buffer and Kafka consumer

A sliding window keeps only the most recent events in memory. As new events arrive, old ones are evicted once they exceed the time threshold. This bounded buffer prevents memory from growing forever while giving you a consistent dataset to compute features from.

• Create a new file called consumer.py inside your src/ folder.
• Add the imports, the SlidingWindow class, and the create_consumer function by pasting the following code:

import json
import time
from collections import deque
from confluent_kafka import Consumer, KafkaError


class SlidingWindow:
    def __init__(self, window_seconds=60):
        self.window_seconds = window_seconds
        self.events = deque()

    def add(self, event):
        self.events.append(event)
        self._evict_old()

    def _evict_old(self):
        cutoff = time.time() - self.window_seconds
        while self.events and self.events[0]["timestamp"] < cutoff:
            self.events.popleft()


def create_consumer():
    return Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "anomaly-detector",
        "auto.offset.reset": "latest",
    })

What does this code do?

• SlidingWindow stores events in a deque (double-ended queue). This allows fast appends at the right and fast eviction at the left.
• _evict_old() removes any event older than 60 seconds from the left side of the deque every time a new event arrives.
• create_consumer() initializes a Kafka consumer with auto.offset.reset set to latest, so it only reads new messages from this point forward.

You won't see a change yet. The next chunk adds feature extraction to make this window useful.

Why a sliding window instead of batch processing?

Batch processing would wait until a fixed number of events accumulate, introducing unpredictable delay. A time-based sliding window gives you a consistent 60-second view of the system regardless of traffic volume.

At 100 events per second, the window holds roughly 6,000 events. At peak load (1,000 events/sec), it holds about 60,000. The deque keeps memory bounded either way.

Implement feature extraction from the sliding window

Raw events alone are not useful for anomaly detection. You need to compress the window into a fixed-size feature vector that captures the system's behaviour: how fast requests are arriving, how long they take, and how many are failing.

• In src/consumer.py, add the get_features method inside the SlidingWindow class, directly below the _evict_old method:

    def get_features(self):
        self._evict_old()
        if len(self.events) < 10:
            return None

        events = list(self.events)
        duration = events[-1]["timestamp"] - events[0]["timestamp"]
        if duration <= 0:
            return None

        request_rate = len(events) / duration
        response_times = [e["response_time_ms"] for e in events]
        mean_response_time = sum(response_times) / len(response_times)
        variance = sum((rt - mean_response_time) ** 2 for rt in response_times) / len(response_times)
        stddev_response_time = variance ** 0.5

        error_count = sum(1 for e in events if e["status_code"] >= 400)
        error_rate = error_count / len(events)

        return {
            "request_rate": request_rate,
            "mean_response_time": mean_response_time,
            "stddev_response_time": stddev_response_time,
            "error_rate": error_rate,
            "window_size": len(events),
        }

What does this code do?

This method compresses the entire sliding window into four numbers that summarise the system's current state:

• request_rate measures throughput (events per second). A sudden spike means an on-sale event just started.
• mean_response_time and stddev_response_time track latency. High mean or high variance both signal degradation.
• error_rate counts 4xx and 5xx status codes as a fraction of total requests. A burst of 500s is an error anomaly.
• The method returns None if fewer than 10 events exist, preventing noisy features during startup.

You won't see a change yet. The next chunk wires the consumer loop that calls this method on a timer.

Wire the consumer evaluation loop

The consumer needs an event loop that continuously polls Kafka, feeds events into the sliding window, and evaluates features on a fixed interval. Evaluating every 5 seconds keeps computational overhead low while still catching anomalies quickly.

• In src/consumer.py, add the run_consumer function and the __main__ block below the create_consumer function:

def run_consumer(evaluation_interval=5):
    consumer = create_consumer()
    consumer.subscribe(["server-logs"])

    window = SlidingWindow(window_seconds=60)
    last_evaluation = time.time()
    features_buffer = []

    print("Consumer started. Waiting for events...")

    try:
        while True:
            msg = consumer.poll(timeout=0.1)
            if msg is None:
                pass
            elif msg.error():
                if msg.error().code() != KafkaError._PARTITION_EOF:
                    print(f"Consumer error: {msg.error()}")
            else:
                event = json.loads(msg.value().decode("utf-8"))
                window.add(event)

            now = time.time()
            if now - last_evaluation >= evaluation_interval:
                features = window.get_features()
                if features is not None:
                    features_buffer.append(features)
                    print(f"Features: rate={features['request_rate']:.1f}/s, "
                          f"resp={features['mean_response_time']:.0f}ms, "
                          f"err={features['error_rate']:.2%}, "
                          f"stddev={features['stddev_response_time']:.0f}ms")
                last_evaluation = now

    except KeyboardInterrupt:
        print("Shutting down consumer...")
    finally:
        consumer.close()

    return features_buffer


if __name__ == "__main__":
    run_consumer()

What does this code do?

• consumer.poll(timeout=0.1) checks for new messages every 100ms. A short timeout keeps the loop responsive without burning CPU.
• Every 5 seconds (controlled by evaluation_interval), the loop calls get_features() and prints the result. This means the Isolation Forest model (next step) only runs 12 times per minute regardless of load.
• features_buffer stores all computed feature vectors in memory. The detector in the next step will use this buffer during its warm-up phase.
• The KeyboardInterrupt handler ensures the consumer cleanly disconnects from Kafka when you press Ctrl+C.

• Save src/consumer.py.
• Open a terminal and start the consumer by running:

python src/consumer.py

• You should see Consumer started. Waiting for events... printed to the terminal.
• Open a second terminal and start the producer by running:

python src/producer.py --rate 100 --duration 30

• Wait about 10 seconds, then check the consumer terminal. You should see feature vectors printed every 5 seconds, similar to:

Features: rate=99.8/s, resp=121ms, err=6.89%, stddev=41ms

• Stop both processes with Ctrl+C when you've confirmed the feature vectors are printing.

Not seeing any feature vectors?

• Make sure the Kafka broker is still running. Run docker compose ps and confirm the broker container is in a running state.
• Check that you started the producer AFTER the consumer. Since auto.offset.reset is set to latest, the consumer only reads messages produced after it subscribes.
• The first feature vector takes about 5 seconds to appear (the evaluation interval). If fewer than 10 events are in the window, get_features() returns None and nothing prints. Wait a few more seconds.

Why isn't my Kafka consumer receiving messages?

✔️ Awesome, I've got everything!

Great. Double check you've saved src/consumer.py and confirmed feature vectors are printing.

ⓧ I'd like to double check the full code

Here is the complete src/consumer.py file. Make sure your code matches this exactly:

import json
import time
from collections import deque
from confluent_kafka import Consumer, KafkaError


class SlidingWindow:
    def __init__(self, window_seconds=60):
        self.window_seconds = window_seconds
        self.events = deque()

    def add(self, event):
        self.events.append(event)
        self._evict_old()

    def _evict_old(self):
        cutoff = time.time() - self.window_seconds
        while self.events and self.events[0]["timestamp"] < cutoff:
            self.events.popleft()

    def get_features(self):
        self._evict_old()
        if len(self.events) < 10:
            return None

        events = list(self.events)
        duration = events[-1]["timestamp"] - events[0]["timestamp"]
        if duration <= 0:
            return None

        request_rate = len(events) / duration
        response_times = [e["response_time_ms"] for e in events]
        mean_response_time = sum(response_times) / len(response_times)
        variance = sum((rt - mean_response_time) ** 2 for rt in response_times) / len(response_times)
        stddev_response_time = variance ** 0.5

        error_count = sum(1 for e in events if e["status_code"] >= 400)
        error_rate = error_count / len(events)

        return {
            "request_rate": request_rate,
            "mean_response_time": mean_response_time,
            "stddev_response_time": stddev_response_time,
            "error_rate": error_rate,
            "window_size": len(events),
        }


def create_consumer():
    return Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "anomaly-detector",
        "auto.offset.reset": "latest",
    })


def run_consumer(evaluation_interval=5):
    consumer = create_consumer()
    consumer.subscribe(["server-logs"])

    window = SlidingWindow(window_seconds=60)
    last_evaluation = time.time()
    features_buffer = []

    print("Consumer started. Waiting for events...")

    try:
        while True:
            msg = consumer.poll(timeout=0.1)
            if msg is None:
                pass
            elif msg.error():
                if msg.error().code() != KafkaError._PARTITION_EOF:
                    print(f"Consumer error: {msg.error()}")
            else:
                event = json.loads(msg.value().decode("utf-8"))
                window.add(event)

            now = time.time()
            if now - last_evaluation >= evaluation_interval:
                features = window.get_features()
                if features is not None:
                    features_buffer.append(features)
                    print(f"Features: rate={features['request_rate']:.1f}/s, "
                          f"resp={features['mean_response_time']:.0f}ms, "
                          f"err={features['error_rate']:.2%}, "
                          f"stddev={features['stddev_response_time']:.0f}ms")
                last_evaluation = now

    except KeyboardInterrupt:
        print("Shutting down consumer...")
    finally:
        consumer.close()

    return features_buffer


if __name__ == "__main__":
    run_consumer()

Your consumer is now transforming raw log events into structured feature vectors in real time. Next up, you'll feed these features into an Isolation Forest model that decides whether the current traffic pattern is normal or anomalous.

Integrate Isolation Forest Anomaly Detection

Your consumer is now extracting meaningful features from the live event stream. Feature vectors alone don't tell you whether traffic is normal or anomalous.

In this step, you'll plug an Isolation Forest model into the pipeline. It will learn what "normal" looks like during a warm-up period, then flag anything that deviates as an anomaly and publish alerts to a dedicated Kafka topic.

In this step, get ready to:

• Build an Isolation Forest detector that trains on a warm-up period of normal traffic.
• Publish anomaly alerts with scores and features to a dedicated Kafka topic.
• Verify real-time anomaly detection with injected spikes, error bursts, and latency degradation.

Build the Isolation Forest detector class

Isolation Forest is an unsupervised algorithm that isolates anomalies by randomly partitioning the data. Outliers require fewer random splits to isolate, so the model flags them without needing labeled examples.

• Create a new file called detector.py inside your src/ folder.
• Add the imports and the AnomalyDetector class by copying this code:

import json
import time
from collections import deque

import numpy as np
from sklearn.ensemble import IsolationForest
from confluent_kafka import Consumer, Producer, KafkaError


class AnomalyDetector:
    def __init__(self, warmup_samples=12, contamination=0.05, n_estimators=100):
        self.warmup_samples = warmup_samples
        self.model = IsolationForest(
            contamination=contamination,
            n_estimators=n_estimators,
            random_state=42,
        )
        self.training_data = []
        self.is_trained = False

    def add_training_sample(self, feature_vector):
        self.training_data.append(feature_vector)
        if len(self.training_data) >= self.warmup_samples and not self.is_trained:
            self._train()

    def _train(self):
        X = np.array(self.training_data)
        self.model.fit(X)
        self.is_trained = True
        print(f"Model trained on {len(self.training_data)} samples.")

    def predict(self, feature_vector):
        if not self.is_trained:
            return None, None
        X = np.array([feature_vector])
        prediction = self.model.predict(X)[0]
        score = self.model.decision_function(X)[0]
        return prediction, score

What does this code do?

The AnomalyDetector class wraps scikit-learn's Isolation Forest with a warm-up workflow.

• contamination=0.05 tells the model to expect about 5% of data points to be anomalies. This sets the decision threshold.
• add_training_sample() collects feature vectors during warm-up. Once 12 samples arrive, it automatically triggers training.
• predict() returns two values: the prediction (+1 for normal, -1 for anomaly) and the anomaly score from decision_function() where lower scores indicate more abnormal behaviour.

You won't see a change yet. The next chunk adds the sliding window that feeds features to this detector.

Why train on a warm-up period?

Isolation Forest is unsupervised, so it doesn't need labeled "anomaly" examples. Instead, it learns the shape of normal traffic during warm-up. With 12 samples at 5-second intervals, that's the first 60 seconds of data. After training, anything that looks sufficiently different from that baseline gets flagged.

• Below the AnomalyDetector class, add the SlidingWindow class by copying this code:

class SlidingWindow:
    def __init__(self, window_seconds=60):
        self.window_seconds = window_seconds
        self.events = deque()

    def add(self, event):
        self.events.append(event)
        self._evict_old()

    def _evict_old(self):
        cutoff = time.time() - self.window_seconds
        while self.events and self.events[0]["timestamp"] < cutoff:
            self.events.popleft()

    def get_features(self):
        self._evict_old()
        if len(self.events) < 10:
            return None

        events = list(self.events)
        duration = events[-1]["timestamp"] - events[0]["timestamp"]
        if duration <= 0:
            return None

        request_rate = len(events) / duration
        response_times = [e["response_time_ms"] for e in events]
        mean_response_time = sum(response_times) / len(response_times)
        variance = sum((rt - mean_response_time) ** 2 for rt in response_times) / len(response_times)
        stddev_response_time = variance ** 0.5

        error_count = sum(1 for e in events if e["status_code"] >= 400)
        error_rate = error_count / len(events)

        return [request_rate, mean_response_time, stddev_response_time, error_rate]

Why does this return a list instead of a dict?

The SlidingWindow in consumer.py returned a dictionary for human-readable console output. This version returns a plain list [request_rate, mean_response_time, stddev_response_time, error_rate] because scikit-learn expects numeric arrays as input.

You won't see a change yet. The next chunk connects these classes into a running pipeline.

Wire up detection with alert publishing

The detection pipeline reads from the server-logs topic, feeds events through the sliding window, runs the Isolation Forest on each evaluation, and publishes alerts to a separate anomaly-alerts topic when anomalies are detected.

• Below the SlidingWindow class, add the run_detector function and entry point by copying this code:

def run_detector(evaluation_interval=5, warmup_seconds=60):
    consumer = Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "anomaly-detector-main",
        "auto.offset.reset": "latest",
    })
    consumer.subscribe(["server-logs"])

    alert_producer = Producer({"bootstrap.servers": "localhost:9092"})

    window = SlidingWindow(window_seconds=60)
    detector = AnomalyDetector(warmup_samples=int(warmup_seconds / evaluation_interval))
    last_evaluation = time.time()
    latencies = []

    print("Detector started. Warming up...")

    try:
        while True:
            msg = consumer.poll(timeout=0.1)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() != KafkaError._PARTITION_EOF:
                    print(f"Consumer error: {msg.error()}")
                continue

            event = json.loads(msg.value().decode("utf-8"))
            window.add(event)

            now = time.time()
            if now - last_evaluation >= evaluation_interval:
                features = window.get_features()
                if features is None:
                    last_evaluation = now
                    continue

                if not detector.is_trained:
                    detector.add_training_sample(features)
                    last_evaluation = now
                    continue

                detection_start = time.time()
                prediction, score = detector.predict(features)
                detection_time = time.time() - detection_start

                if prediction == -1:
                    alert = {
                        "timestamp": now,
                        "anomaly_score": float(score),
                        "features": {
                            "request_rate": features[0],
                            "mean_response_time": features[1],
                            "stddev_response_time": features[2],
                            "error_rate": features[3],
                        },
                        "detection_latency_ms": detection_time * 1000,
                    }
                    alert_producer.produce(
                        "anomaly-alerts",
                        value=json.dumps(alert),
                    )
                    alert_producer.poll(0)
                    print(f"ANOMALY DETECTED | score={score:.4f} | "
                          f"rate={features[0]:.1f}/s | resp={features[1]:.0f}ms | "
                          f"err={features[3]:.2%}")

                producer_ts = event.get("_producer_timestamp")
                if producer_ts:
                    e2e_latency = now - producer_ts
                    latencies.append(e2e_latency)

                last_evaluation = now

    except KeyboardInterrupt:
        print("\nShutting down detector...")
        if latencies:
            arr = np.array(latencies)
            print(f"Latency P50={np.percentile(arr, 50)*1000:.1f}ms "
                  f"P95={np.percentile(arr, 95)*1000:.1f}ms "
                  f"P99={np.percentile(arr, 99)*1000:.1f}ms")
    finally:
        consumer.close()
        alert_producer.flush()


if __name__ == "__main__":
    run_detector()

What does this function do?

The run_detector function orchestrates the full detection pipeline in three phases:

• Warm-up phase: the first 60 seconds of feature vectors are collected and used to train the Isolation Forest. The model learns what normal traffic looks like.
• Detection phase: after training, each new feature vector is passed to predict(). A result of -1 means anomaly. The alert (including the anomaly score, features, and detection latency) is published to the anomaly-alerts topic.
• Console dashboard: each detection prints ANOMALY DETECTED with the score, request rate, response time, and error rate so you can see detections in real-time.
• Latency tracking: each evaluation records the end-to-end latency (producer timestamp to detection output). When you stop the detector with Ctrl+C, it prints P50/P95/P99 percentiles.

• Save src/detector.py.

Seeing an import error?

• Make sure your virtual environment is activated. You installed scikit-learn and numpy in Step 1.
• Check that the import line reads from sklearn.ensemble import IsolationForest (not sklearn.tree or sklearn.forest).

Help me debug an import error in my detector.py file.

✔️ Awesome, I've got everything!

Great. Double check you've saved src/detector.py before moving on.

ⓧ I'd like to double check the full code

import json
import time
from collections import deque

import numpy as np
from sklearn.ensemble import IsolationForest
from confluent_kafka import Consumer, Producer, KafkaError


class AnomalyDetector:
    def __init__(self, warmup_samples=12, contamination=0.05, n_estimators=100):
        self.warmup_samples = warmup_samples
        self.model = IsolationForest(
            contamination=contamination,
            n_estimators=n_estimators,
            random_state=42,
        )
        self.training_data = []
        self.is_trained = False

    def add_training_sample(self, feature_vector):
        self.training_data.append(feature_vector)
        if len(self.training_data) >= self.warmup_samples and not self.is_trained:
            self._train()

    def _train(self):
        X = np.array(self.training_data)
        self.model.fit(X)
        self.is_trained = True
        print(f"Model trained on {len(self.training_data)} samples.")

    def predict(self, feature_vector):
        if not self.is_trained:
            return None, None
        X = np.array([feature_vector])
        prediction = self.model.predict(X)[0]
        score = self.model.decision_function(X)[0]
        return prediction, score


class SlidingWindow:
    def __init__(self, window_seconds=60):
        self.window_seconds = window_seconds
        self.events = deque()

    def add(self, event):
        self.events.append(event)
        self._evict_old()

    def _evict_old(self):
        cutoff = time.time() - self.window_seconds
        while self.events and self.events[0]["timestamp"] < cutoff:
            self.events.popleft()

    def get_features(self):
        self._evict_old()
        if len(self.events) < 10:
            return None

        events = list(self.events)
        duration = events[-1]["timestamp"] - events[0]["timestamp"]
        if duration <= 0:
            return None

        request_rate = len(events) / duration
        response_times = [e["response_time_ms"] for e in events]
        mean_response_time = sum(response_times) / len(response_times)
        variance = sum((rt - mean_response_time) ** 2 for rt in response_times) / len(response_times)
        stddev_response_time = variance ** 0.5

        error_count = sum(1 for e in events if e["status_code"] >= 400)
        error_rate = error_count / len(events)

        return [request_rate, mean_response_time, stddev_response_time, error_rate]


def run_detector(evaluation_interval=5, warmup_seconds=60):
    consumer = Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "anomaly-detector-main",
        "auto.offset.reset": "latest",
    })
    consumer.subscribe(["server-logs"])

    alert_producer = Producer({"bootstrap.servers": "localhost:9092"})

    window = SlidingWindow(window_seconds=60)
    detector = AnomalyDetector(warmup_samples=int(warmup_seconds / evaluation_interval))
    last_evaluation = time.time()
    latencies = []

    print("Detector started. Warming up...")

    try:
        while True:
            msg = consumer.poll(timeout=0.1)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() != KafkaError._PARTITION_EOF:
                    print(f"Consumer error: {msg.error()}")
                continue

            event = json.loads(msg.value().decode("utf-8"))
            window.add(event)

            now = time.time()
            if now - last_evaluation >= evaluation_interval:
                features = window.get_features()
                if features is None:
                    last_evaluation = now
                    continue

                if not detector.is_trained:
                    detector.add_training_sample(features)
                    last_evaluation = now
                    continue

                detection_start = time.time()
                prediction, score = detector.predict(features)
                detection_time = time.time() - detection_start

                if prediction == -1:
                    alert = {
                        "timestamp": now,
                        "anomaly_score": float(score),
                        "features": {
                            "request_rate": features[0],
                            "mean_response_time": features[1],
                            "stddev_response_time": features[2],
                            "error_rate": features[3],
                        },
                        "detection_latency_ms": detection_time * 1000,
                    }
                    alert_producer.produce(
                        "anomaly-alerts",
                        value=json.dumps(alert),
                    )
                    alert_producer.poll(0)
                    print(f"ANOMALY DETECTED | score={score:.4f} | "
                          f"rate={features[0]:.1f}/s | resp={features[1]:.0f}ms | "
                          f"err={features[3]:.2%}")

                producer_ts = event.get("_producer_timestamp")
                if producer_ts:
                    e2e_latency = now - producer_ts
                    latencies.append(e2e_latency)

                last_evaluation = now

    except KeyboardInterrupt:
        print("\nShutting down detector...")
        if latencies:
            arr = np.array(latencies)
            print(f"Latency P50={np.percentile(arr, 50)*1000:.1f}ms "
                  f"P95={np.percentile(arr, 95)*1000:.1f}ms "
                  f"P99={np.percentile(arr, 99)*1000:.1f}ms")
    finally:
        consumer.close()
        alert_producer.flush()


if __name__ == "__main__":
    run_detector()

Run the detector and verify anomaly detection

You need two terminal windows for this test: one running the producer with anomaly injection enabled, and one running the detector.

• Make sure your Kafka broker is still running. If you stopped it earlier, start it again by running:

docker compose up -d

• In your first terminal, start the producer with anomaly injection by running:

python src/producer.py --rate 100 --duration 120 --anomaly-prob 0.05 --anomaly-type error

• Open a second terminal in the same project directory.
• Activate your virtual environment, then start the detector by running:

python src/detector.py

• Wait approximately 60-70 seconds for the warm-up phase to complete. You should see Model trained on 12 samples. in the output.
• After training, watch for ANOMALY DETECTED lines appearing as the injected error bursts trigger the model.
• Press Ctrl+C to stop the detector. It prints P50/P95/P99 latency statistics before exiting.

What should I see?

After the warm-up period, your terminal should show lines like:

Detector started. Warming up...
Model trained on 12 samples.
ANOMALY DETECTED | score=-0.1234 | rate=102.3/s | resp=298ms | err=45.00%

The exact numbers will vary. The key signal is that ANOMALY DETECTED lines appear when the producer is injecting error bursts (high error rate and elevated response times).

Not seeing any anomaly detections?

• Make sure you waited at least 60 seconds for warm-up to finish. Look for the Model trained on 12 samples. message before expecting detections.
• Confirm the producer is running with --anomaly-prob 0.05. Without this flag, no anomalies are injected.
• Check that the detector's consumer group (anomaly-detector-main) is different from any other consumer you have running. Two consumers in the same group split partitions.

Help me troubleshoot why my Isolation Forest detector isn't flagging anomalies.

• Try running the producer with a different anomaly type to see different detection patterns by running:

python src/producer.py --rate 100 --duration 120 --anomaly-prob 0.05 --anomaly-type latency

Latency degradation anomalies show up as high mean response times in the detection output. Error bursts show high error rates. Both should trigger the detector.

Your pipeline now detects anomalies in real-time and publishes structured alerts. Next up, you'll build a benchmarking harness to measure how this pipeline performs as load increases.

Benchmark and Visualize Pipeline Performance

Your pipeline detects anomalies in real-time and publishes alerts to Kafka. But a thesis committee wants empirical proof that the system scales.

In this step, you'll build a benchmarking harness that stress-tests the pipeline from 100 to 1000 events per second. You'll measure end-to-end latency percentiles, throughput, and resource usage at each load level, then generate charts that visualize where the system saturates.

In this step, get ready to:

• Instrument end-to-end latency measurement with P50/P95/P99 percentile computation.
• Build a load test that ramps from 100 to 1000 events/second in increments.
• Generate matplotlib charts showing latency, throughput, and resource usage under load.

Create the benchmark file with resource measurement

Benchmarking a streaming pipeline requires measuring what happens at each stage. The key metric is end-to-end latency: the time from when the producer sends an event to when the detector processes it.

You also need to track CPU and memory consumption so you can identify the saturation point. The psutil library gives you per-process resource stats on any operating system.

• Create a new file called src/benchmark.py and add the imports and resource measurement function:

import json
import time
import threading
import subprocess
import sys
from collections import deque

import numpy as np
import matplotlib.pyplot as plt
import psutil
from confluent_kafka import Consumer, Producer, KafkaError
from sklearn.ensemble import IsolationForest


def measure_resources():
    process = psutil.Process()
    return {
        "cpu_percent": process.cpu_percent(interval=1),
        "memory_mb": process.memory_info().rss / (1024 * 1024),
    }

What does this code do?

• The imports pull in everything the benchmark needs: Kafka clients for producing/consuming, numpy for percentile math, matplotlib for charts, psutil for system resource monitoring, and IsolationForest to replicate the detection workload.
• measure_resources() captures a snapshot of CPU usage and memory for the current Python process. The interval=1 parameter on cpu_percent() blocks for one second to get a meaningful CPU reading.
• memory_info().rss returns the Resident Set Size (actual physical memory used), converted from bytes to megabytes.

• Save src/benchmark.py.

You won't see a change yet. The next chunk builds the main benchmarking loop that uses this function.

Build the load test that ramps events per second

The load test needs to answer a specific question: at what throughput does latency spike? To find out, you'll ramp the producer rate from 100 events/second up to 1000, spending 30 seconds at each level.

At each level, the benchmark spins up its own producer thread, consumes events, runs the Isolation Forest model, and records how long the full pipeline takes.

• In src/benchmark.py, add the run_benchmark() function below measure_resources():

def run_benchmark(load_levels=None, duration_per_level=30, evaluation_interval=5):
    if load_levels is None:
        load_levels = list(range(100, 1100, 100))

    results = {
        "load_levels": load_levels,
        "p50_latency_ms": [],
        "p95_latency_ms": [],
        "p99_latency_ms": [],
        "throughput_eps": [],
        "cpu_percent": [],
        "memory_mb": [],
    }

    for rate in load_levels:
        print(f"\n--- Load level: {rate} events/sec ---")
        latencies = []
        events_processed = 0
        start_time = time.time()

        consumer = Consumer({
            "bootstrap.servers": "localhost:9092",
            "group.id": f"benchmark-{rate}-{int(time.time())}",
            "auto.offset.reset": "latest",
        })
        consumer.subscribe(["server-logs"])

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

        window = deque()
        model = IsolationForest(contamination=0.05, n_estimators=100, random_state=42)
        trained = False
        training_data = []

        producer_thread_stop = threading.Event()

        def produce_events():
            interval = 1.0 / rate
            while not producer_thread_stop.is_set():
                event = {
                    "timestamp": time.time(),
                    "method": "GET",
                    "endpoint": "/api/events",
                    "status_code": 200,
                    "response_time_ms": max(1, __import__("random").gauss(120, 40)),
                    "user_agent": "benchmark",
                    "session_id": "bench_001",
                    "_producer_timestamp": time.time(),
                    "_anomaly_injected": False,
                }
                producer.produce("server-logs", value=json.dumps(event))
                producer.poll(0)
                elapsed_sleep = interval - 0.0001
                if elapsed_sleep > 0:
                    time.sleep(elapsed_sleep)

        thread = threading.Thread(target=produce_events, daemon=True)
        thread.start()

        time.sleep(2)
        last_eval = time.time()

        while time.time() - start_time < duration_per_level:
            msg = consumer.poll(timeout=0.1)
            if msg is None:
                continue
            if msg.error():
                continue

            event = json.loads(msg.value().decode("utf-8"))
            window.append(event)
            events_processed += 1

            cutoff = time.time() - 60
            while window and window[0]["timestamp"] < cutoff:
                window.popleft()

            now = time.time()
            if now - last_eval >= evaluation_interval and len(window) >= 10:
                events_list = list(window)
                duration = events_list[-1]["timestamp"] - events_list[0]["timestamp"]
                if duration > 0:
                    features = [
                        len(events_list) / duration,
                        sum(e["response_time_ms"] for e in events_list) / len(events_list),
                        0,
                        sum(1 for e in events_list if e["status_code"] >= 400) / len(events_list),
                    ]

                    if not trained and len(training_data) < 12:
                        training_data.append(features)
                        if len(training_data) >= 12:
                            model.fit(np.array(training_data))
                            trained = True
                    elif trained:
                        model.predict(np.array([features]))
                        producer_ts = event.get("_producer_timestamp")
                        if producer_ts:
                            latencies.append(now - producer_ts)

                last_eval = now

        producer_thread_stop.set()
        thread.join(timeout=5)
        producer.flush()
        consumer.close()

        resources = measure_resources()
        total_time = time.time() - start_time

        if latencies:
            arr = np.array(latencies) * 1000
            results["p50_latency_ms"].append(float(np.percentile(arr, 50)))
            results["p95_latency_ms"].append(float(np.percentile(arr, 95)))
            results["p99_latency_ms"].append(float(np.percentile(arr, 99)))
        else:
            results["p50_latency_ms"].append(0)
            results["p95_latency_ms"].append(0)
            results["p99_latency_ms"].append(0)

        results["throughput_eps"].append(events_processed / total_time)
        results["cpu_percent"].append(resources["cpu_percent"])
        results["memory_mb"].append(resources["memory_mb"])

        print(f"  Throughput: {events_processed / total_time:.0f} eps | "
              f"P95 latency: {results['p95_latency_ms'][-1]:.1f}ms | "
              f"CPU: {resources['cpu_percent']:.1f}% | "
              f"Mem: {resources['memory_mb']:.0f}MB")

    return results

What does this function do?

This is the core benchmarking loop. For each load level (100, 200, ... 1000 events/sec), it:

• Spins up a producer thread that generates synthetic events at the target rate. Each event carries a _producer_timestamp for end-to-end latency calculation.
• Consumes events, maintains a 60-second sliding window, and runs the Isolation Forest model every 5 seconds (same as your real detector).
• After the model is trained (12 warm-up samples), each prediction triggers a latency measurement: the difference between the current time and the event's _producer_timestamp.
• At the end of each level, uses numpy.percentile() to compute P50, P95, and P99 latency in milliseconds.
• Each consumer gets a unique group.id (including the rate and a timestamp) so Kafka does not rebalance between levels.

• Save src/benchmark.py.

You won't see output yet. The next chunk adds the charting function and the entry point that kicks everything off.

Generate benchmark charts with matplotlib

Raw numbers are hard to interpret. Charts make it obvious where latency spikes, where throughput plateaus, and where memory climbs.

• In src/benchmark.py, add the plot_results() function and the __main__ entry point below run_benchmark():

def plot_results(results, output_dir="results"):
    import os
    os.makedirs(output_dir, exist_ok=True)

    loads = results["load_levels"]

    fig, ax = plt.subplots(figsize=(10, 6))
    ax.plot(loads, results["p50_latency_ms"], "o-", label="P50")
    ax.plot(loads, results["p95_latency_ms"], "s-", label="P95")
    ax.plot(loads, results["p99_latency_ms"], "^-", label="P99")
    ax.set_xlabel("Load (events/sec)")
    ax.set_ylabel("Latency (ms)")
    ax.set_title("End-to-End Latency vs Load")
    ax.legend()
    ax.grid(True)
    plt.tight_layout()
    plt.savefig(f"{output_dir}/latency_vs_load.png", dpi=150)
    plt.close()

    fig, ax = plt.subplots(figsize=(10, 6))
    ax.plot(loads, results["throughput_eps"], "o-", color="green")
    ax.set_xlabel("Target Load (events/sec)")
    ax.set_ylabel("Actual Throughput (events/sec)")
    ax.set_title("Throughput vs Load")
    ax.grid(True)
    plt.tight_layout()
    plt.savefig(f"{output_dir}/throughput_vs_load.png", dpi=150)
    plt.close()

    fig, ax1 = plt.subplots(figsize=(10, 6))
    ax2 = ax1.twinx()
    ax1.plot(loads, results["cpu_percent"], "o-", color="red", label="CPU %")
    ax2.plot(loads, results["memory_mb"], "s-", color="blue", label="Memory (MB)")
    ax1.set_xlabel("Load (events/sec)")
    ax1.set_ylabel("CPU %", color="red")
    ax2.set_ylabel("Memory (MB)", color="blue")
    ax1.set_title("Resource Usage vs Load")
    ax1.legend(loc="upper left")
    ax2.legend(loc="upper right")
    ax1.grid(True)
    plt.tight_layout()
    plt.savefig(f"{output_dir}/resources_vs_load.png", dpi=150)
    plt.close()

    print(f"\nCharts saved to {output_dir}/")


if __name__ == "__main__":
    results = run_benchmark()
    plot_results(results)

What does this code do?

plot_results() generates three publication-ready charts from the benchmark data:

• latency_vs_load.png plots P50, P95, and P99 end-to-end latency against load level. This shows the latency distribution under stress.
• throughput_vs_load.png plots actual events processed per second versus the target rate. When these lines diverge, the pipeline is saturating.
• resources_vs_load.png uses a dual-axis plot showing CPU percentage and memory in MB. This reveals whether the bottleneck is compute-bound or memory-bound.

The __main__ block runs run_benchmark() followed by plot_results(), so a single command produces both the data and the charts.

• Save src/benchmark.py.
• Make sure your Kafka broker is still running (check with docker compose ps).
• Run the benchmark by executing this command:

python src/benchmark.py

How long does this take?

The benchmark runs 10 load levels at 30 seconds each, plus warm-up time. Expect it to take approximately 5-6 minutes total. You'll see progress printed for each level.

• Confirm the charts were generated by running:

ls results/

You should see three PNG files: latency_vs_load.png, throughput_vs_load.png, and resources_vs_load.png.

Benchmark failing or producing empty charts?

• If you see connection errors, confirm your Kafka broker is running with docker compose ps. Restart it with docker compose up -d if needed.
• If all latency values are 0, the model may not have enough warm-up time. Make sure no other consumers are active on the server-logs topic stealing messages. Stop any running producer.py or detector.py processes first.
• If you see an import error for matplotlib or psutil, make sure your virtual environment is activated and all packages from requirements.txt are installed.

Still stuck? Help me debug why my benchmark is failing or producing empty charts.

✔️ Awesome, I've got everything!

Great. Double check you've saved src/benchmark.py and have all three chart files in results/.

ⓧ I'd like to double check the full code

import json
import time
import threading
import subprocess
import sys
from collections import deque

import numpy as np
import matplotlib.pyplot as plt
import psutil
from confluent_kafka import Consumer, Producer, KafkaError
from sklearn.ensemble import IsolationForest


def measure_resources():
    process = psutil.Process()
    return {
        "cpu_percent": process.cpu_percent(interval=1),
        "memory_mb": process.memory_info().rss / (1024 * 1024),
    }


def run_benchmark(load_levels=None, duration_per_level=30, evaluation_interval=5):
    if load_levels is None:
        load_levels = list(range(100, 1100, 100))

    results = {
        "load_levels": load_levels,
        "p50_latency_ms": [],
        "p95_latency_ms": [],
        "p99_latency_ms": [],
        "throughput_eps": [],
        "cpu_percent": [],
        "memory_mb": [],
    }

    for rate in load_levels:
        print(f"\n--- Load level: {rate} events/sec ---")
        latencies = []
        events_processed = 0
        start_time = time.time()

        consumer = Consumer({
            "bootstrap.servers": "localhost:9092",
            "group.id": f"benchmark-{rate}-{int(time.time())}",
            "auto.offset.reset": "latest",
        })
        consumer.subscribe(["server-logs"])

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

        window = deque()
        model = IsolationForest(contamination=0.05, n_estimators=100, random_state=42)
        trained = False
        training_data = []

        producer_thread_stop = threading.Event()

        def produce_events():
            interval = 1.0 / rate
            while not producer_thread_stop.is_set():
                event = {
                    "timestamp": time.time(),
                    "method": "GET",
                    "endpoint": "/api/events",
                    "status_code": 200,
                    "response_time_ms": max(1, __import__("random").gauss(120, 40)),
                    "user_agent": "benchmark",
                    "session_id": "bench_001",
                    "_producer_timestamp": time.time(),
                    "_anomaly_injected": False,
                }
                producer.produce("server-logs", value=json.dumps(event))
                producer.poll(0)
                elapsed_sleep = interval - 0.0001
                if elapsed_sleep > 0:
                    time.sleep(elapsed_sleep)

        thread = threading.Thread(target=produce_events, daemon=True)
        thread.start()

        time.sleep(2)
        last_eval = time.time()

        while time.time() - start_time < duration_per_level:
            msg = consumer.poll(timeout=0.1)
            if msg is None:
                continue
            if msg.error():
                continue

            event = json.loads(msg.value().decode("utf-8"))
            window.append(event)
            events_processed += 1

            cutoff = time.time() - 60
            while window and window[0]["timestamp"] < cutoff:
                window.popleft()

            now = time.time()
            if now - last_eval >= evaluation_interval and len(window) >= 10:
                events_list = list(window)
                duration = events_list[-1]["timestamp"] - events_list[0]["timestamp"]
                if duration > 0:
                    features = [
                        len(events_list) / duration,
                        sum(e["response_time_ms"] for e in events_list) / len(events_list),
                        0,
                        sum(1 for e in events_list if e["status_code"] >= 400) / len(events_list),
                    ]

                    if not trained and len(training_data) < 12:
                        training_data.append(features)
                        if len(training_data) >= 12:
                            model.fit(np.array(training_data))
                            trained = True
                    elif trained:
                        model.predict(np.array([features]))
                        producer_ts = event.get("_producer_timestamp")
                        if producer_ts:
                            latencies.append(now - producer_ts)

                last_eval = now

        producer_thread_stop.set()
        thread.join(timeout=5)
        producer.flush()
        consumer.close()

        resources = measure_resources()
        total_time = time.time() - start_time

        if latencies:
            arr = np.array(latencies) * 1000
            results["p50_latency_ms"].append(float(np.percentile(arr, 50)))
            results["p95_latency_ms"].append(float(np.percentile(arr, 95)))
            results["p99_latency_ms"].append(float(np.percentile(arr, 99)))
        else:
            results["p50_latency_ms"].append(0)
            results["p95_latency_ms"].append(0)
            results["p99_latency_ms"].append(0)

        results["throughput_eps"].append(events_processed / total_time)
        results["cpu_percent"].append(resources["cpu_percent"])
        results["memory_mb"].append(resources["memory_mb"])

        print(f"  Throughput: {events_processed / total_time:.0f} eps | "
              f"P95 latency: {results['p95_latency_ms'][-1]:.1f}ms | "
              f"CPU: {resources['cpu_percent']:.1f}% | "
              f"Mem: {resources['memory_mb']:.0f}MB")

    return results


def plot_results(results, output_dir="results"):
    import os
    os.makedirs(output_dir, exist_ok=True)

    loads = results["load_levels"]

    fig, ax = plt.subplots(figsize=(10, 6))
    ax.plot(loads, results["p50_latency_ms"], "o-", label="P50")
    ax.plot(loads, results["p95_latency_ms"], "s-", label="P95")
    ax.plot(loads, results["p99_latency_ms"], "^-", label="P99")
    ax.set_xlabel("Load (events/sec)")
    ax.set_ylabel("Latency (ms)")
    ax.set_title("End-to-End Latency vs Load")
    ax.legend()
    ax.grid(True)
    plt.tight_layout()
    plt.savefig(f"{output_dir}/latency_vs_load.png", dpi=150)
    plt.close()

    fig, ax = plt.subplots(figsize=(10, 6))
    ax.plot(loads, results["throughput_eps"], "o-", color="green")
    ax.set_xlabel("Target Load (events/sec)")
    ax.set_ylabel("Actual Throughput (events/sec)")
    ax.set_title("Throughput vs Load")
    ax.grid(True)
    plt.tight_layout()
    plt.savefig(f"{output_dir}/throughput_vs_load.png", dpi=150)
    plt.close()

    fig, ax1 = plt.subplots(figsize=(10, 6))
    ax2 = ax1.twinx()
    ax1.plot(loads, results["cpu_percent"], "o-", color="red", label="CPU %")
    ax2.plot(loads, results["memory_mb"], "s-", color="blue", label="Memory (MB)")
    ax1.set_xlabel("Load (events/sec)")
    ax1.set_ylabel("CPU %", color="red")
    ax2.set_ylabel("Memory (MB)", color="blue")
    ax1.set_title("Resource Usage vs Load")
    ax1.legend(loc="upper left")
    ax2.legend(loc="upper right")
    ax1.grid(True)
    plt.tight_layout()
    plt.savefig(f"{output_dir}/resources_vs_load.png", dpi=150)
    plt.close()

    print(f"\nCharts saved to {output_dir}/")


if __name__ == "__main__":
    results = run_benchmark()
    plot_results(results)

You now have empirical evidence of your pipeline's performance characteristics. These charts are exactly what a thesis appendix needs: reproducible, quantitative results showing where the system works well and where it hits its limits.

Secret mission

Z-Score Detector Comparison

Your Isolation Forest detector works, but how do you know it's the best approach? In research, you need empirical evidence. Add a statistical Z-score detector and run both methods head-to-head on the same stream to compare detection accuracy and latency.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally on Docker, so there are no ongoing cloud costs. The Kafka broker container uses approximately 500MB-1GB of RAM while running.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• Docker container running a Kafka broker (broker container, no financial cost).
• Local Python virtual environment with installed packages (confluent-kafka, scikit-learn, numpy, matplotlib, psutil).
• Project files and benchmark results in your project directory on the Desktop.

Keep everything running

No action needed. Choose this if you're still actively experimenting with load levels, tuning the Isolation Forest parameters, or extending the pipeline.

• Your Kafka broker stays running in Docker and is ready for immediate use.
• Your Python virtual environment remains activated in your terminal session.
• Benchmark charts in results/ are available for your thesis documentation.

Pause - I'll come back to this later

Stop the Docker container to free up RAM without losing your Kafka configuration or topic data.

• Stop the Kafka broker by running this command from your project directory:

docker compose stop

• Deactivate your Python virtual environment by running:

deactivate

• When you're ready to resume, run docker compose up -d from the project directory and reactivate your virtual environment.

Delete - I don't want to use this again

Remove all project resources and start fresh. This removes the Docker containers, volumes, virtual environment, and all project files.

• Stop and remove the Kafka container and its volumes by running this from your project directory:

docker compose down -v

• Deactivate your virtual environment if it is still active:

deactivate

• Navigate to your Desktop and delete the project directory.

On macOS or Linux:

cd ~/Desktop
rm -rf tickster-anomaly-pipeline

On Windows:

cd ~\Desktop
rmdir /s /q tickster-anomaly-pipeline

• Confirm the directory is gone by running ls on your Desktop. You should no longer see tickster-anomaly-pipeline in the output.

That's a Wrap!

That's a Wrap!

Nice work. You just built a complete real-time anomaly detection pipeline that ingests simulated Tickster server logs through Apache Kafka, detects anomalies with machine learning, and produces empirical benchmark results proving how the system scales.

You've learned how to:

• Built a real-time streaming pipeline with Kafka producers, consumers, and topics running in KRaft mode on Docker.
• Implemented sliding-window feature engineering and Isolation Forest unsupervised anomaly detection on live data streams.
• Built an empirical benchmarking harness measuring P50/P95/P99 latency, throughput, and resource usage, then generated publication-ready matplotlib charts for thesis documentation.
• Secret Mission: Compared a Z-score statistical detector against ML-based Isolation Forest with empirical accuracy and latency metrics.

Ready to quiz yourself?