Real-Time Weather Data Pipeline

Introduction

30 Second Summary

Weather apps feel simple, but behind each live temperature reading is a streaming data pipeline pulling from APIs, queuing messages, and processing events in real time. That same architecture powers everything from financial trading platforms to IoT sensor networks.

In this project, you will build a complete real-time weather monitoring platform from scratch. You'll ingest live data from an API, stream it through Kafka, process it with Spark Structured Streaming, store it in a layered data lake, and visualize it on a live dashboard.

What You'll Build

A real-time weather monitoring platform that tracks live weather data across multiple cities using the full API to Kafka to Spark to storage to dashboard streaming pattern.

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

• A Python API producer that fetches live weather data from OpenWeatherMap with retry logic, rate limiting, and robust error handling.
• A Kafka-to-Spark streaming pipeline with event-time windowing, watermarking, and checkpointing for fault tolerance.
• A medallion data lake on MinIO (bronze, silver, gold layers) plus a real-time Streamlit dashboard with auto-refreshing KPIs and visualizations.
• Secret Mission: Add Prometheus metrics and a Grafana monitoring dashboard for production observability.

Are there any prerequisites?

You should be comfortable with Python basics (functions, classes, imports) and have a general understanding of Docker containers. Familiarity with basic SQL and data concepts (schemas, tables, filtering) will also help.

Before We Start

Before We Start...

Set Up Your Environment

Before you can build a streaming pipeline, you need the right tools in place. This project relies on Docker for running infrastructure services, Python for the application code, and an API key from OpenWeatherMap for live weather data.

In this step, you will install Docker, set up a Python virtual environment with all required packages, and configure your API credentials securely.

In this step, get ready to:

• Install Docker and Docker Compose, then create the project directory structure.
• Create a Python virtual environment and install project dependencies.
• Obtain an OpenWeatherMap API key and configure environment files securely.

Install Docker and create the project structure

Docker containers will run your Kafka broker and MinIO storage later in the project. You need both Docker Engine and the Compose plugin installed on your Linux machine.

• Check whether Docker is already installed by running this command:

docker --version

✔️ I see a version number

Docker is already installed. Skip ahead to the Docker Compose check below.

ⓧ Command not found

Docker is not installed yet. Follow the steps below to install it from Docker's official apt repository.

• Set up Docker's apt repository and install Docker Engine by running these commands:

# Add Docker's official GPG key:
sudo apt update
sudo apt install ca-certificates curl
sudo install -m 0755 -d /etc/apt/keyrings
sudo curl -fsSL https://download.docker.com/linux/ubuntu/gpg -o /etc/apt/keyrings/docker.asc
sudo chmod a+r /etc/apt/keyrings/docker.asc

# Add the repository to Apt sources:
sudo tee /etc/apt/sources.list.d/docker.sources <<EOF
Types: deb
URIs: https://download.docker.com/linux/ubuntu
Suites: $(. /etc/os-release && echo "${UBUNTU_CODENAME:-$VERSION_CODENAME}")
Components: stable
Architectures: $(dpkg --print-architecture)
Signed-By: /etc/apt/keyrings/docker.asc
EOF

sudo apt update

• Install the Docker packages by running:

sudo apt install docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin

What does this install?

This installs Docker Engine, the Docker CLI, containerd (the container runtime), the Buildx plugin for building images, and the Compose plugin for multi-container orchestration.

• Verify Docker is now working by running:

docker --version

You should see a version number printed to your terminal.

Run Docker without sudo

By default, Docker requires sudo for every command. To run Docker as your regular user, add yourself to the docker group:

sudo usermod -aG docker $USER

Log out and back in (or run newgrp docker) for the change to take effect.

• Verify Docker Compose is available by running:

docker compose version

You should see a Compose version number. The Compose plugin is bundled with the Docker packages you just installed.

docker compose not found?

Make sure you installed docker-compose-plugin (not the older standalone docker-compose). Re-run the install command above to add the plugin.

Help me fix this

• Move to your Desktop by running this command:

cd ~/Desktop

• Create the project directory structure by running this command:

mkdir -p weather-pipeline/{producer,streaming,dashboard,tests,infra,docs}

• Move into the project folder by running:

cd weather-pipeline

• Confirm the directories were created by running:

ls

You should see dashboard, docs, infra, producer, streaming, and tests listed in the output.

What is this directory structure for?

Each folder isolates one responsibility of the pipeline. producer/ holds the API ingestion code, streaming/ holds the Spark jobs, dashboard/ holds the Streamlit app, tests/ holds quality checks and unit tests, infra/ holds Docker Compose files, and docs/ holds documentation.

Create the Python virtual environment and install dependencies

A virtual environment keeps your project's Python packages isolated from your system Python. This prevents version conflicts with other projects on your machine.

• Create and activate a Python virtual environment by running these commands:

python3.12 -m venv .venv
source .venv/bin/activate

Your terminal prompt should now show (.venv) at the beginning, confirming the virtual environment is active.

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install with pip go into .venv/ rather than your system directories.

This means each project can have its own package versions without conflict.

• Install the project dependencies by running:

pip install pyspark==4.1.2 confluent-kafka==2.14.0 minio==7.2.20 streamlit==1.57.0 requests==2.32.3 pytest==9.0.3 python-dotenv==1.1.0

What are these packages?

• pyspark is the Python interface for Apache Spark, used for stream processing.
• confluent-kafka is the official Python client for Apache Kafka.
• minio is the SDK for MinIO, an S3-compatible object store.
• streamlit is the framework for building the real-time dashboard.
• requests handles HTTP calls to the weather API.
• pytest is the testing framework.
• python-dotenv loads environment variables from .env files into your application.

The installation may take a few minutes. You should see Successfully installed at the end of the output.

Installation failing?

Make sure your virtual environment is active (you should see (.venv) in your prompt). If pip is not found, run python3.12 -m pip install --upgrade pip first.

Help me troubleshoot this

Get your OpenWeatherMap API key and configure environment files

The pipeline fetches live weather data from the OpenWeatherMap API. You need a free account and an API key to authenticate your requests.

• Open your browser and navigate to https://home.openweathermap.org/users/sign_up.
• Fill in the signup form with your email, a username, and a password.
• Agree to the Terms and Conditions, then click Create Account.
• Check your email inbox for a verification email from OpenWeather and click the confirmation link.
• Once verified, log in and click the API keys tab in your account settings.
• Copy the API key shown on that page. This is your free-tier key.

How long until the key works?

Your API key activates automatically within 2 hours of registration. The free tier allows 60 API calls per minute and 1,000,000 calls per month, which is far more than this project needs.

Now you will create the environment files that store your configuration securely.

• Create the .env file by running this command (make sure you are inside weather-pipeline/):

cat > .env << 'EOF'
OPENWEATHER_API_KEY=your-actual-key-here
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
POLLING_INTERVAL=30
CITIES=London,Tokyo,New York,Sydney,Paris
EOF

• Open .env in a text editor and replace your-actual-key-here with the API key you just copied from OpenWeatherMap.

What are these environment variables?

• OPENWEATHER_API_KEY authenticates your requests to the weather API.
• KAFKA_BOOTSTRAP_SERVERS tells the producer where to find the Kafka broker.
• POLLING_INTERVAL sets how often (in seconds) the producer fetches new data.
• CITIES is the comma-separated list of cities to monitor.

• Create the .env.example file (a safe-to-commit template) by running:

cat > .env.example << 'EOF'
OPENWEATHER_API_KEY=your-key-here
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
POLLING_INTERVAL=30
CITIES=London,Tokyo,New York,Sydney,Paris
EOF

Why have .env.example?

The .env.example file shows teammates what variables the project needs without exposing your real credentials. It gets committed to version control while the actual .env stays private.

• Create the .gitignore file to prevent secrets from being committed by running:

cat > .gitignore << 'EOF'
.env
.venv/
__pycache__/
*.pyc
.pytest_cache/
/tmp/
EOF

• Verify all three files exist by running:

ls -a

You should see .env, .env.example, and .gitignore alongside your project directories.

Don't see the dotfiles?

Files starting with a dot are hidden by default. Make sure you use ls -a (with the -a flag) to show hidden files.

Help me list hidden files

✔️ Awesome, I've got everything!

Great. Double check that you have replaced the placeholder API key in .env with your real OpenWeatherMap key.

ⓧ I'd like to double check the full code

Here are the three configuration files you should have in your weather-pipeline/ directory.

.env

OPENWEATHER_API_KEY=your-actual-key-here
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
POLLING_INTERVAL=30
CITIES=London,Tokyo,New York,Sydney,Paris

.env.example

OPENWEATHER_API_KEY=your-key-here
KAFKA_BOOTSTRAP_SERVERS=localhost:9092
POLLING_INTERVAL=30
CITIES=London,Tokyo,New York,Sydney,Paris

.gitignore

.env
.venv/
__pycache__/
*.pyc
.pytest_cache/
/tmp/

Your environment is fully set up. Next, you will explore the OpenWeatherMap API documentation and design the event schema for your pipeline.

Read the OpenWeatherMap API Docs

Your environment is set up and your API key is ready. Now it's time to understand the data source your entire pipeline depends on.

Before writing any code, you need to read the API documentation systematically. This means identifying the base URL, required parameters, rate limits, and response schema. Documenting these details upfront prevents surprises mid-build and gives your pipeline a clear contract to code against.

In this step, get ready to:

• Document the OpenWeatherMap API spec in a reference file.
• Test the API with a manual curl call and verify the response.
• Define the event schema your pipeline will use.

Document the API specification

Every data pipeline starts with a contract. For your weather pipeline, that contract is the OpenWeatherMap Current Weather API documentation. You need to identify what the API expects (parameters, authentication) and what it returns (fields, formats, limits).

• Open the OpenWeatherMap Current Weather API docs in your browser:

https://openweathermap.org/current

What are you looking for in API docs?

When reading any API documentation, focus on five things: the base URL (where to send requests), required parameters (what you must include), optional parameters (what you can customise), the authentication method (how you prove you have access), and rate limits (how often you can call it).

• Create the API spec file by running this command from your weather-pipeline directory:

touch docs/api_spec.md

• Open docs/api_spec.md in your text editor and paste the following documentation:

# OpenWeatherMap Current Weather API Spec

## Base URL
https://api.openweathermap.org/data/2.5/weather

## Authentication
API key passed as query parameter: appid={key}

## Required Parameters
- lat, lon OR q (city name)
- appid

## Optional Parameters
- units: metric | imperial | standard
- lang: language code
- mode: json (default) | xml | html

## Rate Limits
- Free tier: 60 calls/minute, 1,000,000 calls/month

## Key Response Fields
- main.temp: temperature
- main.humidity: humidity %
- wind.speed: wind speed m/s
- wind.deg: wind direction degrees
- name: city name
- dt: unix timestamp
- weather[0].main: condition string

What does this document capture?

The base URL https://api.openweathermap.org/data/2.5/weather is where every request goes. The appid parameter is your API key passed as a query string. You can request by geographic coordinates (lat, lon) or by city name (q).

The free tier allows 60 calls per minute and 1,000,000 calls per month. Your pipeline will stay well within these limits.

• Save docs/api_spec.md.
• Confirm the file exists by running:

ls docs/

You should see api_spec.md listed in the output.

Don't see the file?

Make sure you are in the weather-pipeline directory before running touch. If the docs/ folder does not exist, create it first with mkdir -p docs.

help me troubleshoot why my docs/api_spec.md file wasn't created

Test the API with a manual curl call

Before building an automated producer, you should verify the API works with your key and returns the fields you expect. A quick curl call is the simplest way to do this.

• Load your environment variables into the current shell session by running:

source .env

• Test the API by fetching current weather data for London:

curl "https://api.openweathermap.org/data/2.5/weather?q=London&units=metric&appid=$OPENWEATHER_API_KEY"

What does this curl command do?

