Real-Time ML Pipeline with Kafka

Stream sensor data through Apache Kafka and detect anomalies in real time.

Introduction

30 Second Summary

Every time you order a ride or stream a show, dozens of services are processing the same event in real time. One service calculates pricing, another detects fraud, and a third updates your dashboard. All of them read from the same stream of data without slowing each other down.

In this project, you will build a real-time anomaly detection pipeline using Apache Kafka. You'll stream simulated sensor data into a Kafka topic and have two independent consumer services process that stream simultaneously. One flags anomalous temperature readings using statistics, and the other computes running averages per sensor.

What You'll Build

You can demo a live pipeline in separate terminal windows, watching sensor events flow from a producer into Kafka and get picked up independently by an anomaly detector and a stats aggregator running side by side.

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

  • A sensor data producer that continuously generates and streams JSON events to Kafka, with occasional anomaly spikes you can watch get flagged in real time.
  • An anomaly detection consumer that applies z-score based statistics to flag outlier readings the moment they arrive.
  • A statistics aggregation consumer that computes running averages and counts per sensor, proving that multiple services can independently process the same data stream without interfering with each other.
  • Secret Mission: Add a dead letter topic that captures anomalous events and routes them to a separate stream for later investigation.

Are there any prerequisites?

You should be comfortable writing Python scripts and using pip to install packages. You also need Docker Desktop installed and running on your machine. No prior Kafka experience is required.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters. This is your moment to commit to the project goal before any hands-on work begins.

Set Up Kafka with Docker Compose

Before you can stream events between services, you need a running message broker. Apache Kafka is the industry standard for event streaming, but it normally requires Java 17+ installed on your machine.

Docker Compose lets you skip the Java dependency entirely. You'll define a single-node Kafka broker in a YAML file and start it with one command.

In this step, get ready to:
  • Create a docker-compose.yml that configures a Kafka broker in KRaft mode.
  • Start the broker container and confirm it accepts connections.
  • Set up a Python virtual environment with the confluent-kafka client library.
Create the Docker Compose configuration

Kafka 4.x uses KRaft mode, which means the broker manages its own metadata without needing a separate ZooKeeper service. This simplifies your setup to a single container.

  • In your code editor, create a new file called docker-compose.yml in your project folder.
  • Paste the following configuration into the file:
services:
  broker:
    image: apache/kafka:4.3.1
    hostname: broker
    container_name: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker: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?

KAFKA_PROCESS_ROLES: broker,controller tells this single node to act as both the message broker and the cluster controller. This is KRaft mode in action.

KAFKA_LISTENERS binds inside the container on all interfaces (0.0.0.0), while KAFKA_ADVERTISED_LISTENERS advertises localhost:9092 so your Python scripts on the host machine can connect.

KAFKA_NUM_PARTITIONS: 3 sets the default partition count for new topics, which you'll use later to demonstrate consumer group behaviour.

  • Save docker-compose.yml.
  • Confirm the file exists by running:
ls docker-compose.yml

You should see docker-compose.yml listed in the output.

Don't see the file?

Make sure your terminal is in the same folder where you saved the file. Run pwd to check your current directory, then cd into your project folder if needed.

help me find my docker-compose.yml file

Start the Kafka broker
  • Start the Kafka container in detached mode by running:
docker compose up -d

Docker pulls the apache/kafka:4.3.1 image (this may take a minute the first time) and starts the broker in the background.

Before you check the status, what do you expect to see when you run docker compose ps?

  • Verify the broker is running by running:
docker compose ps

You should see the broker container listed with a status of running.

  • Confirm the broker is accepting connections by listing topics inside the container:
docker exec broker /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

You should see either empty output or a blank line. This means the broker is alive and responding to requests, with no topics created yet.

Seeing a connection error?

The broker takes a few seconds to fully start. Wait 5-10 seconds after docker compose up -d and try the topic list command again.

If the container isn't listed in docker compose ps, check that Docker Desktop is running and try docker compose up -d again.

help me troubleshoot my Kafka Docker container not starting

Set up your Python environment

You'll write Python scripts that connect to this broker. A virtual environment keeps your project dependencies isolated from the rest of your system.

  • Create a virtual environment by running:

