Public Sector · Intermediate
Automate Nairobi's OpenAQ air-quality feed into PostgreSQL with an Airflow DAG that can be rerun safely.
Open in SQL workspace
Open in Python Lab
How to do this projectdata/ folder of the Python Lab.You have joined a data engineering team responsible for building an environmental data platform for Nairobi. The analytics team wants to monitor air-quality conditions across the city using data published by OpenAQ. At the moment, analysts manually retrieve data from the OpenAQ API whenever they need to perform analysis, which is unreliable, repetitive and difficult to scale. Over three days you will design and implement an automated batch pipeline with Apache Airflow that periodically retrieves measurements from monitoring stations around Nairobi, validates them, and loads them into a PostgreSQL warehouse for downstream analytics. Airflow is responsible for orchestration only: it is not the database, the transformation engine or the API. Its job is to coordinate when and how the pieces of the pipeline execute.
Your role
Data Engineer, Nairobi environmental data platform team
Analysts pull OpenAQ data by hand, so the platform has no reliable, repeatable feed of Nairobi air-quality measurements. The team needs a scheduled batch pipeline that discovers the city's monitoring locations and sensors, extracts recent pollutant readings, rejects bad records with a count of what was dropped, and loads clean rows into PostgreSQL in a way that never creates duplicates when a run is repeated. If the DAG fails at 2:00 AM, it must be safe to rerun at 8:00 AM without cleaning the database first.
Before writing any DAG code, explore the API by hand. Retrieve locations around Nairobi (latitude -1.286389, longitude 36.817223, 25 km radius) with GET /v3/locations, then GET /v3/locations/{location_id}/sensors for one location and GET /v3/sensors/{sensor_id}/measurements for one sensor. Understand the chain Nairobi → locations → sensors → measurements.
Draw the proposed DAG: extract_locations → extract_sensors → extract_measurements → validate_measurements → load_to_postgres. Nairobi has many locations and each has many sensors, so decide where Dynamic Task Mapping lets Airflow create tasks at runtime instead of hard-coding every location and sensor.
Run the project with Docker Compose: Airflow (API server or web UI, scheduler, DAG processor, metadata database, worker or executor components) plus a PostgreSQL warehouse that is logically separate from Airflow's metadata database. Create two Airflow Connections now: one for the warehouse and one holding your OpenAQ API key.
Design raw_air_quality to hold at minimum location_id, location_name, sensor_id, parameter, unit, value, measurement_timestamp and loaded_timestamp. Do not reach for an auto-incrementing ID first. Decide what uniquely identifies one pollutant measurement and make that the primary key so the table supports idempotency with ON CONFLICT.
Implement the five tasks with the TaskFlow API. extract_locations returns a list of {id, name}. extract_sensors runs per location and must not assume every location has the same sensors. extract_measurements handles pagination, HTTP errors, timeouts, rate limits and empty responses. validate_measurements rejects null values, missing sensor IDs, missing timestamps, impossible values and duplicates, and logs extracted, valid and rejected counts. load_to_postgres loads valid rows idempotently.
Trigger the DAG manually and watch the Graph view fan out into mapped task instances. Open a mapped instance's logs to confirm what it fetched. Record SELECT COUNT(*) FROM raw_air_quality, trigger the DAG again, record the count again and explain the difference. Then run a data-quality query grouped by location_name and parameter with counts and min and max measurement times, and confirm real Nairobi stations appear.
A production engineer demonstrates failure, not just success. Test at least four scenarios and document what you observe for each: an invalid or missing API key (task failure, HTTP error, logs, retry behaviour, final DAG state); the warehouse container stopped mid-run; a null or negative measurement passed through validation; and a plain rerun of the same pipeline.
Configure the DAG to run @daily with catchup set deliberately. Then reason about an hourly schedule: API volume, number of measurements, duplicate handling, time windows, database growth, rate limits and pipeline runtime.
Finish the README so it answers the eight design questions (architecture, batch processing, dynamic task mapping, connections, idempotency, reliability, data quality, scalability), lay the repository out in the required structure with no secrets committed, gather the twelve deliverables, and prepare a live demonstration covering architecture, the OpenAQ API, the DAG, mapped tasks, task logs, PostgreSQL data, the idempotency test and failure and retry behaviour. Explain decisions, not code.
Optional challenges: build a data mart (raw_air_quality → stg_air_quality → daily_air_quality_summary with daily averages by pollutant and station); introduce dbt for transformations and tests; alert when PM2.5 exceeds a threshold; connect Grafana or Power BI to chart PM2.5 and PM10 over time, pollution by station and the worst locations; extract incrementally from the last successful measurement timestamp; and make the city, latitude, longitude and radius configurable so the same DAG serves another city.
There is no file to download. The data comes live from the OpenAQ API v3 at https://api.openaq.org/v3, which exposes a hierarchy of locations, the sensors installed at each location, and the measurements each sensor records. Start from Nairobi's centre (latitude -1.286389, longitude 36.817223) with a 25 km radius, then follow GET /v3/locations, GET /v3/locations/{location_id}/sensors and GET /v3/sensors/{sensor_id}/measurements. You need an OpenAQ API key, stored in an Airflow Connection and never hard-coded or committed to GitHub. Pollutants include PM2.5, PM10, NO2, CO, O3 and SO2; expect paginated responses, a meta object on every reply, timestamps in more than one representation and units that vary by parameter.
The scenario
Objectives
Tasks and business questions
Dataset and data dictionaryraw_air_qualityThe warehouse table a correct pipeline produces: 13,872 hourly readings from seven Nairobi stations and six pollutants over a fortnight, keyed on (sensor_id, measured_at). Supplied so the SQL exercises have something to run against; your own DAG must still build the real one from the OpenAQ API. · 13,872 rows
| Column | Type | Description | Example |
|---|---|---|---|
location_id | integer | OpenAQ location identifier for the monitoring station. | 2178 |
location_name | text | Station name. | Nairobi US Embassy |
sensor_id | integer | Sensor identifier; one sensor measures one pollutant at one station. Half of the primary key. | 21780 |
parameter | text | Pollutant measured: pm25, pm10, no2, co, o3 or so2. | pm25 |
value | decimal | Measured concentration. | 10.33 |
unit | text | Unit of the reading; it varies by pollutant. | µg/m³ |
measured_at | timestamp |
9 exercises worth 190 points. Your query runs against this project’s dataset and is marked automatically.
7 exercises worth 130 points. Your code runs in the Python Lab against this project’s data and is marked automatically.
Marked with the project's own rubric: DAG design (20%), extraction and API handling (15%), dynamic task mapping (10%), PostgreSQL and database design (10%), idempotency (15%), data validation (10%), failure handling and retries (10%), documentation and architecture (5%), presentation and explanation (5%). The idempotency test must be demonstrated live with row counts before and after a rerun; "the DAG ran successfully" is not evidence. All four failure scenarios must be documented with what was observed. Presentations should explain engineering decisions, not read Python line by line.
| Hour the reading describes, in UTC. Half of the primary key. |
2026-02-01 00:00:00 |
ingested_at | timestamp | When the pipeline loaded the row, in UTC. | 2026-02-01 00:20:00 |
GitHub repository containing the complete project in the required structure (dags/, sql/, config/, logs/, tests/, docker-compose.yaml, requirements.txt, .env.example, .gitignore, README.md) with no committed secrets