This sends a GET request to the Current Weather API for London with units=metric (returns Celsius and m/s). The $OPENWEATHER_API_KEY shell variable expands to your actual key from the .env file you sourced.

The response is a JSON object containing main.temp, main.humidity, wind.speed, wind.deg, name, dt, and weather[0].main. These are the exact fields your pipeline will extract.

You should see a JSON response containing temperature, humidity, wind data, and the city name "London" in the output.

Seeing an error instead of JSON?

• If you see {"cod":401,"message":"Invalid API key"}, double-check the key in your .env file. New keys can take up to 2 hours to activate after registration.
• If you see HTTP 429, you have exceeded 60 calls per minute. Wait a minute and try again.
• If $OPENWEATHER_API_KEY appears literally in the URL (not expanded), make sure you ran source .env first.

help me debug why my OpenWeatherMap API curl call is failing

Define the pipeline event schema

The raw API response contains many fields your pipeline does not need. Defining a lean event schema now means every downstream component (Kafka messages, Spark processing, storage) works with a consistent, predictable structure.

• Create the event schema file by running:

touch docs/event_schema.md

• Open docs/event_schema.md and paste the following schema definition:

# Pipeline Event Schema

## Fields
- city: string
- temperature: float
- humidity: int
- wind_speed: float
- wind_deg: int
- weather_condition: string
- event_time: ISO8601 string
- ingestion_time: ISO8601 string

## Rate Limit Strategy
Poll 5 cities every 30 seconds = 10 calls/minute (well within 60/min free tier limit).

Why define a separate event schema?

The raw API returns dozens of fields (pressure, visibility, sunrise, icon codes). Your pipeline only needs the eight fields listed above. By defining this contract, your producer knows exactly what to extract, your Spark job knows what to expect, and your dashboard knows what to query.

The two timestamp fields serve different purposes. The event_time comes from the API's dt field (when the weather was measured). The ingestion_time records when your producer fetched it. This distinction matters for late-arriving data handling later.

Why 10 calls per minute?

Your .env file configures 5 cities polled every 30 seconds. That is 5 calls every 30 seconds, or 10 calls per minute. The free tier allows 60 calls per minute, so you are using less than 17% of your allowance. This leaves plenty of headroom for retries and testing.

• Save docs/event_schema.md.
• Confirm both documentation files exist by running:

ls docs/

You should see both api_spec.md and event_schema.md in the output.

✔️ Awesome, I've got everything!

Great. Double check you have saved both files in the docs/ directory.

ⓧ I'd like to double check the full code

Here is the full content of both files for comparison.

# OpenWeatherMap Current Weather API Spec

## Base URL
https://api.openweathermap.org/data/2.5/weather

## Authentication
API key passed as query parameter: appid={key}

## Required Parameters
- lat, lon OR q (city name)
- appid

## Optional Parameters
- units: metric | imperial | standard
- lang: language code
- mode: json (default) | xml | html

## Rate Limits
- Free tier: 60 calls/minute, 1,000,000 calls/month

## Key Response Fields
- main.temp: temperature
- main.humidity: humidity %
- wind.speed: wind speed m/s
- wind.deg: wind direction degrees
- name: city name
- dt: unix timestamp
- weather[0].main: condition string

# Pipeline Event Schema

## Fields
- city: string
- temperature: float
- humidity: int
- wind_speed: float
- wind_deg: int
- weather_condition: string
- event_time: ISO8601 string
- ingestion_time: ISO8601 string

## Rate Limit Strategy
Poll 5 cities every 30 seconds = 10 calls/minute (well within 60/min free tier limit).

You now have a documented API contract and a verified working connection. Next up, you will spin up the Docker infrastructure that powers your pipeline.

Build the Docker Compose Infrastructure

You have your API documentation and event schema defined. Now it is time to build the actual infrastructure that will carry data through your pipeline.

Your streaming platform needs three core services: a message broker to queue weather events, an object store to hold processed data, and topic channels that organize messages by stage. Docker Compose lets you define and start all three with a single command.

In this step, get ready to:

• Create a Docker Compose stack with Kafka, MinIO, and a topic initializer.
• Start the infrastructure and verify all services are running.
• Initialize MinIO with bronze, silver, and gold buckets for your medallion architecture.

Create the Docker Compose configuration

Your pipeline needs Apache Kafka as a message broker and MinIO as S3-compatible object storage. Both run as containers orchestrated by a single docker-compose.yml file.

• Create the file infra/docker-compose.yml by running:

touch infra/docker-compose.yml

• Open infra/docker-compose.yml in your editor and paste the following configuration:

version: '3.8'
services:
  # Kafka broker in KRaft mode (no ZooKeeper)
  broker:
    image: apache/kafka:latest
    container_name: broker
    ports:
      - 9092:9092
    environment:
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:9093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CLUSTER_ID: ${KAFKA_CLUSTER_ID}
  # S3-compatible object storage
  minio:
    image: minio/minio
    container_name: minio
    ports:
      - 9000:9000
      - 9001:9001
    environment:
      MINIO_ROOT_USER: ${MINIO_ROOT_USER}
      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD}
    command: server /data --console-address :9001
    volumes:
      - minio_data:/data
  # One-shot container that creates Kafka topics on startup
  kafka-init:
    image: apache/kafka:latest
    depends_on:
      - broker
    entrypoint: ["/bin/sh", "-c"]
    command: "sleep 10 && /opt/kafka/bin/kafka-topics.sh --create --if-not-exists --bootstrap-server broker:9092 --topic raw_weather --partitions 3 --replication-factor 1 && /opt/kafka/bin/kafka-topics.sh --create --if-not-exists --bootstrap-server broker:9092 --topic clean_weather --partitions 3 --replication-factor 1 && /opt/kafka/bin/kafka-topics.sh --create --if-not-exists --bootstrap-server broker:9092 --topic aggregates --partitions 3 --replication-factor 1"
volumes:
  minio_data:

What does this configuration do?

This file defines three services that form your pipeline's backbone:

• broker runs Apache Kafka in KRaft mode. KRaft replaces ZooKeeper entirely, so a single container handles both broker and controller roles. Port 9092 is exposed so your Python producer can send messages from outside Docker.
• minio provides S3-compatible object storage on port 9000 (API) and port 9001 (web console). The minio_data volume persists your data across container restarts.
• kafka-init is a one-shot container. It waits 10 seconds for the broker to start, then creates three topics: raw_weather, clean_weather, and aggregates. The --if-not-exists flag makes it safe to re-run without errors.

Environment variables like ${KAFKA_CLUSTER_ID} and ${MINIO_ROOT_USER} are pulled from a .env file in the same directory as the compose file. You will create that next.

Why KRaft instead of ZooKeeper?

Starting with Confluent Platform 8.0, ZooKeeper has been removed entirely. KRaft (Kafka Raft) is the only option going forward. It simplifies deployment by eliminating an extra service and reduces operational overhead.

• Save infra/docker-compose.yml.

You will not see a change yet. The next substep creates the environment file and starts the stack.

Start the infrastructure stack

Docker Compose reads a .env file from the same directory as docker-compose.yml to interpolate variables. You need one at infra/.env with your MinIO credentials and a Kafka cluster ID.

• Create the infrastructure environment file by running:

cat > infra/.env << 'EOF'
MINIO_ROOT_USER=minioadmin
MINIO_ROOT_PASSWORD=minioadmin
KAFKA_CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qg
EOF

What are these values?

MINIO_ROOT_USER and MINIO_ROOT_PASSWORD are the default MinIO admin credentials. You will use these to log in to the MinIO web console.

KAFKA_CLUSTER_ID is a base64-encoded unique identifier for your single-node Kafka cluster. KRaft mode requires it to initialize the metadata log.

• Add infra/.env to your .gitignore to keep credentials out of version control by running:

echo 'infra/.env' >> .gitignore

• Start the Docker Compose stack by running:

docker compose -f infra/docker-compose.yml up -d

What does this command do?

The -f flag points Docker Compose at the file inside infra/. The -d flag runs all containers in detached mode so your terminal stays free. Docker Compose automatically reads infra/.env because it sits in the same directory as the compose file.

Docker will pull the Kafka and MinIO images on the first run. This may take a minute depending on your connection speed.

• Verify all containers are running by running:

docker compose -f infra/docker-compose.yml ps

You should see broker and minio with a status of Up. The kafka-init container will show Exited (0) because it completed its topic-creation task and shut down.

• Verify the Kafka topics were created by running:

docker exec broker /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

You should see three topic names in the output: raw_weather, clean_weather, and aggregates.

• Verify MinIO is accessible by opening http://localhost:9001 in your browser.
• Log in with username minioadmin and password minioadmin.

You should see the MinIO web console with an empty object browser.

Containers not starting?

• Make sure Docker Desktop or the Docker daemon is running.
• Check that ports 9000, 9001, and 9092 are not already in use by another process.
• Run docker compose -f infra/docker-compose.yml logs to see detailed error output from each service.

Help me troubleshoot my Docker Compose stack for Kafka and MinIO.

Initialize MinIO buckets for the medallion architecture

Your pipeline stores data in three layers: raw ingestion (bronze), cleaned data (silver), and aggregated results (gold). Each layer maps to a separate bucket in MinIO. A Python script creates these buckets idempotently so it is safe to re-run.

• Create the bucket initialization script by running:

touch infra/init_minio.py

• Open infra/init_minio.py and paste the following code:

from minio import Minio

# Connect to local MinIO instance
client = Minio(
    'localhost:9000',
    access_key='minioadmin',
    secret_key='minioadmin',
    secure=False
)

# Create medallion architecture buckets if they don't exist
buckets = ['bronze', 'silver', 'gold']
for bucket in buckets:
    if not client.bucket_exists(bucket):
        client.make_bucket(bucket)
        print(f'Created bucket: {bucket}')
    else:
        print(f'Bucket already exists: {bucket}')

What does this script do?

• client.bucket_exists(bucket) checks whether the bucket already exists before trying to create it. This makes the script idempotent.
• client.make_bucket(bucket) creates the bucket in MinIO.
• secure=False disables TLS because your local MinIO instance runs over plain HTTP.

The three buckets map to the medallion architecture: bronze holds raw data as-is, silver holds cleaned and validated data, and gold holds aggregated results ready for dashboards.

• Save infra/init_minio.py.
• Run the script to create the buckets:

python infra/init_minio.py

You should see output confirming all three buckets were created:

Created bucket: bronze
Created bucket: silver
Created bucket: gold

• Verify the buckets by refreshing the MinIO console at http://localhost:9001.

You should see bronze, silver, and gold listed in the object browser.

Seeing a connection error when running the script?

• Make sure your Docker Compose stack is still running. Run docker compose -f infra/docker-compose.yml ps and confirm the minio container shows Up.
• Confirm you installed the minio Python package in your virtual environment during Step 1. Run pip show minio to check.

Help me fix a connection error when running my MinIO bucket initialization script.

✔️ Awesome, I've got everything!

Great. Double check you have saved both infra/docker-compose.yml and infra/init_minio.py.

ⓧ I'd like to double check the full code

Here is the complete infra/docker-compose.yml:

version: '3.8'
services:
  broker:
    image: apache/kafka:latest
    container_name: broker
    ports:
      - 9092:9092
    environment:
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:9093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CLUSTER_ID: ${KAFKA_CLUSTER_ID}
  minio:
    image: minio/minio
    container_name: minio
    ports:
      - 9000:9000
      - 9001:9001
    environment:
      MINIO_ROOT_USER: ${MINIO_ROOT_USER}
      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD}
    command: server /data --console-address :9001
    volumes:
      - minio_data:/data
  kafka-init:
    image: apache/kafka:latest
    depends_on:
      - broker
    entrypoint: ["/bin/sh", "-c"]
    command: "sleep 10 && /opt/kafka/bin/kafka-topics.sh --create --if-not-exists --bootstrap-server broker:9092 --topic raw_weather --partitions 3 --replication-factor 1 && /opt/kafka/bin/kafka-topics.sh --create --if-not-exists --bootstrap-server broker:9092 --topic clean_weather --partitions 3 --replication-factor 1 && /opt/kafka/bin/kafka-topics.sh --create --if-not-exists --bootstrap-server broker:9092 --topic aggregates --partitions 3 --replication-factor 1"