macOS

python3 -m venv venv

Windows

python -m venv venv
  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

  • Install the Kafka client library by running:
pip install confluent-kafka==2.15.0

Why confluent-kafka?

The confluent-kafka package is maintained by Confluent (the company behind Kafka) and wraps the high-performance librdkafka C library. It is the recommended Python client for production Kafka workloads.

You should see output ending with Successfully installed confluent-kafka-2.15.0 (and any dependencies).

Installation failed?

Make sure you see (venv) in your terminal prompt before running pip install. If the virtual environment is not active, the install may fail due to permission issues.

On macOS, if you see an error about missing C compiler tools, run xcode-select --install first, then retry the pip install.

help me fix my confluent-kafka installation error

✔️ Awesome, I've got everything!

Great. Double check you've saved docker-compose.yml and that your (venv) prompt is visible in your terminal.

ⓧ I'd like to double check the full code

Here is the complete docker-compose.yml file:

services:
  broker:
    image: apache/kafka:4.3.1
    hostname: broker
    container_name: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker: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 Kafka broker is live and your Python environment is ready. Next up, you'll write a producer that streams sensor data into Kafka and see events flowing in real time.

Stream Your First Events with a Producer

Your Kafka broker is running and ready to accept events. Before building anything complex, you need to see data actually flowing through Kafka.

In this step, you will create a Python producer that sends simulated sensor temperature readings into a Kafka topic. You will watch events stream through the system in real time.

In this step, get ready to:
  • Create a producer script that generates and streams sensor events to Kafka.
  • Verify messages arrive in the topic using the Kafka console consumer.
Create the producer script

A producer is the component that sends events into a Kafka topic. Your producer will simulate temperature sensors and stream their readings as JSON events.

  • Create a new file called producer.py in your project folder.
  • Add the imports and topic creation function by pasting this code:
import json
import time
import random
from confluent_kafka import Producer
from confluent_kafka.admin import AdminClient, NewTopic

def create_topic():
    # Connect to the Kafka broker and create our topic
    admin = AdminClient({"bootstrap.servers": "localhost:9092"})
    topic = NewTopic("sensor-readings", num_partitions=3, replication_factor=1)
    futures = admin.create_topics([topic])
    for topic_name, future in futures.items():
        try:
            future.result()
            print(f"Topic '{topic_name}' created successfully")
        except Exception as e:
            print(f"Topic '{topic_name}' already exists or error: {e}")

def generate_sensor_reading():
    # Simulate a random sensor sending a temperature reading
    sensor_id = f"sensor-{random.randint(1, 5)}"
    temperature = random.uniform(20.0, 30.0)
    return {
        "sensor_id": sensor_id,
        "temperature": round(temperature, 2),
        "timestamp": time.time(),
    }

What does this code do?

The create_topic() function uses the AdminClient to create a topic called sensor-readings with 3 partitions. Partitions let Kafka distribute events across multiple storage segments for parallel processing.

The generate_sensor_reading() function simulates a temperature sensor by picking a random sensor ID (1 through 5) and a temperature between 20 and 30 degrees. Each reading is returned as a dictionary that you will later serialize to JSON.

  • Add the delivery callback and main loop below your existing code in producer.py:
def delivery_report(err, msg):
    # Callback that fires once Kafka confirms (or rejects) a message
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Sent: sensor={json.loads(msg.value())['sensor_id']} "
              f"temp={json.loads(msg.value())['temperature']} "
              f"partition={msg.partition()}")

def main():
    create_topic()
    producer = Producer({"bootstrap.servers": "localhost:9092"})
    print("Producer started. Streaming sensor data... (Ctrl+C to stop)")
    try:
        while True:
            reading = generate_sensor_reading()
            # Send the reading to Kafka, keyed by sensor_id
            producer.produce(
                topic="sensor-readings",
                key=reading["sensor_id"],
                value=json.dumps(reading),
                callback=delivery_report,
            )
            producer.poll(0)
            time.sleep(1)
    except KeyboardInterrupt:
        print("\nStopping producer...")
    finally:
        producer.flush()

if __name__ == "__main__":
    main()

What does this code do?

