Build a Weather Data ETL Pipeline
Build a Python ETL pipeline that loads weather API data into DuckDB.
Introduction
30 Second Summary
Weather apps give you real-time data at your fingertips, but behind the scenes someone has to pull that data from APIs, clean it up, and store it for analysis. That process is called an ETL pipeline, and it's one of the most in-demand skills in data engineering.
In this project, you will build a Python ETL pipeline that extracts live weather forecast data from a free API, transforms it into clean records, and loads it into an analytical database for querying.
What You'll Build
A class-based Python ETL pipeline that pulls 7-day weather forecasts for multiple cities from the Open-Meteo API, transforms nested JSON into structured records, and loads them into a DuckDB database for analysis with advanced SQL.
By the end of this project, you'll have:
- A Python ETL pipeline class with separate extract, transform, and load methods that pull live weather data from the Open-Meteo API.
- A DuckDB database file containing structured weather readings from five cities around the world.
- Analytical SQL queries using CTEs and window functions that answer real questions about weather trends.
- Secret Mission: Add a data quality validation step that catches bad data before it enters your warehouse.
Are there any prerequisites?
You should be comfortable with basic Python (variables, functions, loops) and basic SQL (SELECT, WHERE, JOIN). No data engineering experience is needed.
Install Python and VS Code
Your ETL pipeline is built on two foundational tools: Python to write and run the pipeline code, and VS Code as your code editor.
In this step, you'll install both tools and confirm they work together from a terminal inside VS Code.
In this step, get ready to:
- Install Python 3.14.
- Install VS Code with the Python extension.
- Verify both tools work from the terminal.
Download and install Python
Python is the programming language you'll use to build every part of the ETL pipeline. Let's check if it's already on your machine.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.
- Check your Python version by running one of these commands:
macOS
python3 --versionWindows
python --version✔️ I see Python 3.14 or higher
Python 3.14 is ready. Skip ahead to the next substep where you'll install VS Code.
ⓧ I see an older version
You have Python installed, but the project requires 3.14 or higher. Let's install the correct version alongside your existing one.
macOS
- Open https://www.python.org/downloads/release/python-3145/ in your browser.
- Scroll down to Files and download the macOS 64-bit universal2 installer.
- Open the downloaded .pkg file and follow the prompts to install.
- After installation, close and reopen your terminal.
- Verify the new version by running:
python3 --versionYou should see Python 3.14.5 in the output.
Why does macOS use python3?
On macOS, the python3 command points to your installed Python 3 version. The plain python command may not exist or may point to an older system version.
Windows
- Open https://www.python.org/downloads/release/python-3145/ in your browser.
- Scroll down to Files and download the Windows installer (64-bit).
- Run the installer.
- Check the Add python.exe to PATH box at the bottom of the first screen.
- Click Install Now.
- After installation, close and reopen your Command Prompt.
- Verify the new version by running:
python --versionYou should see Python 3.14.5 in the output.
ⓧ Command not found
Python isn't installed yet. Let's install it now.
macOS
- Open https://www.python.org/downloads/release/python-3145/ in your browser.
- Scroll down to Files and download the macOS 64-bit universal2 installer.
- Open the downloaded .pkg file and follow the prompts to install.
- After installation, close and reopen your terminal.
- Verify the installation by running:
python3 --versionYou should see Python 3.14.5 in the output.
Windows
- Open https://www.python.org/downloads/release/python-3145/ in your browser.
- Scroll down to Files and download the Windows installer (64-bit).
- Run the installer.
- Check the Add python.exe to PATH box at the bottom of the first screen.
- Click Install Now.
- After installation, close and reopen your Command Prompt.
- Verify the installation by running:
python --versionYou should see Python 3.14.5 in the output.
Still not seeing Python 3.14?
On Windows, make sure you checked the Add python.exe to PATH box. If you missed it, uninstall Python from Settings > Apps and reinstall with the box checked.
On macOS, make sure you closed and reopened the terminal after installation. The new version won't appear in a terminal session that was open during the install.
help me troubleshoot my Python installation
Download and install VS Code
VS Code is the code editor where you'll write your pipeline scripts and run them from a built-in terminal.
- Open https://code.visualstudio.com in your browser.
- Click the download button for your operating system (macOS or Windows).
- Run the installer and follow the prompts to complete the installation.
- Open VS Code by pressing Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Visual Studio Code and press Enter to open it.
Now install the Python extension so VS Code understands your Python code.
- Open the Extensions panel by pressing Cmd+Shift+X (macOS) or Ctrl+Shift+X (Windows).
- Type Python in the search bar.
- Find the extension published by Microsoft (it should be the top result).
- Click Install.
What does the Python extension do?
The Python extension adds syntax highlighting, code suggestions, and debugging support for Python files in VS Code. It also helps VS Code detect which Python version to use when you run your scripts.
Verify your installations
Let's confirm both Python and pip (Python's package installer) work from inside VS Code.
- In VS Code, open a new terminal by pressing Ctrl+` (backtick key, same shortcut on macOS and Windows).
- Check your Python version by running:
macOS
python3 --versionWindows
python --version- You should see Python 3.14.5 in the output.
- Now check that pip is available by running:
macOS
pip3 --versionWindows
pip --version- You should see a version number and a path confirming pip is installed.
Command not found in VS Code terminal?
If the VS Code terminal doesn't recognize python or python3, close VS Code completely and reopen it. VS Code loads PATH variables when it starts, so it won't see newly installed tools until you restart it.
help me get Python working in my VS Code terminal
Your development environment is ready. Next up, you'll create the project structure and install the Python packages that power the ETL pipeline.
Set Up the Project and Install Dependencies
With Python and VS Code ready, you need a dedicated workspace for your pipeline project. A clean folder structure keeps your code organized, and a virtual environment isolates your project's packages from the rest of your system.
In this step, you'll create the project folder, set up an isolated Python environment, and install the two libraries your pipeline depends on.
In this step, get ready to:
- Create a project folder and virtual environment.
- Install the requests and duckdb packages.
- Create the project files.
Create the project folder and virtual environment
Every Python project should live in its own folder with its own virtual environment. The virtual environment gives your project a private copy of Python and any packages you install, so different projects don't conflict with each other.
- In your VS Code terminal, move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir weather-etl
cd weather-etl- Now open this folder in VS Code so it becomes your workspace.
- In VS Code, click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop, select the weather-etl folder, then click Open (macOS) or Select Folder (Windows).
Trust the authors?
VS Code may ask if you trust the authors of the files in this folder. Click Yes, I trust the authors to continue.
- Open a new terminal by pressing Cmd+` (macOS) or Ctrl+` (Windows).
Your terminal should now show weather-etl as the current directory.
- Create a virtual environment by running this command:
python -m venv venvWhat does this command do?
python -m venv venv tells Python to create a virtual environment in a folder called venv. This folder contains a private copy of the Python interpreter and a place to install packages without affecting your system-wide Python installation.
- Activate the virtual environment so that any packages you install go into this project's private space:
macOS
source venv/bin/activateWindows
venv\Scripts\activate- Confirm the virtual environment is active. You should see (venv) at the beginning of your terminal prompt.
Don't see (venv) in your prompt?
On Windows, you may get a "running scripts is disabled" error. Run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell, then try the activate command again.
On macOS, make sure you typed source before the path. Without it, the activation won't stick to your current terminal session.
Help me activate my virtual environment.
Install the required packages
Your pipeline needs two external libraries. The requests library handles HTTP calls to the weather API. DuckDB is a local analytical database where you'll store and query the weather data.
- Install both packages by running this command:
pip install requests==2.34.2 duckdb==1.5.3Why pin version numbers?
The == syntax pins each package to an exact version. This ensures your pipeline works the same way every time, regardless of what newer versions exist.
Production pipelines always pin dependencies to avoid unexpected breakage.
- You should see output ending with Successfully installed followed by the package names and their versions.
Seeing an install error?
If you see an error mentioning Microsoft Visual C++ Redistributable, you may need to install it from Microsoft's official downloads page. This is a Windows-specific requirement for DuckDB.
If the command fails with a permissions error, make sure your virtual environment is active. You should see (venv) in your terminal prompt.
Help me fix my pip install error.
Create the project files
Your project needs two Python files. pipeline.py will hold the main ETL logic. queries.py will hold the analytical SQL queries you run against the loaded data.
- Create both files by running these commands in your terminal:
macOS
touch pipeline.py queries.pyWindows
New-Item pipeline.py -ItemType File
New-Item queries.py -ItemType File- Confirm the files exist by running:
ls- You should see pipeline.py, queries.py, and the venv folder in the output.
Don't see the files?
Make sure your terminal is in the weather-etl folder. If ls shows unexpected contents, check the folder path shown in your terminal prompt.
Help me find my project files.
Your workspace is ready with all dependencies installed. Next up, you'll start building the Extract phase of your ETL pipeline, pulling live weather data from the Open-Meteo API.
Build the Extract Phase
Your project environment is ready with all dependencies installed and both files created. Now it's time to write the first real code for your ETL pipeline: the Extract phase.
The Open-Meteo API provides free weather forecast data for any location on Earth, with no API key required. In this step, you'll build a Python class that pulls hourly weather data for five cities and collects the raw responses for later processing.
In this step, get ready to:
- Create a WeatherETL class with city coordinates and database configuration.
- Write an extract method that calls the Open-Meteo API for each city.
- Test the extract phase by running the script and examining the raw API response.
Create the WeatherETL class
In Object-Oriented Programming, a class groups related data and functions together. Your WeatherETL class will hold the list of cities to query and the database path to load data into.
- In VS Code, open pipeline.py from the Explorer sidebar and paste the following code:
import requests
import duckdb
from datetime import datetime
class WeatherETL:
"""ETL pipeline that extracts weather data from Open-Meteo, transforms it, and loads into DuckDB."""
def __init__(self, db_path="weather.db"):
# Store the path to the DuckDB database file
self.db_path = db_path
# Define the cities we want weather data for, with their coordinates
self.cities = [
{"name": "London", "latitude": 51.51, "longitude": -0.13},
{"name": "New York", "latitude": 40.71, "longitude": -74.01},
{"name": "Tokyo", "latitude": 35.68, "longitude": 139.69},
{"name": "Sydney", "latitude": -33.87, "longitude": 151.21},
{"name": "Cairo", "latitude": 30.04, "longitude": 31.24},
]What does this code do?
The import statements bring in the libraries you installed earlier. requests handles HTTP calls, duckdb connects to your database, and datetime will help with timestamps later.
The __init__ method runs automatically when you create a new WeatherETL object. It saves two things: the database file path and a list of five cities with their GPS coordinates. The API needs latitude and longitude to know which location to return weather for.
- Save pipeline.py by pressing Cmd+S (macOS) or Ctrl+S (Windows).
- In your terminal, run the script to confirm there are no syntax errors:
python pipeline.py- You should see no output and no error messages, which means Python parsed your class successfully.
Seeing a syntax error?
Check that every opening bracket, parenthesis, and quote has a matching closing one. Make sure the self.cities list ends with a closing ] on its own line, indented to match the opening.
help me fix a syntax error in my WeatherETL class
Write the extract method
The extract method is where your pipeline reaches out to the internet and pulls live data. It will loop through each city, call the Open-Meteo API, and collect the raw JSON responses.
- In pipeline.py, add the following extract method directly below the __init__ method, inside the WeatherETL class:
def extract(self):
"""Extract hourly weather data from Open-Meteo API for each city."""
raw_data = []
for city in self.cities:
# Build the API request URL and parameters
url = "https://api.open-meteo.com/v1/forecast"
params = {
"latitude": city["latitude"],
"longitude": city["longitude"],
"hourly": "temperature_2m,relative_humidity_2m,wind_speed_10m",
"forecast_days": 7,
"timezone": "auto",
}
# Send the GET request and check for errors
response = requests.get(url, params=params)
response.raise_for_status()
# Parse the JSON response and tag it with the city name
data = response.json()
data["city_name"] = city["name"]
raw_data.append(data)
print(f" Extracted data for {city['name']}")
return raw_dataWhat does this code do?
- requests.get(url, params=params) sends an HTTP GET request to the Open-Meteo forecast endpoint with the city's coordinates.
- The params dictionary asks for three hourly variables: temperature at 2 meters, relative humidity, and wind speed at 10 meters.
- response.raise_for_status() stops the script with a clear error if the API returns a failure code (like 404 or 500).
- data["city_name"] = city["name"] adds the city's name into the response dictionary so you can identify which data belongs to which city later.
Why use a class for this?
Wrapping ETL stages in a class is the standard pattern used by production data engineering frameworks. Each method handles one responsibility (extract, transform, or load), keeping the code organized and testable. The self keyword lets every method access the shared city list and database path without passing them as arguments each time.
- Save pipeline.py.
- You won't see a change yet. The next section tests this method by actually calling the API.
Test the extract phase
To confirm the extract method works, you'll add a small block of code at the bottom of the file that runs the pipeline and prints the first city's raw response.
- At the very bottom of pipeline.py (outside the class, with no indentation), add this code:
if __name__ == "__main__":
pipeline = WeatherETL()
raw = pipeline.extract()
print(raw[0])What does this code do?
The if __name__ == "__main__" block only runs when you execute the file directly (not when it's imported by another file). It creates a WeatherETL object, calls extract(), and prints the first city's full response so you can see the JSON structure.
- Save pipeline.py.
- Run the script by entering this command in your terminal:
python pipeline.py- You should see five "Extracted data for..." messages followed by a large JSON dictionary printed to the terminal.
What should I see?
The output starts with the five extraction messages, one per city. Then you'll see London's full API response, which is a dictionary containing keys like latitude, longitude, timezone, and hourly.
Inside the hourly key, you'll find parallel arrays: time contains 168 timestamps (7 days of hourly data), and temperature_2m, relative_humidity_2m, wind_speed_10m each contain 168 matching values.
Seeing a connection error?
Make sure you have an active internet connection. The Open-Meteo API requires no authentication, so if you can browse the web, the API should be reachable.
If you see ModuleNotFoundError: No module named 'requests', your virtual environment may not be activated. Run venv\Scripts\activate and try again.
help me fix a connection or import error when running my weather ETL pipeline
✔️ Awesome, I've got everything!
Great. Double check you've saved pipeline.py and you can see the raw JSON output in your terminal.
ⓧ I'd like to double check the full code
Here's what your complete pipeline.py should look like at this point:
import requests
import duckdb
from datetime import datetime
class WeatherETL:
"""ETL pipeline that extracts weather data from Open-Meteo, transforms it, and loads into DuckDB."""
def __init__(self, db_path="weather.db"):
self.db_path = db_path
self.cities = [
{"name": "London", "latitude": 51.51, "longitude": -0.13},
{"name": "New York", "latitude": 40.71, "longitude": -74.01},
{"name": "Tokyo", "latitude": 35.68, "longitude": 139.69},
{"name": "Sydney", "latitude": -33.87, "longitude": 151.21},
{"name": "Cairo", "latitude": 30.04, "longitude": 31.24},
]
def extract(self):
"""Extract hourly weather data from Open-Meteo API for each city."""
raw_data = []
for city in self.cities:
url = "https://api.open-meteo.com/v1/forecast"
params = {
"latitude": city["latitude"],
"longitude": city["longitude"],
"hourly": "temperature_2m,relative_humidity_2m,wind_speed_10m",
"forecast_days": 7,
"timezone": "auto",
}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
data["city_name"] = city["name"]
raw_data.append(data)
print(f" Extracted data for {city['name']}")
return raw_data
if __name__ == "__main__":
pipeline = WeatherETL()
raw = pipeline.extract()
print(raw[0])Your extract phase is pulling live weather data from the internet. Next up, you'll transform that messy nested JSON into clean, flat records ready for your database.
Build the Transform Phase
In the last step, you built the extract phase and saw the raw JSON that Open-Meteo returns. The data arrives as parallel arrays nested inside an hourly object.
Your transform method will flatten this nested structure into simple, uniform dictionaries. Each dictionary will represent one hourly weather reading for one city, ready to load into a database.
In this step, get ready to:
- Write the transform method to flatten nested JSON into clean records.
- Handle missing data and round numeric values to 1 decimal place.
- Test the transform phase by printing sample records.
Write the transform method
The Open-Meteo API returns hourly data as separate arrays that share the same index. Position 0 in the time array corresponds to position 0 in temperature_2m, relative_humidity_2m, and wind_speed_10m.
- In pipeline.py, find the end of the extract method (the line return raw_data).
- Add the following transform method below it, inside the WeatherETL class:
def transform(self, raw_data):
"""Transform nested API responses into flat records."""
records = []
for city_data in raw_data:
city_name = city_data["city_name"]
hourly = city_data["hourly"]
timestamps = hourly["time"]
temperatures = hourly["temperature_2m"]
humidity = hourly["relative_humidity_2m"]
wind_speed = hourly["wind_speed_10m"]
for i in range(len(timestamps)):
record = {
"city": city_name,
"timestamp": timestamps[i],
"temperature_c": round(temperatures[i], 1) if temperatures[i] is not None else None,
"humidity_percent": round(humidity[i], 1) if humidity[i] is not None else None,
"wind_speed_kmh": round(wind_speed[i], 1) if wind_speed[i] is not None else None,
}
records.append(record)
return recordsWhat does this code do?
The outer for loop iterates over each city's raw API response. It pulls out the parallel arrays from the nested hourly object into named variables (timestamps, temperatures, humidity, wind_speed).
The inner for loop walks through each index position. It builds one flat dictionary per hourly reading, combining the city name with the values at that index from each array.
You won't see a change yet. The next substep explains the rounding logic, then you'll test the full pipeline.
Understand the data type conversions
Weather APIs occasionally return None for readings where the sensor had no data. Calling round() on None would crash your pipeline.
How does the None guard work?
Each numeric field uses a conditional expression:
round(temperatures[i], 1) if temperatures[i] is not None else None- If the value exists, round(..., 1) limits it to 1 decimal place. This keeps your data consistent and prevents floating-point noise like 12.300000000000001.
- If the value is None, the record stores None without crashing. You'll handle these gaps in the load phase.
Why flatten parallel arrays?
Databases expect each row to be a complete record. Parallel arrays (where index 0 in five different lists represents the same reading) are compact for API transport but unusable for SQL queries.
Flattening them into one dictionary per reading is the standard ETL transformation pattern. It converts API-friendly structures into database-friendly rows.
Test the transform phase
Now update the test script at the bottom of pipeline.py to call transform and print a sample of the results.
- In pipeline.py, scroll to the bottom and find the if __name__ == "__main__": block.
- Find these lines:
if __name__ == "__main__":
pipeline = WeatherETL()
raw = pipeline.extract()
print(raw[0])- Replace the last two lines (raw = pipeline.extract() and print(raw[0])) with the following:
if __name__ == "__main__":
pipeline = WeatherETL()
raw = pipeline.extract()
records = pipeline.transform(raw)
print(records[:5])- Save pipeline.py.
- Run the pipeline by entering this command in your terminal:
python pipeline.pyYou should see the extraction messages for each city, followed by a list of 5 dictionaries. Each dictionary contains the keys city, timestamp, temperature_c, humidity_percent, and wind_speed_kmh.
Seeing a TypeError or KeyError?
- A TypeError on round() means your None guard is missing. Make sure each numeric field uses the if ... is not None else None pattern.
- A KeyError means a dictionary key doesn't match the API response. Double-check that you used time, temperature_2m, relative_humidity_2m, and wind_speed_10m exactly as shown.
- Make sure the transform method is indented inside the WeatherETL class, at the same level as extract.
Still stuck? Help me debug a TypeError or KeyError in my transform method.
✔️ Awesome, I've got everything!
Great. Double-check you've saved pipeline.py before moving on.
ⓧ I'd like to double check the full code
Here is the complete pipeline.py at this point:
import requests
import duckdb
from datetime import datetime
class WeatherETL:
"""ETL pipeline that extracts weather data from Open-Meteo, transforms it, and loads into DuckDB."""
def __init__(self, db_path="weather.db"):
self.db_path = db_path
self.cities = [
{"name": "London", "latitude": 51.51, "longitude": -0.13},
{"name": "New York", "latitude": 40.71, "longitude": -74.01},
{"name": "Tokyo", "latitude": 35.68, "longitude": 139.69},
{"name": "Sydney", "latitude": -33.87, "longitude": 151.21},
{"name": "Cairo", "latitude": 30.04, "longitude": 31.24},
]
def extract(self):
"""Extract hourly weather data from Open-Meteo API for each city."""
raw_data = []
for city in self.cities:
url = "https://api.open-meteo.com/v1/forecast"
params = {
"latitude": city["latitude"],
"longitude": city["longitude"],
"hourly": "temperature_2m,relative_humidity_2m,wind_speed_10m",
"forecast_days": 7,
"timezone": "auto",
}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
data["city_name"] = city["name"]
raw_data.append(data)
print(f" Extracted data for {city['name']}")
return raw_data
def transform(self, raw_data):
"""Transform nested API responses into flat records."""
records = []
for city_data in raw_data:
city_name = city_data["city_name"]
hourly = city_data["hourly"]
timestamps = hourly["time"]
temperatures = hourly["temperature_2m"]
humidity = hourly["relative_humidity_2m"]
wind_speed = hourly["wind_speed_10m"]
for i in range(len(timestamps)):
record = {
"city": city_name,
"timestamp": timestamps[i],
"temperature_c": round(temperatures[i], 1) if temperatures[i] is not None else None,
"humidity_percent": round(humidity[i], 1) if humidity[i] is not None else None,
"wind_speed_kmh": round(wind_speed[i], 1) if wind_speed[i] is not None else None,
}
records.append(record)
return records
if __name__ == "__main__":
pipeline = WeatherETL()
raw = pipeline.extract()
records = pipeline.transform(raw)
print(records[:5])Your transform phase is working. Next up, you'll build the load phase to push these clean records into a DuckDB database.
Build the Load Phase
Your extract and transform phases are working. You have a list of clean, flat dictionaries ready to store. But right now they only live in memory.
The load phase writes your transformed data into DuckDB so it persists on disk as a real database file. Once the data is stored, you can query it with SQL anytime without re-running the extraction.
In this step, get ready to:
- Write a load method that creates a table and inserts records into DuckDB.
- Add a run method that orchestrates the full ETL pipeline.
- Execute the pipeline end-to-end and verify data loaded correctly.
Write the load method
The load method connects to DuckDB, creates a table to hold your weather data, and inserts every transformed record. It also clears out old data each time so you always have a fresh dataset.
- In pipeline.py, add the following load method directly below the transform method:
def load(self, records):
"""Load transformed records into DuckDB."""
con = duckdb.connect(self.db_path)
con.execute("""
CREATE TABLE IF NOT EXISTS weather_readings (
city VARCHAR,
timestamp TIMESTAMP,
temperature_c DOUBLE,
humidity_percent DOUBLE,
wind_speed_kmh DOUBLE
)
""")
con.execute("DELETE FROM weather_readings")
con.executemany(
"""
INSERT INTO weather_readings (city, timestamp, temperature_c, humidity_percent, wind_speed_kmh)
VALUES (?, ?, ?, ?, ?)
""",
[
(r["city"], r["timestamp"], r["temperature_c"], r["humidity_percent"], r["wind_speed_kmh"])
for r in records
],
)
count = con.execute("SELECT COUNT(*) FROM weather_readings").fetchone()[0]
con.close()
return countWhat does this code do?
This method handles the final "L" in ETL. Here's what each section does:
- duckdb.connect(self.db_path) opens (or creates) a persistent database file called weather.db.
- CREATE TABLE IF NOT EXISTS defines the table schema once. On future runs, it skips creation since the table already exists.
- DELETE FROM weather_readings clears old data so each pipeline run loads a fresh set of forecasts.
- con.executemany() inserts all records in a single batch using parameterized ? placeholders.
- The final SELECT COUNT(*) query returns the total rows inserted so the pipeline can report how much data was loaded.
You won't see a change yet. The next chunk adds a run method that calls load as part of the full pipeline.
Add a run method to orchestrate the pipeline
Production pipelines need a single entry point that calls each phase in the right order. The run method does exactly that, calling extract, transform, and load in sequence while printing progress.
- In pipeline.py, add the following run method directly below the load method:
def run(self):
"""Execute the full ETL pipeline."""
print("=" * 50)
print("WEATHER ETL PIPELINE")
print("=" * 50)
print("\n[1/3] EXTRACTING data from Open-Meteo API...")
raw_data = self.extract()
print(f" Extracted data for {len(raw_data)} cities")
print("\n[2/3] TRANSFORMING raw data...")
records = self.transform(raw_data)
print(f" Transformed into {len(records)} records")
print("\n[3/3] LOADING into DuckDB...")
count = self.load(records)
print(f" Loaded {count} records into weather.db")
print("\n" + "=" * 50)
print("PIPELINE COMPLETE")
print("=" * 50)What does this code do?
The run method is the single entry point for the entire pipeline. It calls extract, transform, and load in order.
The [1/3], [2/3], [3/3] labels act as progress indicators so you can see which phase is running. After each phase, it prints a count so you know exactly how much data flowed through.
You won't see a change yet. The next chunk updates the script's entry point to use run instead of calling methods individually.
Execute the full pipeline and verify
Right now the bottom of your file still calls extract and transform individually from the previous step. You need to replace that with a single call to run.
- In pipeline.py, scroll to the bottom and find this block:
if __name__ == "__main__":
pipeline = WeatherETL()
raw = pipeline.extract()
records = pipeline.transform(raw)
print(records[:5])- Replace the entire block with this:
if __name__ == "__main__":
pipeline = WeatherETL()
pipeline.run()- Save pipeline.py.
- Run the full pipeline by entering this command in your terminal:
python pipeline.pyYou should see output showing all three phases completing, ending with something like:
==================================================
WEATHER ETL PIPELINE
==================================================
[1/3] EXTRACTING data from Open-Meteo API...
Extracted data for London
Extracted data for New York
Extracted data for Tokyo
Extracted data for Sydney
Extracted data for Cairo
Extracted data for 5 cities
[2/3] TRANSFORMING raw data...
Transformed into 840 records
[3/3] LOADING into DuckDB...
Loaded 840 records into weather.db
==================================================
PIPELINE COMPLETE
==================================================Where did weather.db come from?
The duckdb.connect("weather.db") call inside your load method automatically created this file in your weather-etl folder. It's a persistent database file. You can query it anytime without re-running the pipeline.
- Confirm the database file was created by running:
lsYou should see weather.db listed alongside pipeline.py and queries.py in the output.
Pipeline not completing?
- If you see a connection error or timeout, check your internet connection. The extract phase needs to reach the Open-Meteo API.
- If you see NameError: name 'duckdb' is not defined, make sure import duckdb is at the top of your file.
- If you see an indentation error, double check that load and run are indented at the same level as extract and transform (they're all methods inside the class).
Still stuck? Help me debug my WeatherETL pipeline. I'm getting an error when running pipeline.py.
✔️ Awesome, I've got everything!
Great. Double check you've saved pipeline.py and that weather.db appeared in your project folder.
ⓧ I'd like to double check the full code
import requests
import duckdb
from datetime import datetime
class WeatherETL:
"""ETL pipeline that extracts weather data from Open-Meteo, transforms it, and loads into DuckDB."""
def __init__(self, db_path="weather.db"):
self.db_path = db_path
self.cities = [
{"name": "London", "latitude": 51.51, "longitude": -0.13},
{"name": "New York", "latitude": 40.71, "longitude": -74.01},
{"name": "Tokyo", "latitude": 35.68, "longitude": 139.69},
{"name": "Sydney", "latitude": -33.87, "longitude": 151.21},
{"name": "Cairo", "latitude": 30.04, "longitude": 31.24},
]
def extract(self):
"""Extract hourly weather data from Open-Meteo API for each city."""
raw_data = []
for city in self.cities:
url = "https://api.open-meteo.com/v1/forecast"
params = {
"latitude": city["latitude"],
"longitude": city["longitude"],
"hourly": "temperature_2m,relative_humidity_2m,wind_speed_10m",
"forecast_days": 7,
"timezone": "auto",
}
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
data["city_name"] = city["name"]
raw_data.append(data)
print(f" Extracted data for {city['name']}")
return raw_data
def transform(self, raw_data):
"""Transform nested API responses into flat records."""
records = []
for city_data in raw_data:
city_name = city_data["city_name"]
hourly = city_data["hourly"]
timestamps = hourly["time"]
temperatures = hourly["temperature_2m"]
humidity = hourly["relative_humidity_2m"]
wind_speed = hourly["wind_speed_10m"]
for i in range(len(timestamps)):
record = {
"city": city_name,
"timestamp": timestamps[i],
"temperature_c": round(temperatures[i], 1) if temperatures[i] is not None else None,
"humidity_percent": round(humidity[i], 1) if humidity[i] is not None else None,
"wind_speed_kmh": round(wind_speed[i], 1) if wind_speed[i] is not None else None,
}
records.append(record)
return records
def load(self, records):
"""Load transformed records into DuckDB."""
con = duckdb.connect(self.db_path)
con.execute("""
CREATE TABLE IF NOT EXISTS weather_readings (
city VARCHAR,
timestamp TIMESTAMP,
temperature_c DOUBLE,
humidity_percent DOUBLE,
wind_speed_kmh DOUBLE
)
""")
con.execute("DELETE FROM weather_readings")
con.executemany(
"""
INSERT INTO weather_readings (city, timestamp, temperature_c, humidity_percent, wind_speed_kmh)
VALUES (?, ?, ?, ?, ?)
""",
[
(r["city"], r["timestamp"], r["temperature_c"], r["humidity_percent"], r["wind_speed_kmh"])
for r in records
],
)
count = con.execute("SELECT COUNT(*) FROM weather_readings").fetchone()[0]
con.close()
return count
def run(self):
"""Execute the full ETL pipeline."""
print("=" * 50)
print("WEATHER ETL PIPELINE")
print("=" * 50)
print("\n[1/3] EXTRACTING data from Open-Meteo API...")
raw_data = self.extract()
print(f" Extracted data for {len(raw_data)} cities")
print("\n[2/3] TRANSFORMING raw data...")
records = self.transform(raw_data)
print(f" Transformed into {len(records)} records")
print("\n[3/3] LOADING into DuckDB...")
count = self.load(records)
print(f" Loaded {count} records into weather.db")
print("\n" + "=" * 50)
print("PIPELINE COMPLETE")
print("=" * 50)
if __name__ == "__main__":
pipeline = WeatherETL()
pipeline.run()Your pipeline is fully operational. It pulls live weather data, cleans it up, and stores it in a real database. Next up, you'll write SQL queries that dig into this data to uncover weather patterns and trends.
Write Analytical Queries with CTEs and Window Functions
Your ETL pipeline is complete and your DuckDB database is loaded with weather data from 5 cities across 7 days. Now comes the payoff: writing SQL queries that uncover patterns in the data.
CTEs (Common Table Expressions) and window functions are two advanced SQL features that data teams use daily. In this step, you'll write queries that calculate averages, rank cities, track hourly changes, and detect temperature anomalies.
In this step, get ready to:
- Write a CTE query that calculates daily average temperatures per city.
- Use LAG and ROW_NUMBER window functions to analyze temperature trends.
- Build a rolling average query that flags temperature anomalies.
Write a CTE query for daily temperature averages
A CTE lets you define a temporary named result set inside a query. You can think of it like creating a mini-table on the fly, then querying from it in the same statement.
Your first query will calculate the daily average temperature for each city, then sort to find the hottest days. Before writing the query, you need to install one more package that DuckDB uses to display results as tables.
- Install pandas by running this command in your terminal:
pip install pandasWhy install pandas?
The fetchdf() method in DuckDB returns query results as a pandas DataFrame. DataFrames give you nicely formatted table output with .to_string() instead of raw tuples.
- Open queries.py in VS Code and add the following code:
import duckdb
def run_queries():
"""Run analytical queries against the weather database."""
con = duckdb.connect("weather.db")
print("=" * 60)
print("WEATHER DATA ANALYSIS")
print("=" * 60)
# Query 1: CTE - Daily average temperature per city, find hottest and coldest days
print("\n--- Query 1: Hottest and Coldest Days per City (using CTE) ---\n")
result = con.execute("""
WITH daily_avg AS (
SELECT
city,
CAST(timestamp AS DATE) AS day,
ROUND(AVG(temperature_c), 1) AS avg_temp_c
FROM weather_readings
GROUP BY city, CAST(timestamp AS DATE)
)
SELECT
city,
day,
avg_temp_c
FROM daily_avg
ORDER BY avg_temp_c DESC
LIMIT 10
""").fetchdf()
print(result.to_string(index=False))
con.close()
print("\n" + "=" * 60)
print("ANALYSIS COMPLETE")
print("=" * 60)
if __name__ == "__main__":
run_queries()What does this code do?
duckdb.connect("weather.db") opens the database file your pipeline created in the previous step.
The WITH daily_avg AS (...) block is the CTE. It calculates the average temperature per city per day, creating a temporary result called daily_avg. The outer SELECT then queries from that CTE, ordering by temperature descending to show the hottest days first.
fetchdf() converts the result into a pandas DataFrame, and .to_string(index=False) prints it as a clean table without row numbers.
- Save queries.py.
- Run the script to verify your first query works:
python queries.pyYou should see a table showing 10 rows with the hottest city-day combinations at the top, including columns for city, day, and avg_temp_c.
Seeing a ModuleNotFoundError?
If you see an error mentioning pandas not being found, make sure your virtual environment is active (you should see (venv) in your terminal prompt) and run pip install pandas again.
If you see an error about weather.db not being found, make sure you ran pipeline.py successfully in the previous step to create the database.
Help me debug my queries.py file
Add window function queries with LAG and ROW_NUMBER
Window functions perform calculations across a set of rows that are related to the current row. Unlike GROUP BY which collapses rows into summaries, window functions keep every row and add computed columns alongside them.
- In queries.py, find the line that says con.close().
- Add the following Query 2 code directly above con.close():
# Query 2: Window functions - Rank cities by temperature and calculate hourly change
print("\n--- Query 2: Temperature Change from Previous Hour (using LAG) ---\n")
result = con.execute("""
SELECT
city,
timestamp,
temperature_c,
LAG(temperature_c) OVER (PARTITION BY city ORDER BY timestamp) AS prev_hour_temp,
ROUND(
temperature_c - LAG(temperature_c) OVER (PARTITION BY city ORDER BY timestamp),
1
) AS temp_change
FROM weather_readings
WHERE city = 'London'
ORDER BY timestamp
LIMIT 15
""").fetchdf()
print(result.to_string(index=False))What does LAG do?
LAG(temperature_c) OVER (PARTITION BY city ORDER BY timestamp) looks at the previous row's temperature within the same city, ordered by time. By subtracting the previous hour's temperature from the current hour, you get the hourly temperature change.
PARTITION BY city ensures LAG only looks at previous rows within the same city. Without it, the function would accidentally compare London's first row to Cairo's last row.
- Now add Query 3 below Query 2, still above con.close():
# Query 3: ROW_NUMBER - Rank cities by average temperature per day
print("\n--- Query 3: Daily City Temperature Rankings (using ROW_NUMBER) ---\n")
result = con.execute("""
WITH daily_avg AS (
SELECT
city,
CAST(timestamp AS DATE) AS day,
ROUND(AVG(temperature_c), 1) AS avg_temp_c
FROM weather_readings
GROUP BY city, CAST(timestamp AS DATE)
)
SELECT
day,
city,
avg_temp_c,
ROW_NUMBER() OVER (PARTITION BY day ORDER BY avg_temp_c DESC) AS temp_rank
FROM daily_avg
ORDER BY day, temp_rank
LIMIT 15
""").fetchdf()
print(result.to_string(index=False))What does ROW_NUMBER do here?
ROW_NUMBER() OVER (PARTITION BY day ORDER BY avg_temp_c DESC) assigns a rank (1, 2, 3...) to each city within each day, based on their average temperature. The hottest city each day gets rank 1.
This query combines a CTE (daily_avg) with a window function (ROW_NUMBER). The CTE calculates averages first, then the window function ranks the results.
- Save queries.py and run it again:
python queries.pyYou should now see three query result tables printed in sequence: the hottest days, London's hourly temperature changes, and daily city rankings.
Seeing a syntax error in your SQL?
Make sure each query block is indented consistently (4 spaces inside the function). Check that the triple-quoted SQL strings """ are properly closed.
Help me fix my SQL query syntax
Write a rolling average query with anomaly detection
Your final query combines everything: a CTE, a window frame for rolling averages, and conditional logic to flag anomalies. This is the kind of query data engineers write to detect unusual patterns in time-series data.
- In queries.py, add Query 4 below Query 3, still above con.close():
# Query 4: CTE + Window frame - Rolling 24-hour average with anomaly detection
print("\n--- Query 4: Rolling 24h Average with Anomaly Flags (CTE + Window Frame) ---\n")
result = con.execute("""
WITH rolling AS (
SELECT
city,
timestamp,
temperature_c,
ROUND(
AVG(temperature_c) OVER (
PARTITION BY city
ORDER BY timestamp
ROWS BETWEEN 23 PRECEDING AND CURRENT ROW
),
1
) AS rolling_24h_avg
FROM weather_readings
)
SELECT
city,
timestamp,
temperature_c,
rolling_24h_avg,
ROUND(temperature_c - rolling_24h_avg, 1) AS deviation,
CASE
WHEN ABS(temperature_c - rolling_24h_avg) > 5 THEN 'ANOMALY'
ELSE 'NORMAL'
END AS status
FROM rolling
WHERE ABS(temperature_c - rolling_24h_avg) > 5
ORDER BY ABS(temperature_c - rolling_24h_avg) DESC
LIMIT 15
""").fetchdf()
print(result.to_string(index=False))What does this query do?
ROWS BETWEEN 23 PRECEDING AND CURRENT ROW creates a sliding window of 24 rows (the current hour plus the 23 before it). The AVG over this window gives you a rolling 24-hour average temperature for each city.
The outer SELECT calculates how far each hour's actual temperature deviates from that rolling average. Any hour where the deviation exceeds 5 degrees gets flagged as ANOMALY. The WHERE clause filters to only show those flagged readings.
- Save queries.py and run the full analysis:
python queries.pyYou should see all four query result tables followed by ANALYSIS COMPLETE at the bottom. Query 4 shows any temperature readings that deviated significantly from the 24-hour rolling average.
Query 4 returns an empty table?
That can happen if no city's hourly temperature ever deviates more than 5 degrees from its 24-hour average during the forecast period. This is normal for mild weather weeks. The query is still working correctly.
Help me understand my Query 4 results
✔️ Awesome, I've got everything!
Great. Double check you've saved queries.py and move on.
ⓧ I'd like to double check the full code
import duckdb
def run_queries():
"""Run analytical queries against the weather database."""
con = duckdb.connect("weather.db")
print("=" * 60)
print("WEATHER DATA ANALYSIS")
print("=" * 60)
# Query 1: CTE - Daily average temperature per city, find hottest and coldest days
print("\n--- Query 1: Hottest and Coldest Days per City (using CTE) ---\n")
result = con.execute("""
WITH daily_avg AS (
SELECT
city,
CAST(timestamp AS DATE) AS day,
ROUND(AVG(temperature_c), 1) AS avg_temp_c
FROM weather_readings
GROUP BY city, CAST(timestamp AS DATE)
)
SELECT
city,
day,
avg_temp_c
FROM daily_avg
ORDER BY avg_temp_c DESC
LIMIT 10
""").fetchdf()
print(result.to_string(index=False))
# Query 2: Window functions - Rank cities by temperature and calculate hourly change
print("\n--- Query 2: Temperature Change from Previous Hour (using LAG) ---\n")
result = con.execute("""
SELECT
city,
timestamp,
temperature_c,
LAG(temperature_c) OVER (PARTITION BY city ORDER BY timestamp) AS prev_hour_temp,
ROUND(
temperature_c - LAG(temperature_c) OVER (PARTITION BY city ORDER BY timestamp),
1
) AS temp_change
FROM weather_readings
WHERE city = 'London'
ORDER BY timestamp
LIMIT 15
""").fetchdf()
print(result.to_string(index=False))
# Query 3: ROW_NUMBER - Rank cities by average temperature per day
print("\n--- Query 3: Daily City Temperature Rankings (using ROW_NUMBER) ---\n")
result = con.execute("""
WITH daily_avg AS (
SELECT
city,
CAST(timestamp AS DATE) AS day,
ROUND(AVG(temperature_c), 1) AS avg_temp_c
FROM weather_readings
GROUP BY city, CAST(timestamp AS DATE)
)
SELECT
day,
city,
avg_temp_c,
ROW_NUMBER() OVER (PARTITION BY day ORDER BY avg_temp_c DESC) AS temp_rank
FROM daily_avg
ORDER BY day, temp_rank
LIMIT 15
""").fetchdf()
print(result.to_string(index=False))
# Query 4: CTE + Window frame - Rolling 24-hour average with anomaly detection
print("\n--- Query 4: Rolling 24h Average with Anomaly Flags (CTE + Window Frame) ---\n")
result = con.execute("""
WITH rolling AS (
SELECT
city,
timestamp,
temperature_c,
ROUND(
AVG(temperature_c) OVER (
PARTITION BY city
ORDER BY timestamp
ROWS BETWEEN 23 PRECEDING AND CURRENT ROW
),
1
) AS rolling_24h_avg
FROM weather_readings
)
SELECT
city,
timestamp,
temperature_c,
rolling_24h_avg,
ROUND(temperature_c - rolling_24h_avg, 1) AS deviation,
CASE
WHEN ABS(temperature_c - rolling_24h_avg) > 5 THEN 'ANOMALY'
ELSE 'NORMAL'
END AS status
FROM rolling
WHERE ABS(temperature_c - rolling_24h_avg) > 5
ORDER BY ABS(temperature_c - rolling_24h_avg) DESC
LIMIT 15
""").fetchdf()
print(result.to_string(index=False))
con.close()
print("\n" + "=" * 60)
print("ANALYSIS COMPLETE")
print("=" * 60)
if __name__ == "__main__":
run_queries()Your analytical queries are running against real weather data. You've just written the same kind of SQL that data analysts use to build dashboards and detect patterns in production systems. Next up, you'll take on a secret mission to add data quality validation to your pipeline.
Secret mission
Add Data Quality Validation
In production pipelines, bad data can silently corrupt your entire warehouse. What if a sensor goes offline and sends null readings? What if a temperature value of 999°C slips through? Your mission: add a validation gate that catches these problems before they reach the database.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs.
Resources you used:
- weather-etl project folder (contains pipeline.py, queries.py, weather.db, and the venv virtual environment).
- Python 3.14 installation.
- VS Code installation.
Keep everything running
No action needed. Choose this if you're still actively building or plan to extend the pipeline with new data sources or queries.
- Your weather-etl folder stays on your Desktop with all project files intact.
- Python and VS Code remain installed and ready for your next project.
- The weather.db database file persists locally. You can re-run pipeline.py anytime to refresh it with new forecast data.
Pause - I'll come back to this later
Deactivate the virtual environment to free up your terminal session. Your files stay on disk for when you return.
- If your virtual environment is still active, deactivate it by running:
deactivate- When you're ready to resume, navigate back to the weather-etl folder and reactivate with venv\Scripts\activate (Windows) or source venv/bin/activate (macOS/Linux).
Delete - I don't want to use this again
Remove all project files and optionally uninstall the tools you set up.
- Deactivate the virtual environment first (if active) by running:
deactivate- Delete the entire project folder. On macOS/Linux, run:
rm -rf ~/Desktop/weather-etl- On Windows, run:
rmdir /s /q %USERPROFILE%\Desktop\weather-etl- This removes pipeline.py, queries.py, weather.db, and the venv virtual environment in one step.
- To uninstall Python on Windows, open Settings, select Apps, then Installed apps. Search for Python 3.14 and click Uninstall.
- To uninstall VS Code on Windows, follow the same path in Settings > Apps > Installed apps. Search for Visual Studio Code and click Uninstall.
- On macOS, drag the Python and VS Code apps from your Applications folder to the Trash.
Nice Work!
Nice Work!
You just built a real data engineering pipeline from scratch. You extracted live weather data from an API, transformed nested JSON into structured records, loaded everything into an analytical database, and wrote advanced SQL to uncover insights.
You've learned how to:
- Build a Python class-based ETL pipeline that extracts live data from the Open-Meteo API, transforms nested JSON into flat records, and loads them into DuckDB.
- Flatten nested dictionaries from API responses into clean, structured data using Python's Object-Oriented Programming pattern.
- Write analytical SQL using CTEs and window functions (ROW_NUMBER, LAG, rolling averages) to perform time-series analysis on weather data.
- Added data quality validation to catch null values, temperature anomalies, and missing cities before loading data into the warehouse.
Ready to quiz yourself?