volumes:
  minio_data:

Here is infra/.env:

MINIO_ROOT_USER=minioadmin
MINIO_ROOT_PASSWORD=minioadmin
KAFKA_CLUSTER_ID=MkU3OEVBNTcwNTJENDM2Qg

Here is infra/init_minio.py:

from minio import Minio

client = Minio(
    'localhost:9000',
    access_key='minioadmin',
    secret_key='minioadmin',
    secure=False
)

buckets = ['bronze', 'silver', 'gold']
for bucket in buckets:
    if not client.bucket_exists(bucket):
        client.make_bucket(bucket)
        print(f'Created bucket: {bucket}')
    else:
        print(f'Bucket already exists: {bucket}')

Your infrastructure is fully operational. Kafka is ready to accept messages and MinIO is ready to store your data lake. Next up, you will build the Python producer that pulls live weather data from the API and feeds it into Kafka.

Build the API Producer

Your Docker infrastructure is running and your Kafka topics are ready to receive data. Now it's time to build the component that actually feeds live weather data into the pipeline.

The producer is the entry point of your streaming architecture. It calls the OpenWeatherMap API on a schedule, transforms the response into your event schema, and publishes each event to Kafka. If the API is down or rate-limiting you, the producer needs to handle that gracefully without crashing the entire pipeline.

In this step, get ready to:

• Build a WeatherAPIClient class with exponential backoff retry and error handling.
• Implement Kafka producer logic with idempotent delivery and JSON serialization.
• Create a configuration module and verify live messages flowing through the pipeline.

Build the WeatherAPIClient class

The API client is responsible for one thing: reliably fetching weather data from OpenWeatherMap. "Reliably" means handling the real-world problems that happen when calling external APIs.

HTTP 429 responses mean you've exceeded the rate limit. Timeouts mean the server didn't respond. Connection errors mean the network is down. Your client needs to handle all three using exponential backoff (wait 1s, then 2s, then 4s before retrying).

• Create the producer/weather_producer.py file and add the following code:

import json
import time
import signal
import logging
from datetime import datetime, timezone
from typing import Optional
import requests
from confluent_kafka import Producer
from producer.config import API_KEY, KAFKA_BOOTSTRAP_SERVERS, POLLING_INTERVAL, CITIES

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Base URL for the OpenWeatherMap Current Weather API
BASE_URL = 'https://api.openweathermap.org/data/2.5/weather'

class WeatherAPIClient:
    def __init__(self, api_key: str, max_retries: int = 3):
        self.api_key = api_key
        self.max_retries = max_retries

    def fetch_weather(self, city: str) -> Optional[dict]:
        # Retry loop with exponential backoff
        for attempt in range(self.max_retries):
            try:
                response = requests.get(
                    BASE_URL,
                    params={'q': city, 'units': 'metric', 'appid': self.api_key},
                    timeout=10
                )
                # Handle rate limiting with backoff
                if response.status_code == 429:
                    wait = 2 ** attempt
                    logger.warning(f'Rate limited. Waiting {wait}s.')
                    time.sleep(wait)
                    continue
                response.raise_for_status()
                return response.json()
            except requests.exceptions.ConnectionError as e:
                logger.error(f'Connection error for {city}: {e}')
                time.sleep(2 ** attempt)
            except requests.exceptions.Timeout:
                logger.error(f'Timeout for {city}')
                time.sleep(2 ** attempt)
        return None

What does this code do?

The imports at the top pull in everything the full producer needs. The from producer.config import ... line references a config file you'll create shortly in substep 3.

The WeatherAPIClient class wraps all API interaction logic. The fetch_weather method uses ?q={city}&units=metric&appid={key} to request weather in Celsius for a given city.

The retry loop uses exponential backoff: attempt 0 waits 2^0 = 1 second, attempt 1 waits 2^1 = 2 seconds, attempt 2 waits 2^2 = 4 seconds. This prevents hammering a struggling API and respects the 60 calls/minute free tier limit.

• Save producer/weather_producer.py.
• You won't be able to run this file yet because producer/config.py doesn't exist. That's expected.

Seeing a red underline on the config import?

Your editor may show an import error for from producer.config import ... because the file doesn't exist yet. Ignore this for now. You'll create config.py in the final substep of this step.

help me understand why my import is failing

Add Kafka producer logic and graceful shutdown

The API client fetches data, but you need a second piece: the logic that transforms raw API responses into your event schema and pushes them into Kafka. You also need a graceful shutdown handler so the producer finishes flushing messages before exiting.

• In producer/weather_producer.py, add this transform method inside the WeatherAPIClient class, right after the fetch_weather method:

    def transform(self, raw: dict, ingestion_time: str) -> dict:
        # Map raw API fields to our pipeline event schema
        return {
            'city': raw['name'],
            'temperature': raw['main']['temp'],
            'humidity': raw['main']['humidity'],
            'wind_speed': raw['wind']['speed'],
            'wind_deg': raw['wind'].get('deg', 0),
            'weather_condition': raw['weather'][0]['main'],
            'event_time': datetime.fromtimestamp(raw['dt'], tz=timezone.utc).isoformat(),
            'ingestion_time': ingestion_time
        }

What does the transform method do?

This method converts the raw OpenWeatherMap JSON (which has deeply nested fields like main.temp and weather[0].main) into the flat event schema you defined in docs/event_schema.md.

The event_time field comes from the API's dt unix timestamp (when the weather was measured). The ingestion_time records when your producer fetched the data. Tracking both is essential for understanding latency in streaming systems.

• Now add the shutdown handler and main() function below the class (outside of it, at module level):

shutdown = False

def handle_shutdown(signum, frame):
    global shutdown
    shutdown = True
    logger.info('Shutdown signal received.')

# Register signal handlers for graceful termination
signal.signal(signal.SIGINT, handle_shutdown)
signal.signal(signal.SIGTERM, handle_shutdown)

def main():
    client = WeatherAPIClient(API_KEY)
    # Idempotent producer prevents duplicate messages on retries
    producer = Producer({
        'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
        'enable.idempotence': True
    })
    while not shutdown:
        ingestion_time = datetime.now(timezone.utc).isoformat()
        for city in CITIES:
            raw = client.fetch_weather(city)
            if raw:
                event = client.transform(raw, ingestion_time)
                # Use city as message key for ordering guarantees per city
                producer.produce(
                    topic='raw_weather',
                    key=city,
                    value=json.dumps(event)
                )
                logger.info(f'Produced event for {city}')
        producer.flush()
        time.sleep(POLLING_INTERVAL)
    # Final flush on shutdown to ensure no messages are lost
    producer.flush()

if __name__ == '__main__':
    main()

What does this code do?

• The shutdown flag and handle_shutdown function let you press Ctrl+C to stop the producer cleanly. Without this, Kafka messages could be lost mid-flush.
• Setting enable.idempotence to True tells Kafka to deduplicate messages if the producer retries a send. This guarantees exactly-once delivery at the producer level.
• Using the city name as the message key ensures all events for the same city go to the same Kafka partition. This preserves ordering per city.
• The producer.flush() call after each polling loop forces all buffered messages to be sent before sleeping.

• Save producer/weather_producer.py.

Not sure where to place the code?

The transform method goes inside the WeatherAPIClient class (indented under it). The shutdown variable, handle_shutdown function, and main() function go outside the class at the top level of the file.

help me understand the structure of my weather_producer.py file

✔️ Awesome, I've got everything!

Great. Make sure you've saved producer/weather_producer.py before moving on.

ⓧ I'd like to double check the full code

import json
import time
import signal
import logging
from datetime import datetime, timezone
from typing import Optional
import requests
from confluent_kafka import Producer
from producer.config import API_KEY, KAFKA_BOOTSTRAP_SERVERS, POLLING_INTERVAL, CITIES

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

BASE_URL = 'https://api.openweathermap.org/data/2.5/weather'

class WeatherAPIClient:
    def __init__(self, api_key: str, max_retries: int = 3):
        self.api_key = api_key
        self.max_retries = max_retries

    def fetch_weather(self, city: str) -> Optional[dict]:
        for attempt in range(self.max_retries):
            try:
                response = requests.get(
                    BASE_URL,
                    params={'q': city, 'units': 'metric', 'appid': self.api_key},
                    timeout=10
                )
                if response.status_code == 429:
                    wait = 2 ** attempt
                    logger.warning(f'Rate limited. Waiting {wait}s.')
                    time.sleep(wait)
                    continue
                response.raise_for_status()
                return response.json()
            except requests.exceptions.ConnectionError as e:
                logger.error(f'Connection error for {city}: {e}')
                time.sleep(2 ** attempt)
            except requests.exceptions.Timeout:
                logger.error(f'Timeout for {city}')
                time.sleep(2 ** attempt)
        return None

    def transform(self, raw: dict, ingestion_time: str) -> dict:
        return {
            'city': raw['name'],
            'temperature': raw['main']['temp'],
            'humidity': raw['main']['humidity'],
            'wind_speed': raw['wind']['speed'],
            'wind_deg': raw['wind'].get('deg', 0),
            'weather_condition': raw['weather'][0]['main'],
            'event_time': datetime.fromtimestamp(raw['dt'], tz=timezone.utc).isoformat(),
            'ingestion_time': ingestion_time
        }

shutdown = False

def handle_shutdown(signum, frame):
    global shutdown
    shutdown = True
    logger.info('Shutdown signal received.')

signal.signal(signal.SIGINT, handle_shutdown)
signal.signal(signal.SIGTERM, handle_shutdown)

def main():
    client = WeatherAPIClient(API_KEY)
    producer = Producer({
        'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
        'enable.idempotence': True
    })
    while not shutdown:
        ingestion_time = datetime.now(timezone.utc).isoformat()
        for city in CITIES:
            raw = client.fetch_weather(city)
            if raw:
                event = client.transform(raw, ingestion_time)
                producer.produce(
                    topic='raw_weather',
                    key=city,
                    value=json.dumps(event)
                )
                logger.info(f'Produced event for {city}')
        producer.flush()
        time.sleep(POLLING_INTERVAL)
    producer.flush()

if __name__ == '__main__':
    main()

Create the configuration module and verify the pipeline

The producer imports its settings from a separate config.py module. This keeps secrets and tuneable values out of application logic, making the code easier to deploy across environments.

• Create producer/config.py with the following code:

import os
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

API_KEY = os.getenv('OPENWEATHER_API_KEY')
KAFKA_BOOTSTRAP_SERVERS = os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092')
POLLING_INTERVAL = int(os.getenv('POLLING_INTERVAL', 30))
CITIES = os.getenv('CITIES', 'London,Tokyo,New York,Sydney,Paris').split(',')

What does this config module do?

The load_dotenv() call reads your .env file and makes its values available through os.getenv(). Each variable has a sensible default so the producer still works if a value is missing from the file.

The CITIES variable splits the comma-separated string into a Python list. With the default 5 cities polled every 30 seconds, you use 10 API calls per minute. That's well within the free tier limit of 60 calls/minute.

• Save producer/config.py.
• Verify both files exist by running:

ls producer/

You should see config.py and weather_producer.py in the output.

• Start the producer by running this command:

python -m producer.weather_producer

You should see log lines like Produced event for London appearing every 30 seconds for each city. Let it run for at least one full polling cycle (30 seconds) before verifying.

• Open a second terminal and verify messages are in Kafka by running:

docker exec broker /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic raw_weather --from-beginning

You should see JSON messages streaming in the terminal, each containing fields like city, temperature, humidity, and event_time. Press Ctrl+C to stop the consumer when you've confirmed messages are flowing.

Not seeing any messages?

