Madrid Air Quality Pipeline

End-to-end analytics pipeline analysing Madrid's air quality using the METRAQ dataset (~64M rows, hourly, Jan 2001 – Dec 2024).

This project was developed for advanced data processing and visualization, showcasing large-scale time-series cleaning, custom data imputation, spatial/correlation networks, propagation modelling, forecasting, and interactive visualisation.

Key Features

• Scalable Data Processing: Utilizes pandas and pyarrow (Parquet) for fast, memory-efficient checkpointing.
• Advanced Data Imputation: Compares multiple strategies to handle missing sensor data and physically invalid values.
• Network Analysis: Uses NetworkX to build complex spatial and correlation graphs between monitoring stations.
• Interactive Dashboard: A fully featured Streamlit app for dynamic exploration of the dataset, networks, and trends.

Setup

pip install -r requirements.txt

Data

The pipeline expects the METRAQ dataset CSV files to be in data/raw/. If the dataset is not found locally, the script will automatically download the full dataset from Hugging Face into the data/raw/ directory on the first run.

Set the environment variable DATA_DIR if using a different path:

export DATA_DIR=/path/to/your/data

Run (sample data — recommended for development)

python main.py --sample

Run (full dataset)

python main.py --full

Task execution order

Tasks run in order: 1 → 2 → 3 → 4 → 5 → 6 → 7 → 8 → 9 → 10

Each task reads a Parquet checkpoint written by the previous task.

To re-run from a specific task:

python main.py --from-task 4

Interactive Dashboard

After running the pipeline, launch the Streamlit dashboard:

streamlit run streamlit_app.py

Output

• All visualisations saved to outputs/figures/
• All checkpoints saved to data/processed/

Project Structure

madrid-air-quality-analysis/
├── config.py               # All constants — edit only here
├── main.py                 # Orchestrates all tasks sequentially
├── streamlit_app.py        # Interactive dashboard
├── requirements.txt        # Pinned dependencies
├── data/
│   ├── raw/                # Original CSVs (never modified)
│   └── processed/          # Parquet checkpoints written by tasks
├── outputs/
│   └── figures/            # Generated PNG visualisations
├── tasks/
│   ├── task1_load.py       # Load data, schema validation, coverage matrix
│   ├── task2_missing.py    # Missingness analysis, physical validity flags
│   ├── task3_imputation.py # 3 imputation methods with evaluation
│   ├── task4_temporal.py   # Diurnal, seasonal, long-term trend analysis
│   ├── task5_spatial.py    # KNN & threshold spatial networks
│   ├── task6_correlation.py# Correlation-based networks per pollutant
│   ├── task7_propagation.py# (Optional) Diffusion-based propagation model
│   ├── task8_parallel.py   # Parallel correlation matrices + benchmarking
│   ├── task9_forecasting.py# (Optional) ML-based pollution forecasting
│   └── task10_viz.py       # Summary visualisations
└── utils/
    ├── io.py               # load_dataset(), save/load_checkpoint()
    ├── validate.py         # Physical bounds checking
    └── network.py          # Shared graph metrics & drawing