The delivery_report() callback fires after Kafka acknowledges each message. It prints the sensor ID, temperature, and which partition received the event.

The main() function creates the topic, then enters an infinite loop that generates one sensor reading per second and sends it to Kafka. The key=reading["sensor_id"] ensures events from the same sensor always land on the same partition.

  • Save producer.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before running it.

ⓧ I'd like to double check the full code

Your producer.py should contain these sections in order:

  • Imports (json, time, random, Producer, AdminClient, NewTopic).
  • The create_topic() function creating sensor-readings with 3 partitions.
  • The generate_sensor_reading() function returning a dict with sensor_id, temperature, and timestamp.
  • The delivery_report() callback printing confirmations.
  • The main() function with the producer loop and if __name__ == "__main__" guard at the bottom.

Scroll up and confirm each section is present in your file.

Run the producer and verify events arrive

Your producer script is ready. Time to see events actually flowing through Kafka.

  • Before you run this, what do you expect to see in the terminal? Think about what create_topic() and the delivery callback will print.
  • Start the producer by running:
python producer.py

You should see a topic creation confirmation followed by one event printed per second, showing the sensor ID, temperature, and which partition received the message.

What should I see?

Your terminal should print lines like:

Topic 'sensor-readings' created successfully
Producer started. Streaming sensor data... (Ctrl+C to stop)
Sent: sensor=sensor-3 temp=24.17 partition=2
Sent: sensor=sensor-1 temp=27.83 partition=0
Sent: sensor=sensor-5 temp=22.06 partition=1

The exact sensor IDs, temperatures, and partition numbers will differ each run because they are randomly generated.

Not seeing any output?

Make sure your Kafka broker container is still running. Run docker compose ps in a separate terminal and confirm the broker is listed as running.

If you see a connection error, wait a few seconds and try again. The broker may still be starting up.

Help me debug this connection issue

Leave the producer running. Now open a second terminal window to verify the messages are actually stored in the Kafka topic.

  • In a new terminal window, verify events are arriving in the topic by running the Kafka console consumer inside the Docker container:
docker exec broker /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic sensor-readings --from-beginning

You should see raw JSON messages scrolling by, one per second, matching the events your producer is sending.

What does this command do?

This runs the Kafka console consumer directly inside the broker container. The --from-beginning flag tells it to read all messages stored in the topic, not just new ones.

Seeing JSON events here confirms your data made the full trip: Python producer to Kafka broker to Kafka consumer.

  • Press Ctrl+C in the console consumer terminal to stop it.
  • Press Ctrl+C in the producer terminal to stop the producer.

Console consumer shows no messages?

Make sure you typed the topic name exactly as sensor-readings (with a hyphen, not an underscore).

Confirm your producer printed Sent: lines before you started the console consumer. If the producer errored out, no messages were written to the topic.

Help me troubleshoot the console consumer

You have data flowing through Kafka. Next up, you will build a consumer that reads these events and processes them in real time.

Consume Events and Hit the Single-Consumer Limitation

Your producer is streaming sensor data into Kafka. Now you need something on the other side to read those events.

In real ML systems, you often need multiple independent services processing the same data stream. One service detects anomalies. Another computes statistics. In this step, you will build a consumer and discover what happens when you try to run two consumers in the same consumer group.

In this step, get ready to:
  • Create a basic consumer that reads sensor events from Kafka.
  • Run two consumer instances and observe partition rebalancing.
  • Identify the limitation of a single consumer group for independent services.
Create your first consumer

A consumer is the counterpart to your producer. It subscribes to a topic and pulls events as they arrive.

  • Create a new file called consumer_basic.py in your project folder.
  • Add the imports and consumer setup by pasting this code:
import json
from confluent_kafka import Consumer, KafkaError

def main():
    # Configure the consumer with broker address and group settings
    consumer = Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "basic-group",
        "auto.offset.reset": "earliest",
    })
    # Subscribe to the sensor-readings topic
    consumer.subscribe(["sensor-readings"])
    print("Basic consumer started. Reading events... (Ctrl+C to stop)")