• Check that your Docker Compose stack is still running. Run docker ps and confirm the broker container is up.
• Make sure your .env file has a valid OPENWEATHER_API_KEY value (not the placeholder your-key-here).
• Look at the producer terminal output for error messages. Connection errors to the API suggest a network issue. Connection errors to Kafka suggest the broker isn't reachable on localhost:9092.

help me debug why my weather producer isn't sending messages to Kafka

• Once you've confirmed messages are flowing, switch back to your producer terminal and press Ctrl+C to stop it.

Why idempotent producing matters

Without idempotency, a network hiccup between the producer and Kafka could cause the same message to be sent twice. The consumer downstream would see duplicate weather readings.

With enable.idempotence=True, Kafka assigns a sequence number to each message from this producer. If a duplicate arrives, Kafka silently discards it. This gives you exactly-once semantics at the producer level with zero extra code.

✔️ Awesome, I've got everything!

Great. Make sure you've saved producer/config.py before moving on.

ⓧ I'd like to double check the full code

import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv('OPENWEATHER_API_KEY')
KAFKA_BOOTSTRAP_SERVERS = os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'localhost:9092')
POLLING_INTERVAL = int(os.getenv('POLLING_INTERVAL', 30))
CITIES = os.getenv('CITIES', 'London,Tokyo,New York,Sydney,Paris').split(',')

Your producer is live and pumping weather data into Kafka. Next up, you'll build the Spark Structured Streaming job that reads these messages, applies windowed aggregations, and detects anomalies in real time.

Build the Spark Structured Streaming Job

Your producer is generating a steady stream of weather events into Kafka. Raw messages are landing in the raw_weather topic, but they are just sitting there unprocessed.

In this step, you will build a Spark Structured Streaming job that continuously reads those raw events, parses them into a typed schema, computes windowed aggregations (average and max temperature per city), and detects anomalies. The results flow back into Kafka topics for downstream consumers.

In this step, get ready to:

• Create a Spark streaming application that reads from the Kafka raw_weather topic.
• Implement windowed aggregations with watermarking for late data handling.
• Add anomaly detection and submit the job with spark-submit.

Create the streaming app with schema and Kafka source

Spark Structured Streaming treats a live data stream as a continuously appended table. You write batch-style queries against this unbounded table, and Spark executes them incrementally as new data arrives.

• Create a new file called streaming/streaming_app.py inside your weather-pipeline folder.
• Add the imports and event schema definition by pasting this code:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, avg, max as spark_max, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType

schema = StructType([
    StructField('city', StringType()),
    StructField('temperature', FloatType()),
    StructField('humidity', IntegerType()),
    StructField('wind_speed', FloatType()),
    StructField('wind_deg', IntegerType()),
    StructField('weather_condition', StringType()),
    StructField('event_time', StringType()),
    StructField('ingestion_time', StringType())
])

What does this code do?

The imports bring in Spark's streaming, JSON parsing, windowing, and aggregation functions. The StructType schema explicitly defines every field and its type, matching the event schema from your producer.

Defining an explicit schema is important for streaming. Spark cannot infer schemas from a live Kafka stream, so you must tell it exactly what shape the incoming JSON has.

• Below the schema, add the SparkSession initialization, Kafka source, and JSON parsing logic:

spark = SparkSession.builder.appName('WeatherStreaming').getOrCreate()

raw_stream = spark.readStream.format('kafka').option('kafka.bootstrap.servers', 'localhost:9092').option('subscribe', 'raw_weather').option('startingOffsets', 'latest').load()

parsed = raw_stream.select(from_json(col('value').cast('string'), schema).alias('data')).select('data.*').withColumn('event_time', to_timestamp(col('event_time')))

What does this code do?

SparkSession is the entry point for all Spark functionality. The readStream call connects to Kafka at localhost:9092 and subscribes to the raw_weather topic, starting from the latest available offset.

Kafka delivers messages as binary key and value columns. The parsing step casts the value to a string, applies from_json with your schema to extract typed columns, and converts event_time from an ISO8601 string to a Spark TimestampType for windowed operations.

• Save streaming/streaming_app.py.

You won't see output yet. The next chunk adds the aggregation logic that produces results.

Seeing an import error for pyspark?

Make sure your virtual environment is activated. Run source .venv/bin/activate and verify with pip show pyspark. You should see version 4.1.2 listed.

help me debug a pyspark import error in my streaming app

Implement windowed aggregations

Windowed aggregations group events by time intervals and compute metrics across each window. A watermark tells Spark how long to wait for late-arriving data before finalizing a window.

• In streaming/streaming_app.py, add the windowed aggregation logic below your parsed variable:

windowed_agg = parsed.withWatermark('event_time', '2 minutes').groupBy(window(col('event_time'), '5 minutes', '1 minute'), col('city')).agg(avg('temperature').alias('avg_temperature'), spark_max('temperature').alias('max_temperature'), avg('humidity').alias('avg_humidity'), spark_max('wind_speed').alias('max_wind_speed'))

What does this code do?

The withWatermark('event_time', '2 minutes') tells Spark to wait up to 2 minutes for late data before discarding state for a window. This is calculated as the maximum event time seen minus 2 minutes.

The groupBy(window(..., '5 minutes', '1 minute'), col('city')) creates 5-minute sliding windows that advance every 1 minute, grouped by city. Each window computes average temperature, max temperature, average humidity, and max wind speed.

Why a 2-minute watermark?

Your producer polls every 30 seconds. Network jitter could add 10-20 seconds of delay. A 2-minute watermark gives enough buffer for late-arriving events without keeping excessive state in memory.

• Below the windowed_agg variable, add the write stream that sends aggregates to the aggregates Kafka topic:

agg_query = windowed_agg.selectExpr('city AS key', 'to_json(struct(*)) AS value').writeStream.format('kafka').option('kafka.bootstrap.servers', 'localhost:9092').option('topic', 'aggregates').option('checkpointLocation', '/tmp/spark-checkpoints/weather/aggregates').outputMode('update').start()

What does this code do?

The selectExpr reformats the aggregation result into Kafka's required key and value columns. The city becomes the key, and all columns are serialized to JSON as the value.

The outputMode('update') means Spark emits only rows that changed since the last trigger. This is the correct mode for windowed aggregations with watermarking.

The checkpointLocation stores progress state so the job can recover from failures without reprocessing everything.

• Save streaming/streaming_app.py.

You won't see output yet. The next chunk adds anomaly detection and starts the blocking wait.

Confused about update vs append mode?

Use update mode when you have aggregations and want intermediate results emitted as they change. Use append mode for non-aggregation streams or when you only want finalized results after the watermark passes.

help me understand when to use update vs append output mode in Spark Structured Streaming

Add anomaly detection and run the job

Anomaly detection flags events with unusual readings. You will filter for dangerously low humidity (below 20%) and write those flagged events to a separate Kafka topic for alerting.

• In streaming/streaming_app.py, add the anomaly detection stream below the agg_query line:

anomalies = parsed.withWatermark('event_time', '2 minutes').filter(col('humidity') < 20)

anom_query = anomalies.selectExpr('city AS key', 'to_json(struct(*)) AS value').writeStream.format('kafka').option('kafka.bootstrap.servers', 'localhost:9092').option('topic', 'clean_weather').option('checkpointLocation', '/tmp/spark-checkpoints/weather/anomalies').outputMode('append').start()

agg_query.awaitTermination()

What does this code do?

The anomalies stream filters for events where humidity drops below 20%. These are written to the clean_weather topic using append mode, since this is a non-aggregation filter that emits individual events.

Each stream gets its own checkpoint directory (/tmp/spark-checkpoints/weather/anomalies) so Spark can track progress independently.

The agg_query.awaitTermination() call blocks the main thread so the job runs continuously until you stop it with Ctrl+C.

• Save streaming/streaming_app.py.
• Make sure your producer is running in another terminal (so messages flow into raw_weather).
• Submit the streaming job by running this command:

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.1.2 streaming/streaming_app.py

What does this command do?

The --packages flag tells Spark to download the Kafka connector JAR (spark-sql-kafka-0-10_2.13:4.1.2) at runtime. This provides the Kafka source and sink integration for Structured Streaming.

The first run will take a minute or two as Spark downloads the dependency. Subsequent runs use the cached JAR.

• Watch the terminal output. After the initial setup logs, you should see Spark processing micro-batches every few seconds with no errors.
• Verify aggregates are flowing by checking the aggregates topic in a new terminal:

docker exec broker /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic aggregates --from-beginning

You should see JSON messages containing avg_temperature, max_temperature, avg_humidity, and max_wind_speed for each city and time window.

No messages appearing in the aggregates topic?

Windowed aggregates with update mode only emit after Spark processes at least one micro-batch with data. Make sure your producer is running and has been producing for at least 30 seconds before checking.

If you see connection errors, confirm your Docker Compose stack is running with docker ps and that the broker container is healthy.

help me troubleshoot why my Spark streaming job is not producing output to the aggregates Kafka topic

✔️ Awesome, I've got everything!

Great. Double check you have saved streaming/streaming_app.py and that your streaming job is processing micro-batches successfully.

ⓧ I'd like to double check the full code

Here is the complete streaming/streaming_app.py file:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, avg, max as spark_max, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType

schema = StructType([
    StructField('city', StringType()),
    StructField('temperature', FloatType()),
    StructField('humidity', IntegerType()),
    StructField('wind_speed', FloatType()),
    StructField('wind_deg', IntegerType()),
    StructField('weather_condition', StringType()),
    StructField('event_time', StringType()),
    StructField('ingestion_time', StringType())
])

spark = SparkSession.builder.appName('WeatherStreaming').getOrCreate()

raw_stream = spark.readStream.format('kafka').option('kafka.bootstrap.servers', 'localhost:9092').option('subscribe', 'raw_weather').option('startingOffsets', 'latest').load()

parsed = raw_stream.select(from_json(col('value').cast('string'), schema).alias('data')).select('data.*').withColumn('event_time', to_timestamp(col('event_time')))

windowed_agg = parsed.withWatermark('event_time', '2 minutes').groupBy(window(col('event_time'), '5 minutes', '1 minute'), col('city')).agg(avg('temperature').alias('avg_temperature'), spark_max('temperature').alias('max_temperature'), avg('humidity').alias('avg_humidity'), spark_max('wind_speed').alias('max_wind_speed'))

agg_query = windowed_agg.selectExpr('city AS key', 'to_json(struct(*)) AS value').writeStream.format('kafka').option('kafka.bootstrap.servers', 'localhost:9092').option('topic', 'aggregates').option('checkpointLocation', '/tmp/spark-checkpoints/weather/aggregates').outputMode('update').start()

anomalies = parsed.withWatermark('event_time', '2 minutes').filter(col('humidity') < 20)

anom_query = anomalies.selectExpr('city AS key', 'to_json(struct(*)) AS value').writeStream.format('kafka').option('kafka.bootstrap.servers', 'localhost:9092').option('topic', 'clean_weather').option('checkpointLocation', '/tmp/spark-checkpoints/weather/anomalies').outputMode('append').start()

agg_query.awaitTermination()

Your streaming pipeline is now processing weather data in real time. Next up, you will build the medallion architecture on MinIO to persist raw, cleaned, and aggregated data as Parquet files.

Implement the Medallion Architecture on MinIO

Your streaming pipeline processes weather events in real time through Kafka and Spark. Now you need a way to persist that data at different stages of refinement so it can be queried historically, audited, and reprocessed if needed.

The medallion architecture solves this by organizing data into three layers: bronze (raw), silver (cleaned), and gold (aggregated). In this step, you will write three Spark jobs that land data into the MinIO buckets you created earlier, building a complete data lake on S3-compatible object storage.

In this step, get ready to:

• Build a bronze writer that stores raw Kafka events as Parquet in MinIO.
• Build a silver writer that cleans, validates, and deduplicates data from the bronze layer.
• Build a gold writer that materializes windowed aggregations into MinIO for batch queries.

