Infrastructure Telemetry Pipeline

A real-time end-to-end data pipeline that simulates, ingests, processes, and serves server telemetry data. This project demonstrates how to handle streaming data using a modern data engineering stack.

Architecture & Tech Stack

• Data Generation: A Python script simulating realistic telemetry data (CPU, Memory, Disk, IOPS) for 10 mock servers.
• Message Broker: Apache Kafka (with Zookeeper) for high-throughput, ordered message streaming.
• Stream Processing: PySpark (Structured Streaming) to consume Kafka messages, handle late data with watermarking, compute 1-minute windowed aggregations, and sink the results to a database.
• Storage: PostgreSQL to store the aggregated metrics.
• API Layer: FastAPI (with SQLAlchemy and asyncpg) serving asynchronous REST endpoints to query the aggregated metrics.
• Real-time Alerting: A standalone Python Kafka Consumer to monitor the raw stream and detect CPU anomalies in real time.
• Containerization: Docker Compose for running Kafka, Zookeeper, and PostgreSQL locally.

Prerequisites

• Docker & Docker Compose (for running Kafka and PostgreSQL)
• Python 3.9+
• Java 11+ (Required for PySpark)

Setup Instructions

1. Environment Configuration

Create a .env file in the root directory by copying the example file:

cp .env.example .env

Update the values in .env if necessary. The defaults work securely for local development.

2. Start Infrastructure

Bring up Zookeeper, Kafka, and PostgreSQL using Docker Compose:

docker-compose up -d

3. Database Initialization

Before running the pipeline, you need to create the target table in PostgreSQL. Connect to your PostgreSQL instance (e.g., using psql or DBeaver) and run the following SQL:

CREATE TABLE metrics_aggregated (
    server_id VARCHAR(50),
    window_start TIMESTAMP,
    window_end TIMESTAMP,
    cpu_avg FLOAT,
    mem_max FLOAT,
    disk_p95 FLOAT,
    PRIMARY KEY (server_id, window_start)
);

-- Create indexes to optimize FastAPI queries
CREATE INDEX idx_server_window ON metrics_aggregated(server_id, window_start DESC);
CREATE INDEX idx_high_cpu ON metrics_aggregated(cpu_avg) WHERE cpu_avg > 80;

4. Install Python Dependencies

It is recommended to use a virtual environment:

python -m venv venv
source venv/bin/activate  # On Windows use: venv\Scripts\activate

pip install fastapi uvicorn sqlalchemy asyncpg pyspark kafka-python python-dotenv psycopg2-binary

---

Running the Pipeline

To see the pipeline in action, you'll need to run the following components in separate terminal windows. Make sure your virtual environment is activated in each.

Step 1: Start the Data Producer

The producer will start generating and streaming simulated server metrics to the Kafka server-metrics topic.

python producer.py

Step 2: Start the PySpark Processor

The Spark structured streaming job will consume messages from Kafka, perform 1-minute windowed aggregations, and upsert them into PostgreSQL.

python spark_processor.py

Step 3: Start the FastAPI Server

The API server allows you to query the aggregated data stored in PostgreSQL.

uvicorn main:app --reload

You can now view the interactive API documentation at: http://localhost:8000/docs

Step 4 (Optional): Start the Real-time Alerting Consumer

If you want to monitor the raw telemetry stream and get real-time terminal alerts for CPU spikes (> 80%), run the consumer:

python consumer.py

---

API Endpoints

The FastAPI application exposes the following endpoints:

GET /servers/{server_id}/metrics

Fetches the last 10 aggregated 1-minute windows for a specific server.

• Example: curl http://localhost:8000/servers/server-01/metrics

GET /alerts

Returns all servers with an average CPU usage greater than 80% across any 1-minute window. Uses a partial database index for instant lookups.

• Example: curl http://localhost:8000/alerts

GET /analytics/summary

Provides a cluster-wide health check, returning the number of active servers and the cluster's overall average CPU.

• Example: curl http://localhost:8000/analytics/summary

---

Stopping the Project

To stop the Python processes, use Ctrl+C in their respective terminal windows.

To tear down the Docker infrastructure and remove the containers:

docker-compose down

(Note: If you want to clear the PostgreSQL data as well, use docker-compose down -v to remove the volumes).