What does this code do?

  • group.id assigns this consumer to a consumer group called basic-group. Kafka uses this to coordinate which partitions each consumer reads.
  • auto.offset.reset set to earliest means this consumer starts reading from the beginning of the topic if it has never consumed before.
  • subscribe() tells the consumer which topic to listen to.
  • Now add the polling loop below the print statement. This is the heart of the consumer:
    try:
        while True:
            # Poll for a new message with a 1-second timeout
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                print(f"Consumer error: {msg.error()}")
                continue
            # Decode the JSON event and print key fields
            reading = json.loads(msg.value().decode("utf-8"))
            print(f"[Partition {msg.partition()}] "
                  f"sensor={reading['sensor_id']} "
                  f"temp={reading['temperature']}")
    except KeyboardInterrupt:
        print("\nStopping consumer...")
    finally:
        consumer.close()

if __name__ == "__main__":
    main()

What does this code do?

  • consumer.poll(timeout=1.0) asks Kafka for the next message, waiting up to 1 second before returning None.
  • The KafkaError._PARTITION_EOF check handles the case where the consumer reaches the end of a partition. This is normal and the consumer keeps polling for new events.
  • msg.value().decode("utf-8") converts the raw bytes back into a JSON string, which json.loads() parses into a Python dictionary.
  • The print shows which partition each message came from. This will be important when you observe the rebalancing behavior later.
  • Save consumer_basic.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

Scroll up and confirm your consumer_basic.py contains both chunks in order:

  • Chunk 1: imports, consumer configuration with the three settings, subscribe call, and the startup print statement.
  • Chunk 2: the try/while True polling loop with error handling, JSON decoding, partition print, and the if __name__ guard at the bottom.
Run the consumer and see events flowing

You need the producer running so there are events to consume. If it is not still running from the previous step, start it first.

  • Make sure the producer is running in one terminal. If you stopped it, start it again by running:
python producer.py
  • Open a second terminal window in the same project folder with your virtual environment activated.
  • Run the consumer by entering:
python consumer_basic.py

You should see sensor readings printing to the terminal with their partition numbers, like this:

[Partition 0] sensor=sensor-3 temp=24.51
[Partition 2] sensor=sensor-1 temp=27.83
[Partition 1] sensor=sensor-5 temp=22.09

Notice the partition numbers. Your topic has 3 partitions, and this single consumer is reading from all of them.

Not seeing any output?

  • Confirm the producer is running in another terminal. The consumer needs live events to display.
  • Check that your Kafka broker is still running with docker compose ps. If it stopped, run docker compose up -d again.
  • Make sure your virtual environment is activated in the terminal running the consumer.

Why isn't my consumer_basic.py receiving messages from Kafka?

Run a second consumer and observe the limitation

Your consumer works. Now imagine you need a second service processing the same events. Maybe an anomaly detector AND a statistics aggregator, both reading every single event.

  • Keep your first consumer running.
  • Open a third terminal window in the same project folder with your virtual environment activated.

Before you run a second instance, think about what you expect to happen. Will the second consumer see ALL the same events as the first one, or something different?

  • Run a second instance of the same script:
python consumer_basic.py

Watch both terminal windows carefully. You will notice something unexpected.

Each consumer now only receives messages from SOME partitions. One consumer might show [Partition 0] and [Partition 1] messages, while the other only shows [Partition 2].

What just happened?

Kafka performed a consumer rebalance. Because both consumers share the same group.id (basic-group), Kafka treats them as workers in the same team. It splits the 3 partitions between them so each message is processed only once.

This is great for scaling ONE service (processing messages faster by splitting the work). But it is terrible for running MULTIPLE independent services that each need to see every event.

This is the gap you will fix next

With this same-group approach, an anomaly detector would only see a fraction of the sensor readings. It would miss anomalies from sensors assigned to the other consumer. In the next step, you will solve this with separate consumer groups.

  • Stop both consumers by pressing Ctrl+C in each terminal.
  • Stop the producer with Ctrl+C as well.

You have just experienced Kafka's partition rebalancing firsthand. The next step fixes this by giving each service its own consumer group, so every service sees every event independently.

Build Independent Consumers with Separate Consumer Groups

In the previous step, you saw Kafka split partitions between two consumers in the same group. Each consumer only got a portion of the messages. That is how Kafka scales a single service horizontally.

