Build a Weather Data ETL Pipeline
Extract, transform, and load live weather data into DuckDB with Python.
Introduction
30 Second Summary
Every weather app you use pulls raw data from somewhere, cleans it up, and stores it so you can see a tidy 7-day forecast. That "pull, clean, store" pattern is exactly what data engineers do at scale every single day.
In this project, you will build a Python ETL pipeline that extracts live weather forecasts from a free API, transforms the raw data into clean DataFrames, and loads everything into an analytical database you can query with SQL.
What You'll Build
A complete Extract-Transform-Load pipeline in Python that pulls weather data from the Open-Meteo API, reshapes it with pandas, and stores it in DuckDB for analytical querying.
By the end of this project, you'll have:
- A working Python script (etl_pipeline.py) that extracts live weather forecast data from a free REST API.
- A transformation layer that cleans raw JSON into structured, analysis-ready DataFrames.
- A DuckDB database loaded with weather data that you can query using analytical SQL (aggregations, window functions, and filtering).
- Secret Mission: Add scheduling and incremental loading so your pipeline re-runs on a timer without duplicating data.
Are there any prerequisites?
This project assumes basic Python and SQL knowledge. You should be comfortable writing functions, using loops, and running simple queries.
Before We Start
Before you can build your ETL pipeline, you need a Python environment with the right packages installed.
In this step, you'll verify your Python version meets the minimum requirement, create an isolated project folder with a virtual environment, and install the three packages your pipeline depends on.
In this step, get ready to:
- Verify Python 3.10 or higher is installed.
- Create a project folder with a virtual environment.
- Install and verify the required Python packages.
Verify your Python version
The packages this project uses require Python 3.10 or higher. You need to confirm your installed version meets this minimum before continuing.
- Open your terminal by pressing 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 this command:
macOS
python3 --versionWindows
python --version✔️ I see version 3.10 or higher
You should see something like Python 3.12.x or Python 3.11.x. Any version at 3.10 or above works perfectly.
You're all set. Move on to the next substep.
ⓧ I see an older version
If your terminal shows a version below 3.10 (like 3.9.x or 3.8.x), you need to install a newer version.
- Go to python.org/downloads and download the latest Python release (3.12 or newer is recommended).
- Run the installer. On Windows, check the box that says Add Python to PATH before clicking Install Now.
- Close and reopen your terminal, then run the version check command again to confirm the new version appears.
macOS default Python
macOS may ship with an older Python. After installing a newer version, use python3 --version to verify the new install. The python3 command should now point to your newly installed version.
ⓧ Command not found
Python is not installed on your machine yet.
- Go to python.org/downloads and download the latest Python release.
- Run the installer. On Windows, check the box that says Add Python to PATH before clicking Install Now.
- Close and reopen your terminal after installation completes.
- Run the version check command again to confirm Python is installed.
Create your project folder and virtual environment
A virtual environment keeps your project's packages isolated from your system Python. This prevents version conflicts between different projects on your machine.
- 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- Create a virtual environment inside your project folder by running:
macOS
python3 -m venv venvWindows
python -m venv venv- Activate the virtual environment by running:
macOS
source venv/bin/activateWindows
venv\Scripts\activateHow do I know it worked?
After activating, you should see (venv) appear at the beginning of your terminal prompt. This confirms you're now working inside the virtual environment.
Don't see (venv) in your prompt?
Make sure you ran the activate command from inside the weather-etl folder. The venv folder must exist in your current directory for activation to work.
On Windows, if you see an error about execution policies, run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned first, then try activating again.
Help me activate my virtual environment
Install the required packages
Your ETL pipeline needs three packages: requests for calling the weather API, pandas for transforming data, and duckdb for the analytical database.
- Install all three packages by running:
pip install requests pandas duckdbWhat are these packages?
requests is a Python library for making HTTP requests to web APIs. You'll use it to pull weather data from the Open-Meteo API.
pandas is the standard data manipulation library. It provides DataFrames for cleaning and reshaping data.
duckdb is a fast, analytical database that runs in-process. It's purpose-built for the kind of aggregate queries data engineers run daily.
- Verify everything installed correctly by running:
python -c "import requests, pandas, duckdb; print('All packages installed!')"- You should see All packages installed! printed in your terminal.
Seeing an import error?
Make sure you see (venv) in your terminal prompt. If it's missing, the packages installed to your system Python instead of the virtual environment. Activate the environment and run pip install again.
If you see a network error, check your internet connection and try the install command again.
Help me fix my package installation
Your environment is ready to go. Next up, you'll write the extraction function that pulls live weather data from the Open-Meteo API.
Set Up Your Environment
Every data engineering pipeline starts with a clear structure. You need to know what data you want, where it lives, and how to pull it in.
In this step, you'll create your pipeline script and write the extraction layer. This is the "E" in ETL. It connects to the Open-Meteo API and pulls weather forecast data for three cities.
In this step, get ready to:
- Create etl_pipeline.py with configuration constants.
- Write an extract() function that calls the weather API.
- Run the script and verify the API returns data.
Create the pipeline file and add configuration
Before writing any logic, you need a place for your code and the constants that tell it what data to fetch and from where.
- In your code editor, right-click inside the weather-etl folder in the **Explorer** sidebar.
- Select New File and name it etl_pipeline.py.
- Paste the following imports and configuration into the file:
import requests
import pandas as pd
import duckdb
from datetime import datetime
# === CONFIGURATION ===
CITIES = {
"New York": {"latitude": 40.7128, "longitude": -74.0060},
"London": {"latitude": 51.5074, "longitude": -0.1278},
"Tokyo": {"latitude": 35.6762, "longitude": 139.6503},
}
API_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_VARIABLES = "temperature_2m,precipitation,wind_speed_10m,relative_humidity_2m"What does this configuration do?
CITIES is a dictionary mapping city names to their GPS coordinates. The Open-Meteo API uses latitude and longitude to look up forecast data for a specific location.
API_URL points to the Open-Meteo forecast endpoint. HOURLY_VARIABLES lists the weather metrics you want: temperature, precipitation, wind speed, and humidity.
- Save etl_pipeline.py.
- Confirm the file appears in your editor's sidebar under the weather-etl folder.
File not appearing in the sidebar?
Make sure you created the file inside the weather-etl folder, not one level above it. If you opened the wrong folder in your editor, click File then Open Folder and select weather-etl from your Desktop.
Help me find where my file was created
Write the extract function
The extract function is the first stage of your pipeline. It calls the Open-Meteo API once for each city, collects the JSON responses, and returns them as a list.
- In etl_pipeline.py, add the following code below the HOURLY_VARIABLES line:
# === EXTRACT ===
def extract(cities: dict) -> list[dict]:
"""Extract weather forecast data from Open-Meteo API for multiple cities."""
raw_data = []
for city_name, coords in cities.items():
params = {
"latitude": coords["latitude"],
"longitude": coords["longitude"],
"hourly": HOURLY_VARIABLES,
"timezone": "auto",
}
response = requests.get(API_URL, params=params, timeout=30)
response.raise_for_status()
data = response.json()
data["city_name"] = city_name
raw_data.append(data)
print(f"Extracted data for {city_name}: {len(data['hourly']['time'])} hours")
return raw_dataWhat does this function do?
The function loops through each city in the CITIES dictionary. For each one, it builds a request with the city's coordinates and the weather variables you want.
requests.get() sends the HTTP request to Open-Meteo. raise_for_status() throws an error if the API returns a failure code (like 404 or 500), so you catch problems early.
Each response gets tagged with city_name so you can tell which data belongs to which city later in the pipeline.
- Save etl_pipeline.py.
Seeing a red underline on the import?
If your editor highlights import requests with a warning, make sure your editor is using the Python interpreter from your virtual environment. Look in the bottom status bar for venv in the interpreter path.
Help me fix my Python interpreter in VS Code
Run the script and verify the API response
Now you'll add a runner block and execute the script to confirm the API returns real weather data.
- At the bottom of etl_pipeline.py, add this runner code:
if __name__ == "__main__":
raw_data = extract(CITIES)
print(raw_data[0])What does if __name__ == "__main__" do?
This block only runs when you execute the file directly (like python etl_pipeline.py). It won't run if another script imports functions from this file later.
Here it calls extract(CITIES) and prints the first city's raw response so you can inspect the data structure.
- Save etl_pipeline.py.
- Run the script in your terminal:
python etl_pipeline.pyYou should see three "Extracted data for..." lines (one per city), followed by a large JSON object printed to your terminal.
- Confirm you see the hourly object in the output containing arrays for time, temperature_2m, precipitation, wind_speed_10m, and relative_humidity_2m.
What should I see?
The output is a large JSON dictionary. Look for a key called hourly containing arrays of 168 values (7 days times 24 hours). Each array holds one week of hourly readings for that weather variable.
You'll also see city_name set to "New York" at the top level. That's the tag you added so the transform step can identify which city each response belongs to.
Seeing a connection error?
Make sure you have an active internet connection. The script calls the Open-Meteo API over HTTPS.
If you see ModuleNotFoundError, your virtual environment may not be activated. Run venv\Scripts\activate (Windows) or source venv/bin/activate (macOS/Linux) and try again.
Help me debug my extraction script
✔️ Awesome, I've got everything!
Great. Double check you've saved etl_pipeline.py and your terminal shows data for all three cities.
ⓧ I'd like to double check the full code
Here's what your complete etl_pipeline.py should look like at the end of this step:
import requests
import pandas as pd
import duckdb
from datetime import datetime
# === CONFIGURATION ===
CITIES = {
"New York": {"latitude": 40.7128, "longitude": -74.0060},
"London": {"latitude": 51.5074, "longitude": -0.1278},
"Tokyo": {"latitude": 35.6762, "longitude": 139.6503},
}
API_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_VARIABLES = "temperature_2m,precipitation,wind_speed_10m,relative_humidity_2m"
# === EXTRACT ===
def extract(cities: dict) -> list[dict]:
"""Extract weather forecast data from Open-Meteo API for multiple cities."""
raw_data = []
for city_name, coords in cities.items():
params = {
"latitude": coords["latitude"],
"longitude": coords["longitude"],
"hourly": HOURLY_VARIABLES,
"timezone": "auto",
}
response = requests.get(API_URL, params=params, timeout=30)
response.raise_for_status()
data = response.json()
data["city_name"] = city_name
raw_data.append(data)
print(f"Extracted data for {city_name}: {len(data['hourly']['time'])} hours")
return raw_data
if __name__ == "__main__":
raw_data = extract(CITIES)
print(raw_data[0])Your extraction layer is working. You've proven that the API returns structured hourly data for each city. Next up, you'll transform this raw JSON into clean, analysis-ready DataFrames with pandas.
Extract Weather Data from the API
Your environment is ready. Now it's time to pull real weather data from the internet into your Python script.
The first phase of any ETL pipeline is Extract. You need to reach out to an external data source, grab raw data, and bring it into your program's memory. In this step, you'll use the Open-Meteo API to pull 7-day weather forecasts for three cities.
In this step, get ready to:
- Write an extraction function that calls a live weather API.
- Loop over multiple cities and collect all API responses.
- Run the script and verify raw data arrives correctly.
Create the extraction function
The Open-Meteo API is free, requires no API key, and returns structured JSON. You send it geographic coordinates and it returns hourly weather data as parallel arrays.
- Open etl_pipeline.py in your code editor.
- Find the # === CONFIGURATION === section at the top of your file. This section defines the cities and API details your pipeline will use:
# === CONFIGURATION ===
CITIES = {
"New York": {"latitude": 40.7128, "longitude": -74.0060},
"London": {"latitude": 51.5074, "longitude": -0.1278},
"Tokyo": {"latitude": 35.6762, "longitude": 139.6503},
}
API_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_VARIABLES = "temperature_2m,precipitation,wind_speed_10m,relative_humidity_2m"What does this configuration do?
CITIES is a dictionary mapping city names to their latitude and longitude coordinates. The Open-Meteo API uses these coordinates to locate the weather station nearest to each city.
API_URL is the Open-Meteo forecast endpoint. HOURLY_VARIABLES tells the API which measurements you want back: temperature, precipitation, wind speed, and humidity.
- Below the configuration, find the extract() function. This is the core of your Extract phase:
# === EXTRACT ===
def extract(cities: dict) -> list[dict]:
"""Extract weather forecast data from Open-Meteo API for multiple cities."""
raw_data = []
for city_name, coords in cities.items():
params = {
"latitude": coords["latitude"],
"longitude": coords["longitude"],
"hourly": HOURLY_VARIABLES,
"timezone": "auto",
}
response = requests.get(API_URL, params=params, timeout=30)
response.raise_for_status()
data = response.json()
data["city_name"] = city_name
raw_data.append(data)
print(f"Extracted data for {city_name}: {len(data['hourly']['time'])} hours")
return raw_dataWhat does this function do?
The function loops over each city in the CITIES dictionary. For each city, it builds a params dictionary with the coordinates and desired weather variables, then sends a GET request to the API.
response.raise_for_status() throws an error if the API returns a non-200 status code (like 404 or 500). The timeout=30 prevents the script from hanging forever if the API is slow.
After parsing the JSON, the function attaches the city name to each response so you can identify which data belongs to which city later in the pipeline.
Why does the API use parallel arrays?
The Open-Meteo API returns an hourly object containing arrays that line up by index. The first element in time[] corresponds to the first element in temperature_2m[], precipitation[], and so on.
This is a common pattern in REST APIs optimized for compactness. Your Transform phase will reshape these parallel arrays into proper table rows.
Run the pipeline and verify the output
The bottom of your file contains the entry point that calls extract() and prints the first city's raw response. Time to run it.
- Run the script by executing this command in your terminal:
python etl_pipeline.pyYou should see three lines confirming data was extracted for each city, followed by a large JSON object printed to the terminal.
- Confirm you see output similar to this:
Extracted data for New York: 168 hours
Extracted data for London: 168 hours
Extracted data for Tokyo: 168 hoursWhat should I see in the JSON?
The printed JSON object contains an hourly key. Inside it, look for these arrays: time, temperature_2m, precipitation, wind_speed_10m, and relative_humidity_2m.
Each array has 168 elements (7 days multiplied by 24 hours). The time array contains ISO 8601 timestamps in the city's local timezone.
Seeing a connection error?
Make sure you have an active internet connection. The script needs to reach api.open-meteo.com to fetch data.
If you see a ModuleNotFoundError, make sure your virtual environment is activated and packages are installed.
Help me debug a connection or import error when running my weather ETL script.
✔️ Awesome, I've got everything!
Great. Double check you have saved etl_pipeline.py and your terminal shows extraction output for all three cities.
ⓧ I'd like to double check the full code
import requests
import pandas as pd
import duckdb
from datetime import datetime
# === CONFIGURATION ===
CITIES = {
"New York": {"latitude": 40.7128, "longitude": -74.0060},
"London": {"latitude": 51.5074, "longitude": -0.1278},
"Tokyo": {"latitude": 35.6762, "longitude": 139.6503},
}
API_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_VARIABLES = "temperature_2m,precipitation,wind_speed_10m,relative_humidity_2m"
# === EXTRACT ===
def extract(cities: dict) -> list[dict]:
"""Extract weather forecast data from Open-Meteo API for multiple cities."""
raw_data = []
for city_name, coords in cities.items():
params = {
"latitude": coords["latitude"],
"longitude": coords["longitude"],
"hourly": HOURLY_VARIABLES,
"timezone": "auto",
}
response = requests.get(API_URL, params=params, timeout=30)
response.raise_for_status()
data = response.json()
data["city_name"] = city_name
raw_data.append(data)
print(f"Extracted data for {city_name}: {len(data['hourly']['time'])} hours")
return raw_data
if __name__ == "__main__":
raw_data = extract(CITIES)
print(raw_data[0])Your pipeline can now pull live weather data for three cities. Next, you'll transform this raw JSON into a clean, structured DataFrame with pandas.
Transform the Data with pandas
Your extract function is pulling live weather data from the Open-Meteo API for all three cities. Now it's time to shape that raw data into something clean and queryable.
The API returns data as parallel arrays. Each variable (temperature, precipitation, wind speed, humidity) comes as its own list of 168 values. Your transform function will reshape these arrays into a structured DataFrame, then enrich it with derived columns.
In this step, get ready to:
- Build a DataFrame from the API's parallel array format.
- Apply data cleaning and add derived columns.
- Verify the transformation by printing the output shape.
Convert the raw JSON into DataFrames
The Open-Meteo API returns an hourly object containing parallel arrays. The time array lines up index-for-index with the temperature_2m array, the precipitation array, and so on. pandas can build a table from a dictionary of these arrays directly.
- In etl_pipeline.py, find the line that says if __name__ == "__main__": and add the following function above it:
# === TRANSFORM ===
def transform(raw_data: list[dict]) -> pd.DataFrame:
"""Transform raw API responses into a clean pandas DataFrame."""
dataframes = []
for data in raw_data:
city_name = data["city_name"]
hourly = data["hourly"]
# Build a DataFrame from the parallel arrays
df = pd.DataFrame(
{
"timestamp": hourly["time"],
"temperature_2m": hourly["temperature_2m"],
"precipitation": hourly["precipitation"],
"wind_speed_10m": hourly["wind_speed_10m"],
"relative_humidity_2m": hourly["relative_humidity_2m"],
}
)
# Tag each row with its city
df["city"] = city_name
dataframes.append(df)What does this code do?
The function loops over each city's raw API response. For each one, it grabs the hourly object and builds a DataFrame where each column maps directly to one of the parallel arrays.
Adding a city column tags every row so you can tell which city's forecast you're looking at after all DataFrames are combined.
- Save etl_pipeline.py.
- You won't see a change yet. The next chunk completes the function and lets you run it.
Apply data cleaning and enrichment
Raw timestamps from the API arrive as plain strings. You need to parse them into proper datetime objects so pandas can extract components like date and hour.
- Directly below the loop you just wrote (still inside the transform function), add the following code:
combined = pd.concat(dataframes, ignore_index=True)
# Parse timestamp and extract date/hour
combined["timestamp"] = pd.to_datetime(combined["timestamp"])
combined["date"] = combined["timestamp"].dt.date
combined["hour"] = combined["timestamp"].dt.hour
# Add computed column
combined["feels_cold"] = combined["temperature_2m"] < 10
# Reorder columns
combined = combined[
[
"city",
"timestamp",
"date",
"hour",
"temperature_2m",
"precipitation",
"wind_speed_10m",
"relative_humidity_2m",
"feels_cold",
]
]
print(f"Transformed data: {combined.shape[0]} rows, {combined.shape[1]} columns")
return combinedWhat does this code do?
- pd.concat(dataframes, ignore_index=True) stacks all city DataFrames into one big table and resets the row numbers.
- pd.to_datetime() converts the string timestamps into real datetime objects. Then .dt.date and .dt.hour extract just the date or hour component.
- The feels_cold column is a boolean derived from temperature. It's True whenever temperature is below 10°C.
- Reordering columns puts city first so the data reads naturally from left to right.
- Save etl_pipeline.py.
- You won't see output yet. The next chunk wires up the transform call so you can run it.
Seeing an indentation error?
All the code in this chunk must be indented one level (4 spaces) inside the transform function. Make sure the combined = pd.concat(...) line sits at the same indent level as the for loop above it, not inside the loop.
Why isn't my transform function working?
Run the pipeline and verify the output
With the transform function complete, you need to call it from your main block so the script actually runs the transformation after extraction.
- At the bottom of etl_pipeline.py, find the if __name__ == "__main__": block.
- Find this line:
print(raw_data[0])- Replace it with these two lines:
df = transform(raw_data)
print(df.head())- Save etl_pipeline.py.
- Run the script by running this command:
python etl_pipeline.py- You should see extraction messages for each city, a line saying Transformed data: 504 rows, 9 columns, and a table showing the first 5 rows of your DataFrame.
Why 504 rows?
The Open-Meteo API returns 7 days of hourly data per city. That's 168 hours multiplied by 3 cities, which gives you 504 rows.
Seeing a NameError for transform?
Make sure the transform function is defined above the if __name__ == "__main__": block. Python reads top to bottom, so the function must exist before it's called.
Help me debug my NameError in etl_pipeline.py
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
Here's the full etl_pipeline.py as it should look at the end of this step:
import requests
import pandas as pd
import duckdb
from datetime import datetime
# === CONFIGURATION ===
CITIES = {
"New York": {"latitude": 40.7128, "longitude": -74.0060},
"London": {"latitude": 51.5074, "longitude": -0.1278},
"Tokyo": {"latitude": 35.6762, "longitude": 139.6503},
}
API_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_VARIABLES = "temperature_2m,precipitation,wind_speed_10m,relative_humidity_2m"
# === EXTRACT ===
def extract(cities: dict) -> list[dict]:
"""Extract weather forecast data from Open-Meteo API for multiple cities."""
raw_data = []
for city_name, coords in cities.items():
params = {
"latitude": coords["latitude"],
"longitude": coords["longitude"],
"hourly": HOURLY_VARIABLES,
"timezone": "auto",
}
response = requests.get(API_URL, params=params, timeout=30)
response.raise_for_status()
data = response.json()
data["city_name"] = city_name
raw_data.append(data)
print(f"Extracted data for {city_name}: {len(data['hourly']['time'])} hours")
return raw_data
# === TRANSFORM ===
def transform(raw_data: list[dict]) -> pd.DataFrame:
"""Transform raw API responses into a clean pandas DataFrame."""
dataframes = []
for data in raw_data:
city_name = data["city_name"]
hourly = data["hourly"]
df = pd.DataFrame(
{
"timestamp": hourly["time"],
"temperature_2m": hourly["temperature_2m"],
"precipitation": hourly["precipitation"],
"wind_speed_10m": hourly["wind_speed_10m"],
"relative_humidity_2m": hourly["relative_humidity_2m"],
}
)
df["city"] = city_name
dataframes.append(df)
combined = pd.concat(dataframes, ignore_index=True)
# Parse timestamp and extract date/hour
combined["timestamp"] = pd.to_datetime(combined["timestamp"])
combined["date"] = combined["timestamp"].dt.date
combined["hour"] = combined["timestamp"].dt.hour
# Add computed column
combined["feels_cold"] = combined["temperature_2m"] < 10
# Reorder columns
combined = combined[
[
"city",
"timestamp",
"date",
"hour",
"temperature_2m",
"precipitation",
"wind_speed_10m",
"relative_humidity_2m",
"feels_cold",
]
]
print(f"Transformed data: {combined.shape[0]} rows, {combined.shape[1]} columns")
return combined
if __name__ == "__main__":
raw_data = extract(CITIES)
df = transform(raw_data)
print(df.head())Your raw API data is now a clean, structured DataFrame with 504 rows and 9 columns. Next up, you'll load this data into a DuckDB database so you can query it with SQL.
Load Data into DuckDB
Your pipeline can now extract weather data from the API and transform it into a clean DataFrame. The problem is that this data lives only in memory.
In this step, you'll write the "Load" function that persists your transformed data into DuckDB, a columnar analytical database. Once loaded, you'll wire up the full Extract-Transform-Load pipeline to run end-to-end.
In this step, get ready to:
- Write the load function that creates a DuckDB table and inserts your DataFrame.
- Wire up the full ETL pipeline to run extract, transform, and load in sequence.
- Verify the pipeline completes end-to-end without errors.
Write the load function
The load function is the final stage of your ETL pipeline. It takes the clean DataFrame and persists it into a DuckDB database file on disk.
- In etl_pipeline.py, find the line that says if __name__ == "__main__": at the bottom of your file.
- Copy the following code and paste it above that line, leaving one blank line between the new function and the if __name__ block:
# === LOAD ===
def load(df: pd.DataFrame, db_path: str = "weather.duckdb") -> None:
"""Load the transformed DataFrame into DuckDB."""
conn = duckdb.connect(db_path)
conn.execute("DROP TABLE IF EXISTS weather_forecasts")
conn.execute(
"""
CREATE TABLE weather_forecasts (
city VARCHAR,
timestamp TIMESTAMP,
date DATE,
hour INTEGER,
temperature_2m DOUBLE,
precipitation DOUBLE,
wind_speed_10m DOUBLE,
relative_humidity_2m DOUBLE,
feels_cold BOOLEAN
)
"""
)
conn.execute("INSERT INTO weather_forecasts SELECT * FROM df")
row_count = conn.execute("SELECT COUNT(*) FROM weather_forecasts").fetchone()[0]
print(f"Loaded {row_count} rows into DuckDB")
conn.close()What does this code do?
- duckdb.connect(db_path) opens (or creates) a DuckDB database file called weather.duckdb. This is a persistent file on disk, not an in-memory database.
- DROP TABLE IF EXISTS ensures a clean slate each time you run the pipeline. CREATE TABLE defines the schema with typed columns matching your DataFrame.
- INSERT INTO weather_forecasts SELECT * FROM df is DuckDB's zero-copy ingestion. DuckDB can read a pandas DataFrame variable directly by name in SQL, so no row-by-row insertion is needed.
- The row count query confirms how many rows actually landed in the table.
Why DuckDB instead of SQLite?
DuckDB is purpose-built for analytical workloads (OLAP). It supports window functions, CTEs, and vectorized columnar execution that make aggregate queries fast.
Data engineers work with OLAP databases daily. Using DuckDB here teaches you the right mental model, while still being zero-config and file-based like SQLite.
- Save etl_pipeline.py.
Seeing a syntax error in the CREATE TABLE statement?
- Make sure the triple-quoted SQL string uses three double quotes on each side, with no extra indentation breaking the closing parenthesis.
- Check that each column definition ends with a comma except the last one (feels_cold BOOLEAN has no trailing comma).
- Help me fix my CREATE TABLE statement in DuckDB
Wire up the full pipeline
Now that all three functions exist, you'll connect them into a single pipeline that runs Extract, Transform, and Load in sequence.
- In etl_pipeline.py, scroll to the bottom and find the existing if __name__ == "__main__": block:
if __name__ == "__main__":
raw_data = extract(CITIES)
df = transform(raw_data)
print(df.head())- Replace the entire if __name__ block with this updated version that runs all three pipeline stages:
if __name__ == "__main__":
print("=" * 50)
print("WEATHER ETL PIPELINE")
print("=" * 50)
print("\n[1/3] Extracting data from Open-Meteo API...")
raw_data = extract(CITIES)
print("\n[2/3] Transforming data...")
df = transform(raw_data)
print("\n[3/3] Loading data into DuckDB...")
load(df)
print("\nPipeline complete!")What does this code do?
- The print statements with step numbers ([1/3], [2/3], [3/3]) give you visibility into which pipeline stage is running.
- extract(CITIES) pulls raw JSON from the API. transform(raw_data) cleans it into a DataFrame. load(df) persists it to DuckDB.
- This is the classic ETL pattern: each function has one job, and they chain together cleanly.
- Save etl_pipeline.py.
- Run the full pipeline by running this command in your terminal:
python etl_pipeline.py- You should see output showing all three stages completing, ending with a row count (around 504 rows for 3 cities x 168 hours) and "Pipeline complete!" at the bottom.
Seeing a NameError for load?
- Make sure your load function is defined above the if __name__ block, not inside it.
- Check that the function name is exactly load (lowercase) and that your call in __main__ is load(df) with the df variable from the transform step.
- Help me debug my ETL pipeline run
- Confirm the database file was created by running:
ls- You should see weather.duckdb listed alongside etl_pipeline.py in your project folder.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
import requests
import pandas as pd
import duckdb
from datetime import datetime
# === CONFIGURATION ===
CITIES = {
"New York": {"latitude": 40.7128, "longitude": -74.0060},
"London": {"latitude": 51.5074, "longitude": -0.1278},
"Tokyo": {"latitude": 35.6762, "longitude": 139.6503},
}
API_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_VARIABLES = "temperature_2m,precipitation,wind_speed_10m,relative_humidity_2m"
# === EXTRACT ===
def extract(cities: dict) -> list[dict]:
"""Extract weather forecast data from Open-Meteo API for multiple cities."""
raw_data = []
for city_name, coords in cities.items():
params = {
"latitude": coords["latitude"],
"longitude": coords["longitude"],
"hourly": HOURLY_VARIABLES,
"timezone": "auto",
}
response = requests.get(API_URL, params=params, timeout=30)
response.raise_for_status()
data = response.json()
data["city_name"] = city_name
raw_data.append(data)
print(f"Extracted data for {city_name}: {len(data['hourly']['time'])} hours")
return raw_data
# === TRANSFORM ===
def transform(raw_data: list[dict]) -> pd.DataFrame:
"""Transform raw API responses into a clean pandas DataFrame."""
dataframes = []
for data in raw_data:
city_name = data["city_name"]
hourly = data["hourly"]
df = pd.DataFrame(
{
"timestamp": hourly["time"],
"temperature_2m": hourly["temperature_2m"],
"precipitation": hourly["precipitation"],
"wind_speed_10m": hourly["wind_speed_10m"],
"relative_humidity_2m": hourly["relative_humidity_2m"],
}
)
df["city"] = city_name
dataframes.append(df)
combined = pd.concat(dataframes, ignore_index=True)
combined["timestamp"] = pd.to_datetime(combined["timestamp"])
combined["date"] = combined["timestamp"].dt.date
combined["hour"] = combined["timestamp"].dt.hour
combined["feels_cold"] = combined["temperature_2m"] < 10
combined = combined[
[
"city",
"timestamp",
"date",
"hour",
"temperature_2m",
"precipitation",
"wind_speed_10m",
"relative_humidity_2m",
"feels_cold",
]
]
print(f"Transformed data: {combined.shape[0]} rows, {combined.shape[1]} columns")
return combined
# === LOAD ===
def load(df: pd.DataFrame, db_path: str = "weather.duckdb") -> None:
"""Load the transformed DataFrame into DuckDB."""
conn = duckdb.connect(db_path)
conn.execute("DROP TABLE IF EXISTS weather_forecasts")
conn.execute(
"""
CREATE TABLE weather_forecasts (
city VARCHAR,
timestamp TIMESTAMP,
date DATE,
hour INTEGER,
temperature_2m DOUBLE,
precipitation DOUBLE,
wind_speed_10m DOUBLE,
relative_humidity_2m DOUBLE,
feels_cold BOOLEAN
)
"""
)
conn.execute("INSERT INTO weather_forecasts SELECT * FROM df")
row_count = conn.execute("SELECT COUNT(*) FROM weather_forecasts").fetchone()[0]
print(f"Loaded {row_count} rows into DuckDB")
conn.close()
if __name__ == "__main__":
print("=" * 50)
print("WEATHER ETL PIPELINE")
print("=" * 50)
print("\n[1/3] Extracting data from Open-Meteo API...")
raw_data = extract(CITIES)
print("\n[2/3] Transforming data...")
df = transform(raw_data)
print("\n[3/3] Loading data into DuckDB...")
load(df)
print("\nPipeline complete!")Your ETL pipeline is fully connected. Data flows from the API, through pandas, and into a persistent DuckDB database. Next up, you'll query that data with analytical SQL to answer real questions about the weather.
Query Your Data with Analytical SQL
Your pipeline can now extract, transform, and load weather data into DuckDB. The data is sitting in a structured table, but raw rows aren't useful on their own.
The real value of an ETL pipeline comes from what you do after the data is loaded. In this step, you'll write analytical SQL queries that answer questions like "Which city is the warmest?" and "What's the rainiest day?"
In this step, get ready to:
- Write an aggregation query to compare weather metrics across cities.
- Use a window function to rank temperatures and find the hottest hour per city.
- Find the rainiest day across all cities and print a pipeline summary.
Write the aggregation query
Aggregation queries collapse many rows into summaries. You'll use AVG, SUM, and MAX to compare weather patterns across New York, London, and Tokyo.
- In etl_pipeline.py, find the load() function. Below it (after its closing line), add the start of an analyze() function with the first query by copying this code:
# === ANALYZE ===
def analyze(db_path: str = "weather.duckdb") -> None:
"""Run analytical queries on the loaded data."""
conn = duckdb.connect(db_path)
# Query 1: Average temperature, total precipitation, max wind speed per city
print("\n--- Average Temperature, Total Precipitation, Max Wind per City ---")
result = conn.execute(
"""
SELECT
city,
ROUND(AVG(temperature_2m), 1) AS avg_temp_c,
ROUND(SUM(precipitation), 1) AS total_precip_mm,
ROUND(MAX(wind_speed_10m), 1) AS max_wind_kmh
FROM weather_forecasts
GROUP BY city
ORDER BY avg_temp_c DESC
"""
).fetchdf()
print(result.to_string(index=False))What does this code do?
- analyze() opens a connection to the same weather.duckdb file your pipeline loaded data into.
- GROUP BY city collapses all hourly rows for each city into a single summary row.
- .fetchdf() returns the query result as a pandas DataFrame, which makes printing a neat table easy.
- Save etl_pipeline.py.
Seeing a syntax error?
- Make sure the new analyze() function is defined at the same indentation level as load() (no extra indentation).
- Check that the triple-quoted SQL string uses matching triple quotes at the start and end.
Why is my analyze function causing an error?
Add the window function query
Window functions let you rank rows within groups without collapsing them. You'll use RANK() to find the single hottest hour recorded in each city.
- In etl_pipeline.py, add the following code directly below the last print(result.to_string(index=False)) line inside your analyze() function:
# Query 2: Hottest hour per city using window function
print("\n--- Hottest Hour per City (Window Function) ---")
result = conn.execute(
"""
SELECT city, timestamp, temperature_2m
FROM (
SELECT
city,
timestamp,
temperature_2m,
RANK() OVER (PARTITION BY city ORDER BY temperature_2m DESC) AS temp_rank
FROM weather_forecasts
)
WHERE temp_rank = 1
"""
).fetchdf()
print(result.to_string(index=False))What does this query do?
RANK() OVER (PARTITION BY city ORDER BY temperature_2m DESC) assigns a rank number to every hour within each city, ordered by temperature from highest to lowest. Rank 1 is the hottest hour.
The outer query filters to only rows where temp_rank = 1, giving you the peak temperature moment for each city.
- Save etl_pipeline.py.
Getting an indentation error?
- This code must be indented one level inside the analyze() function body. Each line should have 4 spaces of indentation.
- Check that it sits below the first query's print statement, not outside the function.
Help me fix my indentation in the analyze function.
Find the rainiest day and print a pipeline summary
Your final query groups data by date to find which day had the most rain across all three cities. You'll also print a summary showing the full scope of data your pipeline processed.
- In etl_pipeline.py, add the following code below the window function query's print(result.to_string(index=False)) line, still inside analyze():
# Query 3: Rainiest day across all cities
print("\n--- Rainiest Day Across All Cities ---")
result = conn.execute(
"""
SELECT
date,
ROUND(SUM(precipitation), 2) AS total_precip_mm,
COUNT(DISTINCT city) AS cities_reporting
FROM weather_forecasts
GROUP BY date
ORDER BY total_precip_mm DESC
LIMIT 5
"""
).fetchdf()
print(result.to_string(index=False))
# Summary
summary = conn.execute(
"""
SELECT
COUNT(*) AS total_rows,
COUNT(DISTINCT city) AS cities,
MIN(date) AS start_date,
MAX(date) AS end_date
FROM weather_forecasts
"""
).fetchone()
print(f"\n--- Pipeline Summary ---")
print(f"Total rows: {summary[0]}")
print(f"Cities: {summary[1]}")
print(f"Date range: {summary[2]} to {summary[3]}")
conn.close()What does this code do?
- The rainiest-day query groups all hourly rows by date, sums precipitation, and shows the top 5 wettest days.
- .fetchone() returns a single row (the summary), which you unpack by index to print total rows, city count, and date range.
- conn.close() releases the database connection after all queries are done.
Now you need to call analyze() from the main pipeline. You'll also update the print statements to show the full pipeline flow.
- Scroll down to the if __name__ == "__main__": block at the bottom of your file.
- Find this line:
print("\nPipeline complete!")- Replace that line with the following code:
print("\n[ANALYSIS] Running analytical queries...")
analyze()
print("\n" + "=" * 50)
print("Pipeline complete!")
print("=" * 50)What's happening here?
This adds a fourth stage to your pipeline. After extracting, transforming, and loading, the pipeline now automatically runs your analytical queries and prints results.
- Save etl_pipeline.py.
- Run the full pipeline by running this command in your terminal:
python etl_pipeline.py- You should see the extraction, transformation, and loading steps run, followed by three formatted tables and a pipeline summary showing the total rows, cities, and date range.
Not seeing any query output?
- Check that you called analyze() inside the if __name__ == "__main__": block. Without this call, the function exists but never runs.
- Make sure conn.close() is the last line inside the analyze() function, not after it.
- If you see an error about the table not existing, confirm that load(df) runs before analyze() in your main block.
My pipeline runs but I don't see query results. Help me debug.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
import requests
import pandas as pd
import duckdb
from datetime import datetime
# === CONFIGURATION ===
CITIES = {
"New York": {"latitude": 40.7128, "longitude": -74.0060},
"London": {"latitude": 51.5074, "longitude": -0.1278},
"Tokyo": {"latitude": 35.6762, "longitude": 139.6503},
}
API_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_VARIABLES = "temperature_2m,precipitation,wind_speed_10m,relative_humidity_2m"
# === EXTRACT ===
def extract(cities: dict) -> list[dict]:
"""Extract weather forecast data from Open-Meteo API for multiple cities."""
raw_data = []
for city_name, coords in cities.items():
params = {
"latitude": coords["latitude"],
"longitude": coords["longitude"],
"hourly": HOURLY_VARIABLES,
"timezone": "auto",
}
response = requests.get(API_URL, params=params, timeout=30)
response.raise_for_status()
data = response.json()
data["city_name"] = city_name
raw_data.append(data)
print(f"Extracted data for {city_name}: {len(data['hourly']['time'])} hours")
return raw_data
# === TRANSFORM ===
def transform(raw_data: list[dict]) -> pd.DataFrame:
"""Transform raw API responses into a clean pandas DataFrame."""
dataframes = []
for data in raw_data:
city_name = data["city_name"]
hourly = data["hourly"]
df = pd.DataFrame(
{
"timestamp": hourly["time"],
"temperature_2m": hourly["temperature_2m"],
"precipitation": hourly["precipitation"],
"wind_speed_10m": hourly["wind_speed_10m"],
"relative_humidity_2m": hourly["relative_humidity_2m"],
}
)
df["city"] = city_name
dataframes.append(df)
combined = pd.concat(dataframes, ignore_index=True)
combined["timestamp"] = pd.to_datetime(combined["timestamp"])
combined["date"] = combined["timestamp"].dt.date
combined["hour"] = combined["timestamp"].dt.hour
combined["feels_cold"] = combined["temperature_2m"] < 10
combined = combined[
[
"city",
"timestamp",
"date",
"hour",
"temperature_2m",
"precipitation",
"wind_speed_10m",
"relative_humidity_2m",
"feels_cold",
]
]
print(f"Transformed data: {combined.shape[0]} rows, {combined.shape[1]} columns")
return combined
# === LOAD ===
def load(df: pd.DataFrame, db_path: str = "weather.duckdb") -> None:
"""Load the transformed DataFrame into DuckDB."""
conn = duckdb.connect(db_path)
conn.execute("DROP TABLE IF EXISTS weather_forecasts")
conn.execute(
"""
CREATE TABLE weather_forecasts (
city VARCHAR,
timestamp TIMESTAMP,
date DATE,
hour INTEGER,
temperature_2m DOUBLE,
precipitation DOUBLE,
wind_speed_10m DOUBLE,
relative_humidity_2m DOUBLE,
feels_cold BOOLEAN
)
"""
)
conn.execute("INSERT INTO weather_forecasts SELECT * FROM df")
row_count = conn.execute("SELECT COUNT(*) FROM weather_forecasts").fetchone()[0]
print(f"Loaded {row_count} rows into DuckDB")
conn.close()
# === ANALYZE ===
def analyze(db_path: str = "weather.duckdb") -> None:
"""Run analytical queries on the loaded data."""
conn = duckdb.connect(db_path)
print("\n--- Average Temperature, Total Precipitation, Max Wind per City ---")
result = conn.execute(
"""
SELECT
city,
ROUND(AVG(temperature_2m), 1) AS avg_temp_c,
ROUND(SUM(precipitation), 1) AS total_precip_mm,
ROUND(MAX(wind_speed_10m), 1) AS max_wind_kmh
FROM weather_forecasts
GROUP BY city
ORDER BY avg_temp_c DESC
"""
).fetchdf()
print(result.to_string(index=False))
print("\n--- Hottest Hour per City (Window Function) ---")
result = conn.execute(
"""
SELECT city, timestamp, temperature_2m
FROM (
SELECT
city,
timestamp,
temperature_2m,
RANK() OVER (PARTITION BY city ORDER BY temperature_2m DESC) AS temp_rank
FROM weather_forecasts
)
WHERE temp_rank = 1
"""
).fetchdf()
print(result.to_string(index=False))
print("\n--- Rainiest Day Across All Cities ---")
result = conn.execute(
"""
SELECT
date,
ROUND(SUM(precipitation), 2) AS total_precip_mm,
COUNT(DISTINCT city) AS cities_reporting
FROM weather_forecasts
GROUP BY date
ORDER BY total_precip_mm DESC
LIMIT 5
"""
).fetchdf()
print(result.to_string(index=False))
summary = conn.execute(
"""
SELECT
COUNT(*) AS total_rows,
COUNT(DISTINCT city) AS cities,
MIN(date) AS start_date,
MAX(date) AS end_date
FROM weather_forecasts
"""
).fetchone()
print(f"\n--- Pipeline Summary ---")
print(f"Total rows: {summary[0]}")
print(f"Cities: {summary[1]}")
print(f"Date range: {summary[2]} to {summary[3]}")
conn.close()
# === RUN PIPELINE ===
if __name__ == "__main__":
print("=" * 50)
print("WEATHER ETL PIPELINE")
print("=" * 50)
print("\n[1/3] Extracting data from Open-Meteo API...")
raw_data = extract(CITIES)
print("\n[2/3] Transforming data...")
df = transform(raw_data)
print("\n[3/3] Loading data into DuckDB...")
load(df)
print("\n[ANALYSIS] Running analytical queries...")
analyze()
print("\n" + "=" * 50)
print("Pipeline complete!")
print("=" * 50)Your ETL pipeline is complete. You can extract live data, transform it, load it into an analytical database, and query it with real SQL. Next up, the Secret Mission challenges you to add scheduling and incremental loading.
Secret mission
Add Scheduling and Incremental Loading
Your pipeline works great for a one-off run. But real data engineering pipelines run continuously. Can you add a scheduling loop that re-runs the pipeline every 60 seconds and only inserts new data, avoiding duplicates?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources to continue experimenting, pause them to come back later, or delete them entirely.
Resources you used:
- weather-etl project folder (contains etl_pipeline.py, weather.duckdb, and the venv virtual environment).
- Python packages (requests, pandas, duckdb) installed inside the virtual environment.
Keep everything running
No action needed. Choose this if you're still actively experimenting with your ETL pipeline or want to add more cities and queries.
- Your weather-etl folder stays on your Desktop with all files intact.
- You can reactivate your virtual environment and run the pipeline anytime.
Pause - I'll come back to this later
Stop the running pipeline and deactivate the virtual environment to free up any background processes.
- If your pipeline is still running in the terminal, press Ctrl+C to stop it.
- Deactivate your virtual environment by running:
deactivateYour files and database remain on disk. To resume later, navigate back to weather-etl and reactivate the virtual environment.
Delete - I don't want to use this again
Remove all project files, the virtual environment, and the database. This cannot be undone.
- If your pipeline is still running, press Ctrl+C to stop it.
- Deactivate your virtual environment by running:
deactivate- Navigate to your Desktop and delete the entire project folder.
macOS
cd ~/Desktop
rm -rf weather-etlWindows
cd ~\Desktop
rmdir /s /q weather-etlThis removes etl_pipeline.py, weather.duckdb, the venv folder, and all installed packages inside it.
Nice Work!
Nice Work!
You just built a complete weather data ETL pipeline from scratch. You pulled live forecast data from an API, cleaned it with pandas, loaded it into an analytical database, and queried it with SQL.
Here's what you learned:
- Extracted live weather data from the Open-Meteo API using Python's requests library, handling multiple cities with error handling and timeouts.
- Transformed raw JSON into clean, structured pandas DataFrames with parsed timestamps, extracted date and hour columns, and computed fields.
- Loaded data into DuckDB and ran analytical SQL queries using aggregations and window functions to answer real questions about the data.
- Secret Mission: Added scheduling and incremental loading to re-run the pipeline automatically while avoiding duplicate rows.
Ready to quiz yourself?