Create the bronze layer writer

The bronze layer captures raw events exactly as they arrive from Kafka. No transformation, no filtering. This preserves the original data so you can always replay or reprocess from the source of truth.

• Create the bronze writer by adding the following code to a new file at streaming/bronze_writer.py:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, to_date, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType

schema = StructType([
    StructField('city', StringType()),
    StructField('temperature', FloatType()),
    StructField('humidity', IntegerType()),
    StructField('wind_speed', FloatType()),
    StructField('wind_deg', IntegerType()),
    StructField('weather_condition', StringType()),
    StructField('event_time', StringType()),
    StructField('ingestion_time', StringType())
])

# Configure SparkSession with S3A credentials for MinIO
spark = SparkSession.builder.appName('BronzeWriter').config('spark.hadoop.fs.s3a.endpoint', 'http://localhost:9000').config('spark.hadoop.fs.s3a.access.key', 'minioadmin').config('spark.hadoop.fs.s3a.secret.key', 'minioadmin').config('spark.hadoop.fs.s3a.path.style.access', 'true').getOrCreate()

# Read raw events from the raw_weather Kafka topic
raw_stream = spark.readStream.format('kafka').option('kafka.bootstrap.servers', 'localhost:9092').option('subscribe', 'raw_weather').option('startingOffsets', 'earliest').load()

# Parse JSON and extract a date column for partitioning
parsed = raw_stream.select(from_json(col('value').cast('string'), schema).alias('data')).select('data.*').withColumn('event_time', to_timestamp(col('event_time'))).withColumn('date', to_date(col('event_time')))

# Write as Parquet to MinIO bronze bucket, partitioned by date and city
bronze_query = parsed.writeStream.format('parquet').option('path', 's3a://bronze/weather/').option('checkpointLocation', '/tmp/spark-checkpoints/bronze').partitionBy('date', 'city').outputMode('append').start()

bronze_query.awaitTermination()

What does this code do?

The SparkSession is configured with four S3A settings that tell Spark how to connect to MinIO instead of real AWS S3. The fs.s3a.endpoint points to your local MinIO. The fs.s3a.path.style.access setting ensures Spark uses path-style URLs (required for MinIO) instead of virtual-hosted-style.

The streaming query reads from raw_weather with startingOffsets set to earliest so it picks up all historical messages. It then writes Parquet files partitioned by date and city, making later queries efficient for time-range and city-specific lookups.

• Save streaming/bronze_writer.py.
• Make sure your producer is running so the raw_weather topic has messages. Then submit the bronze writer by running:

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.1.2,org.apache.hadoop:hadoop-aws:3.3.4 streaming/bronze_writer.py

• Let it run for about 30-60 seconds so data lands in MinIO.
• Open the MinIO console at http://localhost:9001 and navigate to the bronze bucket. You should see a weather/ directory containing date-partitioned and city-partitioned Parquet files.
• Press Ctrl+C in the terminal to stop the bronze writer.

Not seeing files in MinIO?

• Confirm your Docker Compose stack is still running. If MinIO or Kafka have stopped, restart with docker compose -f infra/docker-compose.yml up -d.
• Make sure the producer has run long enough for at least one batch of messages to appear in the raw_weather topic.
• Check the Spark logs for S3A connection errors. A common issue is MinIO not being accessible at localhost:9000.

Why isn't my bronze writer connecting to MinIO?

✔️ Awesome, I've got everything!

Great. Double check you have saved streaming/bronze_writer.py and confirmed Parquet files exist in the bronze bucket.

ⓧ I'd like to double check the full code

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, to_date, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType

schema = StructType([
    StructField('city', StringType()),
    StructField('temperature', FloatType()),
    StructField('humidity', IntegerType()),
    StructField('wind_speed', FloatType()),
    StructField('wind_deg', IntegerType()),
    StructField('weather_condition', StringType()),
    StructField('event_time', StringType()),
    StructField('ingestion_time', StringType())
])

spark = SparkSession.builder.appName('BronzeWriter').config('spark.hadoop.fs.s3a.endpoint', 'http://localhost:9000').config('spark.hadoop.fs.s3a.access.key', 'minioadmin').config('spark.hadoop.fs.s3a.secret.key', 'minioadmin').config('spark.hadoop.fs.s3a.path.style.access', 'true').getOrCreate()

raw_stream = spark.readStream.format('kafka').option('kafka.bootstrap.servers', 'localhost:9092').option('subscribe', 'raw_weather').option('startingOffsets', 'earliest').load()

parsed = raw_stream.select(from_json(col('value').cast('string'), schema).alias('data')).select('data.*').withColumn('event_time', to_timestamp(col('event_time'))).withColumn('date', to_date(col('event_time')))

bronze_query = parsed.writeStream.format('parquet').option('path', 's3a://bronze/weather/').option('checkpointLocation', '/tmp/spark-checkpoints/bronze').partitionBy('date', 'city').outputMode('append').start()

bronze_query.awaitTermination()

Create the silver layer writer

The silver layer reads from the bronze Parquet files and applies data quality rules. It removes nulls, validates value ranges, deduplicates by composite key, and stamps a processed_at timestamp. This gives downstream consumers a clean, trustworthy dataset.

• Create the silver writer by adding the following code to a new file at streaming/silver_writer.py:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_timestamp

# Configure SparkSession with S3A credentials for MinIO
spark = SparkSession.builder.appName('SilverWriter').config('spark.hadoop.fs.s3a.endpoint', 'http://localhost:9000').config('spark.hadoop.fs.s3a.access.key', 'minioadmin').config('spark.hadoop.fs.s3a.secret.key', 'minioadmin').config('spark.hadoop.fs.s3a.path.style.access', 'true').getOrCreate()

# Read raw Parquet data from the bronze layer
bronze_df = spark.read.parquet('s3a://bronze/weather/')

# Clean: drop nulls, validate ranges, deduplicate, add processing timestamp
silver_df = bronze_df.dropna().filter((col('temperature') >= -60) & (col('temperature') <= 60)).filter((col('humidity') >= 0) & (col('humidity') <= 100)).dropDuplicates(['city', 'event_time']).withColumn('processed_at', current_timestamp())

# Overwrite silver layer with cleaned data
silver_df.write.mode('overwrite').parquet('s3a://silver/weather/')

What does this code do?

• dropna() removes any rows with null values in critical fields.
• The temperature filter keeps only readings between -60 and 60 degrees Celsius. Anything outside that range is sensor noise or corrupt data.
• The humidity filter keeps values between 0 and 100 percent.
• dropDuplicates(['city', 'event_time']) ensures idempotency. If the same city reports the same timestamp twice (from a producer retry), only one record survives.
• current_timestamp() stamps when the cleaning happened, useful for auditing data freshness.

Why is the silver writer a batch job?

Unlike the bronze and gold writers which use Spark Structured Streaming, the silver writer reads Parquet files that already exist on disk. This is a deliberate design choice. The bronze layer accumulates data over time, and the silver job processes all of it in one pass. You would typically schedule this as a periodic batch job (every few minutes or hours) to refresh the cleaned dataset.

• Save streaming/silver_writer.py.
• Run the silver writer by submitting it:

spark-submit --packages org.apache.hadoop:hadoop-aws:3.3.4 streaming/silver_writer.py

• Return to the MinIO console at http://localhost:9001 and navigate to the silver bucket. You should see a weather/ directory containing Parquet files.

Seeing a path not found error?

The silver writer reads from s3a://bronze/weather/. If you did not run the bronze writer long enough to produce output files, this path will not exist yet. Go back and re-run the bronze writer for at least 30 seconds before attempting the silver job.

Help me debug my silver writer failing to read from the bronze layer.

✔️ Awesome, I've got everything!

Great. Double check you have saved streaming/silver_writer.py and confirmed Parquet files exist in the silver bucket.

ⓧ I'd like to double check the full code

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_timestamp

spark = SparkSession.builder.appName('SilverWriter').config('spark.hadoop.fs.s3a.endpoint', 'http://localhost:9000').config('spark.hadoop.fs.s3a.access.key', 'minioadmin').config('spark.hadoop.fs.s3a.secret.key', 'minioadmin').config('spark.hadoop.fs.s3a.path.style.access', 'true').getOrCreate()

bronze_df = spark.read.parquet('s3a://bronze/weather/')

silver_df = bronze_df.dropna().filter((col('temperature') >= -60) & (col('temperature') <= 60)).filter((col('humidity') >= 0) & (col('humidity') <= 100)).dropDuplicates(['city', 'event_time']).withColumn('processed_at', current_timestamp())

silver_df.write.mode('overwrite').parquet('s3a://silver/weather/')

Create the gold layer writer

The gold layer contains pre-computed business metrics ready for dashboard consumption. Your streaming_app.py already sends windowed aggregations to the aggregates Kafka topic. The gold writer performs the same windowed computation but materializes the results as Parquet on MinIO for historical batch queries.

• Create the gold writer by adding the following code to a new file at streaming/gold_writer.py:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, avg, max as spark_max, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType

schema = StructType([
    StructField('city', StringType()),
    StructField('temperature', FloatType()),
    StructField('humidity', IntegerType()),
    StructField('wind_speed', FloatType()),
    StructField('wind_deg', IntegerType()),
    StructField('weather_condition', StringType()),
    StructField('event_time', StringType()),
    StructField('ingestion_time', StringType())
])

# Configure SparkSession with S3A credentials for MinIO
spark = SparkSession.builder.appName('GoldWriter').config('spark.hadoop.fs.s3a.endpoint', 'http://localhost:9000').config('spark.hadoop.fs.s3a.access.key', 'minioadmin').config('spark.hadoop.fs.s3a.secret.key', 'minioadmin').config('spark.hadoop.fs.s3a.path.style.access', 'true').getOrCreate()

# Read raw events from Kafka
raw_stream = spark.readStream.format('kafka').option('kafka.bootstrap.servers', 'localhost:9092').option('subscribe', 'raw_weather').option('startingOffsets', 'latest').load()

# Parse JSON values into structured columns
parsed = raw_stream.select(from_json(col('value').cast('string'), schema).alias('data')).select('data.*').withColumn('event_time', to_timestamp(col('event_time')))

# Compute windowed aggregations per city
windowed_agg = parsed.withWatermark('event_time', '2 minutes').groupBy(window(col('event_time'), '5 minutes', '1 minute'), col('city')).agg(avg('temperature').alias('avg_temperature'), spark_max('temperature').alias('max_temperature'), avg('humidity').alias('avg_humidity'), spark_max('wind_speed').alias('max_wind_speed'))

# Write aggregates as Parquet to the gold bucket
gold_query = windowed_agg.writeStream.format('parquet').option('path', 's3a://gold/weather_aggregates/').option('checkpointLocation', '/tmp/spark-checkpoints/gold').outputMode('append').start()

gold_query.awaitTermination()

What does this code do?

The gold writer mirrors the windowing logic from streaming_app.py but writes results to MinIO instead of a Kafka topic. It computes 5-minute sliding windows (advancing every 1 minute) grouped by city, producing average temperature, max temperature, average humidity, and max wind speed per window.

The watermark of 2 minutes handles late-arriving events. The append output mode means each completed window is written once and never updated. This makes gold-layer data immutable and safe for downstream batch consumers.

• Save streaming/gold_writer.py.
• Make sure your producer is still running, then submit the gold writer:

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.1.2,org.apache.hadoop:hadoop-aws:3.3.4 streaming/gold_writer.py

• Let it run for 2-3 minutes so at least one 5-minute window completes and flushes to MinIO.
• Return to the MinIO console and navigate to the gold bucket. You should see a weather_aggregates/ directory containing Parquet files.
• Press Ctrl+C to stop the gold writer.

Gold bucket is empty after 2 minutes?

Windowed aggregations only emit results when a window closes. With a 5-minute window and 2-minute watermark, data appears after roughly 7 minutes of wall-clock time if you started with startingOffsets set to latest. Let the producer and gold writer run longer, or restart the producer so fresh events accumulate.