But in a real ML pipeline, you need multiple independent services that each see every event. An anomaly detector and a statistics aggregator both need the full stream. The fix is giving each service its own consumer group.

In this step, get ready to:
  • Build an anomaly detector that flags unusual sensor readings using z-score statistics.
  • Build a statistics aggregator that computes running averages per sensor.
  • Run both consumers alongside the producer and confirm they each receive every event independently.
Create the anomaly detector consumer

The anomaly detector is your first ML-style consumer. It reads every sensor event, builds a statistical baseline per sensor, and flags any reading that deviates more than 2 standard deviations from the mean.

  • Create a new file called anomaly_detector.py in your project folder.
  • Add the consumer setup and message polling loop by pasting this code:
import json
import math
from confluent_kafka import Consumer, KafkaError

def main():
    consumer = Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "anomaly-detector-group",
        "auto.offset.reset": "earliest",
    })
    consumer.subscribe(["sensor-readings"])

    sensor_stats = {}
    print("Anomaly Detector started. Monitoring sensors... (Ctrl+C to stop)")

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

            reading = json.loads(msg.value().decode("utf-8"))
            sensor_id = reading["sensor_id"]
            temperature = reading["temperature"]

What does this code do?

  • group.id is set to anomaly-detector-group. This is a completely different group from basic-group, so this consumer gets its own independent copy of every event.
  • sensor_stats is an empty dictionary that will store running statistics for each sensor as readings arrive.
  • The poll loop is identical to your basic consumer. The difference is what happens after parsing each message.
  • Now add the z-score detection logic directly below the last line. This code continues inside the same loop, at the same indentation level:
            if sensor_id not in sensor_stats:
                sensor_stats[sensor_id] = {"readings": [], "count": 0}

            stats = sensor_stats[sensor_id]
            stats["readings"].append(temperature)
            stats["count"] += 1

            if stats["count"] >= 5:
                mean = sum(stats["readings"]) / len(stats["readings"])
                variance = sum((x - mean) ** 2 for x in stats["readings"]) / len(stats["readings"])
                std_dev = math.sqrt(variance) if variance > 0 else 0.01
                z_score = abs(temperature - mean) / std_dev

                if z_score > 2.0:
                    print(f"ALERT: {sensor_id} temp={temperature} "
                          f"z_score={z_score:.2f} (mean={mean:.2f}, std={std_dev:.2f})")
                else:
                    print(f"  OK:  {sensor_id} temp={temperature} "
                          f"z_score={z_score:.2f}")
            else:
                print(f"  WARMUP: {sensor_id} temp={temperature} "
                      f"(collecting baseline: {stats['count']}/5)")

    except KeyboardInterrupt:
        print("\nStopping anomaly detector...")
    finally:
        consumer.close()

if __name__ == "__main__":
    main()

How does the z-score detection work?

The detector waits until it has at least 5 readings per sensor to build a baseline. After that, it computes a z-score for every new reading.

  • A z-score measures how many standard deviations a value is from the mean. Above 2.0 means the reading is statistically unusual.
  • Readings below the threshold print as OK. Readings above it print as ALERT.
  • During the first 5 readings per sensor, you see WARMUP messages while the baseline is still being collected.
  • Save anomaly_detector.py.
  • Make sure your producer is running in a terminal. If it is not, start it by running:
python producer.py
  • Open a second terminal in your project folder and run the anomaly detector by running:
python anomaly_detector.py

You should see WARMUP messages for the first few readings per sensor, then OK messages as the baseline stabilizes.

  • Press Ctrl+C in the anomaly detector terminal to stop it for now.

Not seeing any output?

  • Check that your producer is still running in its terminal. The anomaly detector needs live events to process.
  • Make sure you saved anomaly_detector.py before running it. Unsaved changes are not picked up.
  • Verify the indentation matches exactly. Python is sensitive to whitespace, and the detection logic must be inside the while True loop.

Why isn't my anomaly detector printing any output?

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

Your anomaly_detector.py should have two sections in order:

  • The consumer setup and poll loop (imports, Consumer config, subscribe, and the message parsing loop).
  • The z-score detection logic (sensor_stats tracking, mean/variance/std_dev calculation, ALERT/OK/WARMUP printing, and the KeyboardInterrupt handler).

Scroll up and confirm both code blocks are present in your file, in order.

Create the statistics aggregator consumer

Your second consumer computes running statistics per sensor. It uses its own consumer group, so it receives the exact same events as the anomaly detector independently.

  • Create a new file called stats_aggregator.py in your project folder.
  • Add the consumer setup and message polling loop by pasting this code:
import json
from confluent_kafka import Consumer, KafkaError

def main():
    consumer = Consumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "stats-aggregator-group",
        "auto.offset.reset": "earliest",
    })
    consumer.subscribe(["sensor-readings"])

    sensor_stats = {}
    total_count = 0
    print("Stats Aggregator started. Computing statistics... (Ctrl+C to stop)")

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

            reading = json.loads(msg.value().decode("utf-8"))
            sensor_id = reading["sensor_id"]
            temperature = reading["temperature"]

What is different about this consumer?

  • group.id is stats-aggregator-group. This is a third unique group name in your project (alongside basic-group and anomaly-detector-group).
  • total_count tracks the total number of events processed across all sensors, used to trigger periodic summary printouts.
  • Add the statistics tracking and summary logic directly below the last line, at the same indentation level:
            if sensor_id not in sensor_stats:
                sensor_stats[sensor_id] = {"sum": 0.0, "count": 0, "min": float("inf"), "max": float("-inf")}

            stats = sensor_stats[sensor_id]
            stats["sum"] += temperature
            stats["count"] += 1
            stats["min"] = min(stats["min"], temperature)
            stats["max"] = max(stats["max"], temperature)
            total_count += 1

            if total_count % 10 == 0:
                print(f"\n--- Stats Summary (after {total_count} events) ---")
                for sid, s in sensor_stats.items():
                    avg = s["sum"] / s["count"]
                    print(f"  {sid}: avg={avg:.2f} min={s['min']:.2f} "
                          f"max={s['max']:.2f} count={s['count']}")
                print("---")

    except KeyboardInterrupt:
        print("\nStopping stats aggregator...")
    finally:
        consumer.close()

if __name__ == "__main__":
    main()

What does the aggregation logic do?

  • For each sensor, it tracks the sum, count, min, and max of all temperature readings.
  • Every 10 messages (when total_count % 10 == 0), it prints a summary table showing the average, min, max, and count for every sensor seen so far.
  • Save stats_aggregator.py.

Seeing an indentation error?

  • The second code block must be indented to align inside the same loop as the message parsing above it. Each line starts with 12 spaces (three levels of indentation).
  • Make sure there is no gap or blank line between the last line of the first block (temperature = reading["temperature"]) and the first line of the second block.

Help me fix my indentation in stats_aggregator.py

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

Your stats_aggregator.py should have two sections in order:

  • The consumer setup and poll loop (imports, Consumer config with stats-aggregator-group, subscribe, sensor_stats and total_count init, and the message parsing loop).
  • The statistics tracking and summary logic (per-sensor stats dict with sum/count/min/max, incrementing total_count, printing a summary every 10 events, and the KeyboardInterrupt handler).

Scroll up and confirm both code blocks are present in your file, in order.

Run all three scripts and prove consumer group isolation

This is the moment where separate consumer groups prove their value. You will run the producer and both consumers simultaneously and confirm that each consumer receives every event.

  • Make sure your producer is still running. If not, start it in its own terminal by running:
python producer.py
  • Open a second terminal in your project folder and start the anomaly detector by running:
python anomaly_detector.py
  • Open a third terminal in your project folder and start the stats aggregator by running:
python stats_aggregator.py

Before you look at the output, think about this: do you expect both consumers to print output for every event, or will they split events between them like the basic consumers did in the previous step?

Watch all three terminals for about 30 seconds. Both consumers print output for every single event the producer sends. The anomaly detector shows WARMUP then OK for each reading. The stats aggregator prints a summary table every 10 messages.

Why do both consumers see every event now?

Each consumer group maintains its own offset tracking independently. Kafka delivers every message to one consumer per group. Since your anomaly detector and stats aggregator are in different groups, each group gets the full stream.