Help me understand why my gold writer is not producing Parquet output.

✔️ Awesome, I've got everything!

Great. Double check you have saved streaming/gold_writer.py and confirmed Parquet files exist in the gold bucket.

ⓧ I'd like to double check the full code

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, avg, max as spark_max, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType

schema = StructType([
    StructField('city', StringType()),
    StructField('temperature', FloatType()),
    StructField('humidity', IntegerType()),
    StructField('wind_speed', FloatType()),
    StructField('wind_deg', IntegerType()),
    StructField('weather_condition', StringType()),
    StructField('event_time', StringType()),
    StructField('ingestion_time', StringType())
])

spark = SparkSession.builder.appName('GoldWriter').config('spark.hadoop.fs.s3a.endpoint', 'http://localhost:9000').config('spark.hadoop.fs.s3a.access.key', 'minioadmin').config('spark.hadoop.fs.s3a.secret.key', 'minioadmin').config('spark.hadoop.fs.s3a.path.style.access', 'true').getOrCreate()

raw_stream = spark.readStream.format('kafka').option('kafka.bootstrap.servers', 'localhost:9092').option('subscribe', 'raw_weather').option('startingOffsets', 'latest').load()

parsed = raw_stream.select(from_json(col('value').cast('string'), schema).alias('data')).select('data.*').withColumn('event_time', to_timestamp(col('event_time')))

windowed_agg = parsed.withWatermark('event_time', '2 minutes').groupBy(window(col('event_time'), '5 minutes', '1 minute'), col('city')).agg(avg('temperature').alias('avg_temperature'), spark_max('temperature').alias('max_temperature'), avg('humidity').alias('avg_humidity'), spark_max('wind_speed').alias('max_wind_speed'))

gold_query = windowed_agg.writeStream.format('parquet').option('path', 's3a://gold/weather_aggregates/').option('checkpointLocation', '/tmp/spark-checkpoints/gold').outputMode('append').start()

gold_query.awaitTermination()

Your medallion architecture is now fully operational. Raw events land in bronze, cleaned data lives in silver, and pre-aggregated metrics sit in gold. Next up, you will build a real-time Streamlit dashboard that reads from these layers to display live weather KPIs.

Build the Streamlit Dashboard

Your pipeline is now producing real-time windowed aggregations into the aggregates Kafka topic. The data is flowing, but nobody can see it.

In this step, you will build a live Streamlit dashboard that consumes aggregated weather data from Kafka and displays it as interactive charts and KPI metrics that auto-refresh every few seconds.

In this step, get ready to:

• Create a Streamlit app with a Kafka consumer that fetches aggregate data.
• Build KPI metrics, a time-series chart, and a bar chart visualization layer.
• Add auto-refresh and sidebar controls for a real-time monitoring experience.

Set up the dashboard file and data fetcher

The dashboard needs to pull data directly from the aggregates Kafka topic where your Spark streaming job writes windowed results. You will use the confluent-kafka Consumer to poll messages and store them in Streamlit's session state so data persists across page reruns.

• Create the file dashboard/app.py by running:

touch dashboard/app.py

• Open dashboard/app.py in your editor and add the following imports, page configuration, and data fetcher function:

import streamlit as st
import json
from confluent_kafka import Consumer
import pandas as pd

# Configure the Streamlit page layout
st.set_page_config(page_title='Weather Monitor', layout='wide')

# Kafka connection for reading aggregates
KAFKA_BOOTSTRAP_SERVERS = 'localhost:9092'

def fetch_aggregates(n=100):
    # Create a consumer that reads from the aggregates topic
    consumer = Consumer({
        'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
        'group.id': 'dashboard-consumer',
        'auto.offset.reset': 'latest'
    })
    consumer.subscribe(['aggregates'])
    messages = []
    # Poll up to n messages with a short timeout
    for _ in range(n):
        msg = consumer.poll(timeout=0.5)
        if msg is None:
            break
        if msg.error():
            continue
        messages.append(json.loads(msg.value()))
    consumer.close()
    return messages

# Initialize session state to persist data across reruns
if 'data' not in st.session_state:
    st.session_state['data'] = []

What does this code do?

The fetch_aggregates function creates a short-lived Kafka consumer that polls the aggregates topic for up to 100 messages. Each message is a JSON-encoded windowed aggregation produced by your Spark streaming job.

The st.session_state dictionary keeps fetched data alive between Streamlit reruns. Without it, every page interaction would wipe your accumulated data.

• Save dashboard/app.py.
• Confirm the file exists by running:

ls dashboard/

You should see app.py listed in the output.

Don't see app.py?

Make sure you created the dashboard/ directory in a previous step. If not, run mkdir -p dashboard first.

help me create the dashboard directory and file

Build the visualization layer with auto-refresh

Now you will add sidebar controls and a fragment-decorated function that renders KPI metrics, a temperature line chart, and a wind speed bar chart. The @st.fragment decorator lets this function rerun on a timer without refreshing the entire page.

• In dashboard/app.py, add the following code below the session state initialization:

# Sidebar controls for filtering and refresh rate
st.sidebar.title('Controls')
city_filter = st.sidebar.multiselect('Filter Cities', ['London', 'Tokyo', 'New York', 'Sydney', 'Paris'], default=['London', 'Tokyo', 'New York', 'Sydney', 'Paris'])
refresh_rate = st.sidebar.slider('Refresh Rate (s)', 5, 60, 5)

@st.fragment(run_every=refresh_rate)
def live_dashboard():
    # Fetch new data from Kafka and append to session state
    new_data = fetch_aggregates()
    if new_data:
        st.session_state['data'].extend(new_data)
    df = pd.DataFrame(st.session_state['data'])
    if df.empty:
        st.info('Waiting for data...')
        return
    # Apply the city filter from the sidebar
    df = df[df['city'].isin(city_filter)]
    # KPI metrics showing latest temperature per city
    st.subheader('Current KPIs')
    cols = st.columns(len(city_filter))
    for i, city in enumerate(city_filter):
        city_df = df[df['city'] == city]
        if not city_df.empty:
            latest = city_df.iloc[-1]
            prev = city_df.iloc[-2] if len(city_df) > 1 else latest
            cols[i].metric(city, f"{latest['avg_temperature']:.1f} C", delta=f"{latest['avg_temperature'] - prev['avg_temperature']:.1f}")
    # Time-series line chart of average temperature per city
    st.subheader('Temperature Over Time')
    if 'window' in df.columns:
        pivot = df.pivot_table(index='window', columns='city', values='avg_temperature')
        st.line_chart(pivot)
    # Bar chart showing top 5 cities by maximum wind speed
    st.subheader('Top 5 Cities by Wind Speed')
    wind_df = df.groupby('city')['max_wind_speed'].max().nlargest(5).reset_index()
    st.bar_chart(wind_df.set_index('city'))
    # Pipeline health indicator in the sidebar
    last_ts = df['window'].max() if 'window' in df.columns else 'N/A'
    st.sidebar.write(f'Last update: {last_ts}')

# Call the fragment to render the dashboard
live_dashboard()

What does this code do?

The sidebar creates a city filter multiselect and a refresh rate slider. The slider value feeds directly into @st.fragment(run_every=refresh_rate) which controls how often the fragment re-executes.

Inside live_dashboard(), the function polls Kafka for new aggregates, builds a DataFrame, and renders three visualization sections. The KPI row uses st.metric with delta indicators to show temperature changes between the latest and previous readings.

The line chart pivots the DataFrame by city and window timestamp. The bar chart groups by city and picks the top 5 by maximum wind speed. The last update timestamp in the sidebar acts as a pipeline health indicator.

Why use st.fragment for auto-refresh?

Without fragments, Streamlit reruns the entire script on every interaction or timer tick. With @st.fragment(run_every=refresh_rate), only the dashboard function re-executes on the timer. The sidebar and page config stay untouched, which prevents flickering and improves performance.

• Save dashboard/app.py.

Seeing import errors?

Make sure your virtual environment is activated (source .venv/bin/activate) and that you installed streamlit, confluent-kafka, and pandas in Step 1.

help me fix import errors when running my Streamlit dashboard

✔️ Awesome, I've got everything!

Great. Double check you have saved dashboard/app.py before proceeding.

ⓧ I'd like to double check the full code

Here is the complete dashboard/app.py file:

import streamlit as st
import json
from confluent_kafka import Consumer
import pandas as pd

st.set_page_config(page_title='Weather Monitor', layout='wide')

KAFKA_BOOTSTRAP_SERVERS = 'localhost:9092'

def fetch_aggregates(n=100):
    consumer = Consumer({
        'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
        'group.id': 'dashboard-consumer',
        'auto.offset.reset': 'latest'
    })
    consumer.subscribe(['aggregates'])
    messages = []
    for _ in range(n):
        msg = consumer.poll(timeout=0.5)
        if msg is None:
            break
        if msg.error():
            continue
        messages.append(json.loads(msg.value()))
    consumer.close()
    return messages

if 'data' not in st.session_state:
    st.session_state['data'] = []

st.sidebar.title('Controls')
city_filter = st.sidebar.multiselect('Filter Cities', ['London', 'Tokyo', 'New York', 'Sydney', 'Paris'], default=['London', 'Tokyo', 'New York', 'Sydney', 'Paris'])
refresh_rate = st.sidebar.slider('Refresh Rate (s)', 5, 60, 5)

@st.fragment(run_every=refresh_rate)
def live_dashboard():
    new_data = fetch_aggregates()
    if new_data:
        st.session_state['data'].extend(new_data)
    df = pd.DataFrame(st.session_state['data'])
    if df.empty:
        st.info('Waiting for data...')
        return
    df = df[df['city'].isin(city_filter)]
    st.subheader('Current KPIs')
    cols = st.columns(len(city_filter))
    for i, city in enumerate(city_filter):
        city_df = df[df['city'] == city]
        if not city_df.empty:
            latest = city_df.iloc[-1]
            prev = city_df.iloc[-2] if len(city_df) > 1 else latest
            cols[i].metric(city, f"{latest['avg_temperature']:.1f} C", delta=f"{latest['avg_temperature'] - prev['avg_temperature']:.1f}")
    st.subheader('Temperature Over Time')
    if 'window' in df.columns:
        pivot = df.pivot_table(index='window', columns='city', values='avg_temperature')
        st.line_chart(pivot)
    st.subheader('Top 5 Cities by Wind Speed')
    wind_df = df.groupby('city')['max_wind_speed'].max().nlargest(5).reset_index()
    st.bar_chart(wind_df.set_index('city'))
    last_ts = df['window'].max() if 'window' in df.columns else 'N/A'
    st.sidebar.write(f'Last update: {last_ts}')

live_dashboard()

Launch the dashboard

With the file complete, you can start the Streamlit development server. Make sure your Docker Compose stack (Kafka and MinIO) is running and that the producer and streaming jobs are active so there is data flowing into the aggregates topic.

• Start the Streamlit dashboard by running:

streamlit run dashboard/app.py

• Open your browser to http://localhost:8501.
• You should see the Weather Monitor page with a sidebar showing city filters and a refresh rate slider.
• If data is flowing from your pipeline, KPI metric cards appear at the top, a temperature line chart renders below, and a wind speed bar chart shows at the bottom.
• If you see the info message "Waiting for data...", confirm your producer and streaming jobs are running in separate terminals.

Dashboard shows "Waiting for data..." indefinitely?

This means no messages are arriving on the aggregates topic. Check that your producer (python -m producer.weather_producer) is running and that streaming/streaming_app.py is active via spark-submit.

You can verify data exists on the topic by running the Kafka console consumer:

docker exec broker /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic aggregates --from-beginning

If no messages appear there either, the streaming job is not writing to the topic. Check the Spark logs for errors.

help me debug why my Streamlit dashboard is not receiving Kafka aggregates data

Your live monitoring layer is now complete. Next up, you will add data quality checks and automated tests to make this pipeline production-ready.