This is the pattern that powers real ML pipelines. One topic of raw data can feed an anomaly detector, a feature store writer, a metrics dashboard, and a model retraining pipeline all at the same time, without any service interfering with another.

  • Press Ctrl+C in each terminal to stop all three scripts when you are done observing.

One consumer is not printing anything?

  • Make sure each terminal is running from your project folder. If the terminal opened in a different directory, the script will not be found.
  • If stats_aggregator.py seems silent, wait for 10 events. It only prints a summary every 10 messages.
  • Verify that the producer terminal is actively printing sent messages. Both consumers need live events to display output.

Help me debug why one of my consumers is not printing output

You now have two independent ML-style services consuming the same data stream in parallel. Next up, you will make the producer generate realistic anomalies so the detector has something interesting to flag.

Make the Producer Generate Realistic Anomalies

Your anomaly detector from the previous step is working. It can compute z-scores and flag outliers.

The problem is that your producer only generates temperatures between 20 and 30 degrees. With such a narrow, uniform range, the detector rarely triggers. In this step, you will update the producer to occasionally inject dramatic temperature spikes so the detection logic has something real to catch.

In this step, get ready to:
  • Update the producer to inject anomalous readings roughly 10% of the time.
  • Run the full pipeline and watch the anomaly detector flag spikes in real time.
Update the producer to inject anomaly spikes

Right now, every reading lands in the 20-30 degree range. The anomaly detector needs readings that are dramatically outside that range to demonstrate z-score detection working.

  • In producer.py, find the generate_sensor_reading() function:
def generate_sensor_reading():
    sensor_id = f"sensor-{random.randint(1, 5)}"
    temperature = random.uniform(20.0, 30.0)
    return {
        "sensor_id": sensor_id,
        "temperature": round(temperature, 2),
        "timestamp": time.time(),
    }
  • Replace the entire generate_sensor_reading() function with this updated version:
def generate_sensor_reading():
    sensor_id = f"sensor-{random.randint(1, 5)}"
    # 10% chance of generating an anomalous spike
    is_anomaly = random.random() < 0.1
    if is_anomaly:
        temperature = random.uniform(150.0, 200.0)
    else:
        temperature = random.uniform(20.0, 30.0)
    return {
        "sensor_id": sensor_id,
        "temperature": round(temperature, 2),
        "timestamp": time.time(),
        "is_anomaly": is_anomaly,
    }

What does this code do?

The function now flips a coin on each event. If random.random() < 0.1 is true (roughly 10% of the time), it generates a temperature between 150 and 200 degrees instead of the normal 20-30 range.

The is_anomaly field acts as ground truth metadata. In production ML systems, this kind of labeling lets you measure how well your detection model performs against known anomalies.

  • Save producer.py.
  • Run the updated producer to confirm it generates mixed readings:
python producer.py

Watch the output for about 10 seconds. Most readings show temperatures in the 20-30 range. Every few seconds, you should see a spike with a temperature between 150 and 200.

  • Press Ctrl+C to stop the producer once you have confirmed the spikes appear.

Only seeing normal temperatures?

Since anomalies only happen 10% of the time, you may need to wait 10-15 seconds to see one. If nothing appears after 20 seconds, check that you replaced the entire function body and saved the file.

Make sure the is_anomaly = random.random() < 0.1 line is inside the function, not outside it.

Help me debug my producer not generating anomalies.

Run the full pipeline and observe anomaly detection

Now that the producer injects spikes, you can see the entire system work end-to-end. The anomaly detector should flag the spikes while the stats aggregator processes everything independently.

  • You need three terminal windows. Open two new terminal windows alongside your current one.
  • In each new terminal, activate your virtual environment and navigate to your project folder.
  • In the first terminal, start the producer by running:
python producer.py
  • In the second terminal, start the anomaly detector by running:
python anomaly_detector.py
  • In the third terminal, start the stats aggregator by running:
python stats_aggregator.py

Before you watch the output, think about this: what do you expect to see in the anomaly detector terminal when a 175-degree reading arrives after a baseline of 20-30 degree readings?