Add Data Quality Checks, Tests, and CI/CD

Your streaming pipeline is live, processing weather data from API to dashboard. But how do you know the data flowing through is actually correct?

Production pipelines need guardrails. In this step, you will add data quality checks that validate every event, automated tests that catch regressions before they reach production, and a CI pipeline that enforces standards on every push.

In this step, get ready to:

• Write data quality checks that validate schema conformance, value ranges, and freshness.
• Build unit and integration tests for the producer and streaming logic.
• Create a CI pipeline, Makefile, and README to tie everything together.

Create data quality checks

Data quality checks act as contracts for your pipeline. They validate that every event conforms to the expected schema, falls within realistic value ranges, and arrives within an acceptable time window.

• Create the file tests/test_quality.py inside your weather-pipeline folder by running:

touch tests/test_quality.py

• Open tests/test_quality.py and add the data quality test suite by pasting this code:

import pytest
from datetime import datetime, timezone, timedelta

# Required fields that every weather event must contain
REQUIRED_FIELDS = ['city', 'temperature', 'humidity', 'wind_speed', 'wind_deg', 'weather_condition', 'event_time', 'ingestion_time']

def test_schema_conformance():
    # Verify all required fields are present in a valid event
    event = {'city': 'London', 'temperature': 15.0, 'humidity': 70, 'wind_speed': 5.0, 'wind_deg': 180, 'weather_condition': 'Clear', 'event_time': datetime.now(timezone.utc).isoformat(), 'ingestion_time': datetime.now(timezone.utc).isoformat()}
    for field in REQUIRED_FIELDS:
        assert field in event

def test_temperature_range():
    # Temperature must be between -60 and 60 Celsius
    assert -60 <= 15.0 <= 60

def test_humidity_range():
    # Humidity must be between 0 and 100 percent
    assert 0 <= 70 <= 100

def test_wind_speed_non_negative():
    # Wind speed cannot be negative
    assert 5.0 >= 0

def test_freshness():
    # Events must not be older than 10 minutes
    event_time = datetime.now(timezone.utc)
    assert datetime.now(timezone.utc) - event_time < timedelta(minutes=10)

def test_no_null_critical_fields():
    # No critical field should be None
    event = {'city': 'London', 'temperature': 15.0, 'humidity': 70, 'wind_speed': 5.0, 'wind_deg': 180, 'weather_condition': 'Clear', 'event_time': datetime.now(timezone.utc).isoformat(), 'ingestion_time': datetime.now(timezone.utc).isoformat()}
    for field in REQUIRED_FIELDS:
        assert event[field] is not None

What does this code do?

Each test function validates one aspect of data quality:

• test_schema_conformance checks that all 8 required fields exist in every event.
• test_temperature_range and test_humidity_range enforce physically realistic bounds (-60 to 60 C, 0 to 100%).
• test_freshness ensures no stale data older than 10 minutes reaches the silver layer.
• test_no_null_critical_fields guards against incomplete records slipping through.

• Save tests/test_quality.py.
• Run the data quality tests by running:

pytest tests/test_quality.py -v

• You should see all 6 tests passing with green PASSED markers.

Tests failing?

• Make sure you are in the weather-pipeline root folder when running pytest.
• Confirm your virtual environment is active (you should see (.venv) in your terminal prompt).
• help me debug failing pytest data quality tests

✔️ Awesome, I've got everything!

Great. Double check you have saved tests/test_quality.py.

ⓧ I'd like to double check the full code

import pytest
from datetime import datetime, timezone, timedelta

REQUIRED_FIELDS = ['city', 'temperature', 'humidity', 'wind_speed', 'wind_deg', 'weather_condition', 'event_time', 'ingestion_time']

def test_schema_conformance():
    event = {'city': 'London', 'temperature': 15.0, 'humidity': 70, 'wind_speed': 5.0, 'wind_deg': 180, 'weather_condition': 'Clear', 'event_time': datetime.now(timezone.utc).isoformat(), 'ingestion_time': datetime.now(timezone.utc).isoformat()}
    for field in REQUIRED_FIELDS:
        assert field in event

def test_temperature_range():
    assert -60 <= 15.0 <= 60

def test_humidity_range():
    assert 0 <= 70 <= 100

def test_wind_speed_non_negative():
    assert 5.0 >= 0

def test_freshness():
    event_time = datetime.now(timezone.utc)
    assert datetime.now(timezone.utc) - event_time < timedelta(minutes=10)

def test_no_null_critical_fields():
    event = {'city': 'London', 'temperature': 15.0, 'humidity': 70, 'wind_speed': 5.0, 'wind_deg': 180, 'weather_condition': 'Clear', 'event_time': datetime.now(timezone.utc).isoformat(), 'ingestion_time': datetime.now(timezone.utc).isoformat()}
    for field in REQUIRED_FIELDS:
        assert event[field] is not None

Write unit and integration tests

Quality checks validate the data itself. Unit tests validate that your code logic works correctly in isolation, without needing the real API or a running Kafka cluster.

• Create the producer unit test file by running:

touch tests/test_producer.py

• Open tests/test_producer.py and paste the following test suite:

import pytest
from unittest.mock import patch, MagicMock
from producer.weather_producer import WeatherAPIClient
from datetime import datetime, timezone

# Sample API response matching OpenWeatherMap format
MOCK_RESPONSE = {'name': 'London', 'main': {'temp': 15.0, 'humidity': 70}, 'wind': {'speed': 5.0, 'deg': 180}, 'weather': [{'main': 'Clear'}], 'dt': 1700000000}

def test_transform():
    # Verify the transform method correctly maps raw API response to event schema
    client = WeatherAPIClient('fake-key')
    ingestion_time = datetime.now(timezone.utc).isoformat()
    result = client.transform(MOCK_RESPONSE, ingestion_time)
    assert result['city'] == 'London'
    assert result['temperature'] == 15.0
    assert result['humidity'] == 70

@patch('producer.weather_producer.requests.get')
def test_fetch_weather_success(mock_get):
    # Verify successful API call returns the response JSON
    mock_resp = MagicMock()
    mock_resp.status_code = 200
    mock_resp.json.return_value = MOCK_RESPONSE
    mock_get.return_value = mock_resp
    client = WeatherAPIClient('fake-key')
    result = client.fetch_weather('London')
    assert result == MOCK_RESPONSE

@patch('producer.weather_producer.requests.get')
def test_fetch_weather_retry_on_429(mock_get):
    # Verify the client retries after receiving a 429 rate limit response
    mock_resp_429 = MagicMock()
    mock_resp_429.status_code = 429
    mock_resp_ok = MagicMock()
    mock_resp_ok.status_code = 200
    mock_resp_ok.json.return_value = MOCK_RESPONSE
    mock_get.side_effect = [mock_resp_429, mock_resp_ok]
    client = WeatherAPIClient('fake-key', max_retries=2)
    result = client.fetch_weather('London')
    assert result == MOCK_RESPONSE

What does this code do?

• test_transform verifies that the raw OpenWeatherMap JSON is correctly mapped to your pipeline's event schema.
• test_fetch_weather_success uses unittest.mock to simulate a successful API call without hitting the real endpoint.
• test_fetch_weather_retry_on_429 confirms that the exponential backoff retry logic kicks in when the API returns a rate limit error.

• Save tests/test_producer.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved tests/test_producer.py.

ⓧ I'd like to double check the full code

import pytest
from unittest.mock import patch, MagicMock
from producer.weather_producer import WeatherAPIClient
from datetime import datetime, timezone

MOCK_RESPONSE = {'name': 'London', 'main': {'temp': 15.0, 'humidity': 70}, 'wind': {'speed': 5.0, 'deg': 180}, 'weather': [{'main': 'Clear'}], 'dt': 1700000000}

def test_transform():
    client = WeatherAPIClient('fake-key')
    ingestion_time = datetime.now(timezone.utc).isoformat()
    result = client.transform(MOCK_RESPONSE, ingestion_time)
    assert result['city'] == 'London'
    assert result['temperature'] == 15.0
    assert result['humidity'] == 70

@patch('producer.weather_producer.requests.get')
def test_fetch_weather_success(mock_get):
    mock_resp = MagicMock()
    mock_resp.status_code = 200
    mock_resp.json.return_value = MOCK_RESPONSE
    mock_get.return_value = mock_resp
    client = WeatherAPIClient('fake-key')
    result = client.fetch_weather('London')
    assert result == MOCK_RESPONSE

@patch('producer.weather_producer.requests.get')
def test_fetch_weather_retry_on_429(mock_get):
    mock_resp_429 = MagicMock()
    mock_resp_429.status_code = 429
    mock_resp_ok = MagicMock()
    mock_resp_ok.status_code = 200
    mock_resp_ok.json.return_value = MOCK_RESPONSE
    mock_get.side_effect = [mock_resp_429, mock_resp_ok]
    client = WeatherAPIClient('fake-key', max_retries=2)
    result = client.fetch_weather('London')
    assert result == MOCK_RESPONSE

Now you will test the Spark streaming logic. This test uses a local SparkSession with sample data to verify windowed aggregations produce correct results.

• Create the streaming test file by running:

touch tests/test_streaming.py

• Open tests/test_streaming.py and paste the following test:

import pytest
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType

@pytest.fixture(scope='module')
def spark():
    # Create a local SparkSession for testing (no cluster needed)
    return SparkSession.builder.master('local').appName('test').getOrCreate()

def test_windowed_aggregation(spark):
    # Verify that grouping by city and averaging temperature produces correct results
    data = [('London', 15.0, 70, 5.0, 180, 'Clear', '2024-01-01T12:00:00+00:00'), ('London', 17.0, 65, 6.0, 190, 'Clear', '2024-01-01T12:01:00+00:00')]
    schema = StructType([StructField('city', StringType()), StructField('temperature', FloatType()), StructField('humidity', IntegerType()), StructField('wind_speed', FloatType()), StructField('wind_deg', IntegerType()), StructField('weather_condition', StringType()), StructField('event_time', StringType())])
    df = spark.createDataFrame(data, schema).withColumn('event_time', to_timestamp(col('event_time')))
    result = df.groupBy(col('city')).agg(avg('temperature').alias('avg_temp')).collect()
    assert abs(result[0]['avg_temp'] - 16.0) < 0.01

What does this code do?

The spark fixture creates a local SparkSession that runs entirely in memory. The test creates a DataFrame with two London readings (15.0 and 17.0 degrees), groups by city, and checks the average equals 16.0. This verifies your aggregation logic without needing Kafka or MinIO running.

• Save tests/test_streaming.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved tests/test_streaming.py.

ⓧ I'd like to double check the full code

import pytest
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType

@pytest.fixture(scope='module')
def spark():
    return SparkSession.builder.master('local').appName('test').getOrCreate()

def test_windowed_aggregation(spark):
    data = [('London', 15.0, 70, 5.0, 180, 'Clear', '2024-01-01T12:00:00+00:00'), ('London', 17.0, 65, 6.0, 190, 'Clear', '2024-01-01T12:01:00+00:00')]
    schema = StructType([StructField('city', StringType()), StructField('temperature', FloatType()), StructField('humidity', IntegerType()), StructField('wind_speed', FloatType()), StructField('wind_deg', IntegerType()), StructField('weather_condition', StringType()), StructField('event_time', StringType())])
    df = spark.createDataFrame(data, schema).withColumn('event_time', to_timestamp(col('event_time')))
    result = df.groupBy(col('city')).agg(avg('temperature').alias('avg_temp')).collect()
    assert abs(result[0]['avg_temp'] - 16.0) < 0.01

Finally, you will write an integration test that verifies messages flow end-to-end through Kafka. This test requires your Docker infrastructure to be running.

• Create the integration test file by running:

touch tests/test_integration.py

• Open tests/test_integration.py and paste the following end-to-end test:

import pytest
import json
from confluent_kafka import Producer, Consumer