Within 30 seconds, you should see ALERT messages in the anomaly detector terminal whenever a spike event arrives. The z-score for those readings is far above 2.0 because 150+ degrees is many standard deviations from the 20-30 degree baseline. Meanwhile, the stats aggregator terminal continues printing summaries every 10 events, processing everything independently.

Why do both consumers see every event?

The anomaly detector uses consumer group anomaly-detector-group and the stats aggregator uses stats-aggregator-group. Because they belong to different consumer groups, Kafka delivers a copy of every event to each group independently.

This is the core pattern powering real-time ML systems. One data stream feeds multiple services without any coordination between them.

  • Let the pipeline run for at least 30 seconds to observe several anomaly spikes.
  • Press Ctrl+C in each terminal to stop all three scripts when you are done observing.

Not seeing ALERT messages in the anomaly detector?

The detector needs at least 5 readings from a sensor before it starts computing z-scores. Watch for WARMUP messages first. After the warmup phase, anomalous readings trigger ALERT output.

If you still only see OK messages after 30 seconds, confirm your producer is sending spikes by checking the first terminal for temperatures above 150.

Help me figure out why my anomaly detector isn't printing ALERT messages.

✔️ Awesome, I've got everything!

Great. Make sure you have saved producer.py with the updated generate_sensor_reading() function.

ⓧ I'd like to double check the full code

Your producer.py should contain these sections in order. Scroll up and confirm each is present:

  • Imports: json, time, random, Producer, AdminClient, NewTopic.
  • The create_topic() function (unchanged from the earlier step).
  • The updated generate_sensor_reading() function with is_anomaly logic and temperature ranges 150-200 (anomaly) or 20-30 (normal).
  • The delivery_report() callback function (unchanged).
  • The main() function with the produce loop (unchanged).

Your pipeline is alive. Sensor data streams in, anomalies get flagged, and statistics get computed. All three services run independently on the same event stream. Next up, you will tackle the Secret Mission and route those flagged anomalies to a dedicated topic for investigation.

Secret mission

Add a Dead Letter Topic for Anomalies

In production ML systems, flagged anomalies can't just be printed to a terminal. They need to go somewhere durable for investigation. Your mission: implement the consume-transform-produce pattern by routing anomalous events to a dedicated dead letter topic.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. The only resource using system memory is the Docker container running the Kafka broker.

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

Resources you used:

  • Docker container running the Kafka broker (uses ~300-500 MB RAM while running).
  • Project folder on your Desktop containing Python scripts and a virtual environment.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with your Kafka pipeline.

  • Your Kafka broker container continues running in the background.
  • You can restart any of your Python scripts at any time to resume streaming events.
  • The confluent-kafka package remains installed inside your virtual environment.

Pause - I'll come back to this later

Shut down the Kafka container to free up memory while keeping your data and project files intact.

  • Stop any running Python scripts (producer, consumers) with Ctrl+C in each terminal.
  • Stop the Kafka container by running:
docker compose stop

When you are ready to resume, run docker compose start from your project folder to bring the broker back up with all your topic data preserved.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes the Kafka container, all topic data, and your project files.

  • Stop any running Python scripts with Ctrl+C in each terminal.
  • Remove the Kafka container and all its data by running:
docker compose down -v
  • Delete the project folder from your Desktop by running:

macOS

cd ~/Desktop
rm -rf kafka-pipeline

Windows

cd ~\Desktop
rmdir /s /q kafka-pipeline

This removes your virtual environment, all Python scripts, and the docker-compose.yml file.

That's a Wrap!

That's a Wrap!

Nice work. You just built a real-time ML prediction pipeline powered by Apache Kafka, streaming sensor data through independent consumers that detect anomalies and aggregate statistics in parallel.

You've learned how to:

  • Set up a local Apache Kafka cluster using Docker Compose with KRaft mode, and built a Python producer that streams simulated sensor events as JSON to a Kafka topic.
  • Experienced consumer group partition rebalancing firsthand and understood why independent services need separate consumer groups to each receive every event.
  • Built an anomaly detection consumer that applies z-score statistics to flag outlier readings in real time, alongside a statistics aggregation consumer that processes the same stream independently.
  • Secret Mission: Implemented the consume-transform-produce pattern by routing anomalies to a dead letter topic for later investigation.

Ready to quiz yourself?