KAFKA_BOOTSTRAP = 'localhost:9092'

def test_producer_consumer_roundtrip():
    # Produce a test event and verify it can be consumed from Kafka
    producer = Producer({'bootstrap.servers': KAFKA_BOOTSTRAP})
    test_event = {'city': 'TestCity', 'temperature': 20.0}
    producer.produce('raw_weather', key='TestCity', value=json.dumps(test_event))
    producer.flush()
    consumer = Consumer({'bootstrap.servers': KAFKA_BOOTSTRAP, 'group.id': 'integration-test', 'auto.offset.reset': 'earliest'})
    consumer.subscribe(['raw_weather'])
    received = None
    for _ in range(20):
        msg = consumer.poll(timeout=1.0)
        if msg and not msg.error():
            received = json.loads(msg.value())
            if received.get('city') == 'TestCity':
                break
    consumer.close()
    assert received is not None
    assert received['city'] == 'TestCity'

What does this code do?

This test produces a message with a unique city name (TestCity) to the raw_weather topic, then creates a consumer that reads from the beginning until it finds that specific message. It verifies the full round-trip from producer to consumer through your real Kafka broker.

• Save tests/test_integration.py.
• Run the full test suite (unit tests and integration test) by running:

pytest tests/test_producer.py tests/test_streaming.py tests/test_quality.py -v

• You should see all tests passing. The streaming test may take a few seconds to initialize the local SparkSession.

Why not run the integration test here?

The integration test in test_integration.py requires a running Kafka broker. You can run it separately when your Docker stack is up. The CI pipeline will only run the unit tests since it does not spin up Docker services.

Seeing import errors?

• If you see ModuleNotFoundError: No module named 'producer', make sure you are running pytest from the weather-pipeline root folder (not from inside tests/).
• If Spark tests fail with a Java error, confirm that Java 11 or higher is installed on your system.
• help me debug my pytest import errors

✔️ Awesome, I've got everything!

Great. Double check you have saved tests/test_integration.py.

ⓧ I'd like to double check the full code

import pytest
import json
from confluent_kafka import Producer, Consumer

KAFKA_BOOTSTRAP = 'localhost:9092'

def test_producer_consumer_roundtrip():
    producer = Producer({'bootstrap.servers': KAFKA_BOOTSTRAP})
    test_event = {'city': 'TestCity', 'temperature': 20.0}
    producer.produce('raw_weather', key='TestCity', value=json.dumps(test_event))
    producer.flush()
    consumer = Consumer({'bootstrap.servers': KAFKA_BOOTSTRAP, 'group.id': 'integration-test', 'auto.offset.reset': 'earliest'})
    consumer.subscribe(['raw_weather'])
    received = None
    for _ in range(20):
        msg = consumer.poll(timeout=1.0)
        if msg and not msg.error():
            received = json.loads(msg.value())
            if received.get('city') == 'TestCity':
                break
    consumer.close()
    assert received is not None
    assert received['city'] == 'TestCity'

Create the CI pipeline, Makefile, and README

With tests in place, you need automation to enforce them. A CI/CD pipeline runs your linter and tests on every push, catching issues before they reach your main branch.

• Install the ruff linter by running:

pip install ruff

• Create the GitHub Actions workflow directory and file by running:

mkdir -p .github/workflows
touch .github/workflows/ci.yml

• Open .github/workflows/ci.yml and paste the following pipeline configuration:

name: CI
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - name: Install dependencies
        run: pip install pyspark==4.1.2 confluent-kafka==2.14.0 minio==7.2.20 streamlit==1.57.0 requests==2.32.3 pytest==9.0.3 python-dotenv==1.1.0 ruff
      - name: Lint
        run: ruff check .
      - name: Unit Tests
        run: pytest tests/test_producer.py tests/test_streaming.py tests/test_quality.py -v

What does this pipeline do?

• Triggers on every push and pull request to any branch.
• Sets up Python 3.12 and installs all project dependencies.
• Runs ruff check . to lint for style and syntax issues.
• Runs the unit tests (excluding the integration test which needs Docker).

• Save .github/workflows/ci.yml.

✔️ Awesome, I've got everything!

Great. Double check you have saved .github/workflows/ci.yml.

ⓧ I'd like to double check the full code

name: CI
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - name: Install dependencies
        run: pip install pyspark==4.1.2 confluent-kafka==2.14.0 minio==7.2.20 streamlit==1.57.0 requests==2.32.3 pytest==9.0.3 python-dotenv==1.1.0 ruff
      - name: Lint
        run: ruff check .
      - name: Unit Tests
        run: pytest tests/test_producer.py tests/test_streaming.py tests/test_quality.py -v

Next, you will create a Makefile that provides shortcuts for every command in your workflow. This lets anyone on the team run the pipeline with a single command.

• Create the Makefile in the project root by running:

touch Makefile

• Open Makefile and paste the following targets:

.PHONY: lint test up produce stream dashboard

lint:
	ruff check .

test:
	pytest tests/ -v

up:
	docker compose -f infra/docker-compose.yml up -d

produce:
	python -m producer.weather_producer

stream:
	spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.1.2 streaming/streaming_app.py

dashboard:
	streamlit run dashboard/app.py

What does each target do?

• make lint runs the ruff linter across your codebase.
• make test runs the full test suite with verbose output.
• make up starts Kafka and MinIO in Docker.
• make produce, make stream, and make dashboard start each pipeline component.

Tabs vs spaces in Makefiles

Makefiles require actual tab characters (not spaces) for indentation under each target. If you see an error like "missing separator", your editor may have converted tabs to spaces. Check your editor settings to ensure tabs are preserved in the Makefile.

• Save Makefile.
• Verify the Makefile works by running:

make lint

• You should see ruff output. If it reports linting issues, that is expected for now. The important thing is that the command runs successfully.

Seeing "missing separator" error?

• This means your Makefile uses spaces instead of tabs for indentation. Open the file and replace the leading spaces on each command line with a single tab character.
• help me fix my Makefile tabs

✔️ Awesome, I've got everything!

Great. Double check you have saved your Makefile.

ⓧ I'd like to double check the full code

.PHONY: lint test up produce stream dashboard

lint:
	ruff check .

test:
	pytest tests/ -v

up:
	docker compose -f infra/docker-compose.yml up -d

produce:
	python -m producer.weather_producer

stream:
	spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.1.2 streaming/streaming_app.py

dashboard:
	streamlit run dashboard/app.py

The last piece is a README.md that documents the architecture, setup instructions, and design decisions for anyone who encounters this project.

• Create the README file by running:

touch README.md

• Open README.md and paste the following documentation:

# Real-Time Weather Data Pipeline

## Architecture
OpenWeatherMap API -> Producer -> Kafka (raw_weather) -> Spark Streaming -> MinIO (bronze/silver/gold) -> Streamlit Dashboard

## Setup
1. Install Docker
2. Copy .env.example to .env and add your OpenWeatherMap API key
3. Run make up to start infrastructure
4. Run python infra/init_minio.py to create buckets
5. Run make produce to start the producer
6. Run make stream to start Spark streaming
7. Run make dashboard to start the dashboard

## Design Decisions
- KRaft mode Kafka (no ZooKeeper) for simplicity
- Medallion architecture for data quality progression
- Idempotent Kafka producer for exactly-once semantics
- Watermarking for late data handling

Why document design decisions?

A good README does more than list setup steps. The Design Decisions section explains the reasoning behind your architecture choices. When a future developer asks "why KRaft instead of ZooKeeper?", the answer is right there in the repo.

• Save README.md.
• Confirm all new files exist by running:

ls tests/ .github/workflows/ Makefile README.md

• You should see test_quality.py, test_producer.py, test_streaming.py, and test_integration.py in the tests output, ci.yml in the workflows output, plus your Makefile and README.md.

✔️ Awesome, I've got everything!

Great. Double check you have saved README.md.

ⓧ I'd like to double check the full code

# Real-Time Weather Data Pipeline

## Architecture
OpenWeatherMap API -> Producer -> Kafka (raw_weather) -> Spark Streaming -> MinIO (bronze/silver/gold) -> Streamlit Dashboard

## Setup
1. Install Docker
2. Copy .env.example to .env and add your OpenWeatherMap API key
3. Run make up to start infrastructure
4. Run python infra/init_minio.py to create buckets
5. Run make produce to start the producer
6. Run make stream to start Spark streaming
7. Run make dashboard to start the dashboard

## Design Decisions
- KRaft mode Kafka (no ZooKeeper) for simplicity
- Medallion architecture for data quality progression
- Idempotent Kafka producer for exactly-once semantics
- Watermarking for late data handling

Your pipeline is now battle-tested. You have data quality gates, automated tests, a CI pipeline to catch regressions, and a Makefile that makes the entire system operable with single commands. This is exactly how production streaming systems maintain reliability as they evolve.

Secret mission

Add Production Observability

Your pipeline processes data reliably, but how do you know when something goes wrong in production? In this secret mission, you'll add Prometheus metrics to your producer and Spark jobs, wire up a Grafana dashboard for live visualization, and configure alerting rules that fire when processing lag or error rates breach your thresholds.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally in Docker containers and a Python virtual environment. There are no ongoing cloud costs.

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

Resources you used:

• Docker containers: broker, minio, prometheus, grafana, kafka-init.
• Docker volume: minio_data (stores all MinIO object data including bronze, silver, and gold layer Parquet files).
• Spark checkpoints at /tmp/spark-checkpoints/weather.
• Python virtual environment (.venv) and project directory (weather-pipeline).
• OpenWeatherMap free-tier API key.

Keep everything running

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

• Leave Docker Compose running. Data continues to accumulate in MinIO volumes as the producer polls the API.
• Your OpenWeatherMap API key remains active on the free tier (no charges).
• The Python virtual environment (.venv) stays on disk inside your project folder.
• Prometheus and Grafana dashboards remain accessible at http://localhost:9090 and http://localhost:3000.

Pause - I'll come back to this later

Shut down running containers to free up CPU and memory. Your data and configuration are preserved.

• Stop all containers (preserves volumes and data) by running:

docker compose -f infra/docker-compose.yml stop

• When you are ready to resume, start everything back up by running:

docker compose -f infra/docker-compose.yml start

• Your MinIO data, Kafka topics, Prometheus metrics history, and Grafana dashboards remain intact.
• The Spark checkpoint directory at /tmp/spark-checkpoints/weather persists between reboots on most Linux systems, but may be cleared on restart. If cleared, Spark reprocesses from the latest Kafka offset.

Delete - I don't want to use this again

Remove all project containers, volumes, files, and local state. Nothing remains on your machine after these steps.

• Stop and remove all containers and volumes (this destroys all MinIO data and Kafka state) by running:

docker compose -f infra/docker-compose.yml down -v

• Remove the Spark checkpoint directory by running:

rm -rf /tmp/spark-checkpoints/weather

• Navigate to your Desktop and remove the entire project directory by running:

cd ~/Desktop
rm -rf weather-pipeline

• Your OpenWeatherMap API key remains active on your account. If you want to revoke it, log in to openweathermap.org, go to the API keys tab, and delete the key.

Nice Work!

Nice Work!

You just built a complete, production-grade real-time data engineering platform from scratch. From live API ingestion through stream processing to a real-time dashboard, you now own every layer of a modern streaming architecture.

Here's what you accomplished:

• Designed and built a full API > Kafka > Spark > MinIO > Dashboard streaming pipeline, connecting a live OpenWeatherMap data source through Apache Kafka and Spark Structured Streaming into a Streamlit dashboard.
• Implemented event-time windowed aggregations with watermarking and checkpointing, plus a medallion architecture (bronze/silver/gold) on MinIO object storage.
• Added data quality checks, unit and integration tests with pytest, and a CI pipeline with Docker Compose orchestration and secure .env management.
• Secret Mission: Added Prometheus and Grafana monitoring with pre-configured dashboards and alerting rules for processing lag and error rates.

Ready to quiz yourself?