Build a Weather Data ETL Pipeline
Extract, transform, and load live weather data into MySQL with Python.
Introduction
30 Second Summary
Every weather app, logistics tracker, and energy grid relies on pipelines that pull raw data from external sources, clean it up, and store it somewhere useful. This pattern of extracting, transforming, and loading data is the backbone of modern data engineering.
In this project, you will build a Python ETL pipeline that pulls 7-day weather forecasts from a live API, transforms the raw JSON into clean records, and loads them into a MySQL database.
What You'll Build
A Python ETL pipeline that extracts weather forecasts from the Open-Meteo API for multiple cities, transforms the data into structured records with derived fields, and loads them into MySQL using idempotent upsert logic.
By the end of this project, you'll have:
- A working Python script that orchestrates a full Extract-Transform-Load flow against a live weather API.
- A MySQL database with a normalized schema and unique constraints that support idempotent loading.
- Transformation logic that computes derived fields like temperature category and comfort index from raw data.
- Secret Mission: a second aggregation stage that computes daily summaries and loads them into a dedicated summary table.
Are there any prerequisites?
You should be comfortable with basic Python (variables, functions, loops, dictionaries) and SQL fundamentals (CREATE TABLE, INSERT, SELECT). No prior experience with ETL pipelines or APIs is required.
Design Your Database Schema
Your environment is ready. Now you need a place to store the weather data your pipeline will collect.
Every ETL pipeline needs a destination. In this step, you will design a MySQL database schema that stores weather forecasts cleanly and supports re-running the pipeline without creating duplicate rows.
In this step, get ready to:
- Design a forecasts table with columns for weather data, derived fields, and a unique constraint.
- Create and execute a SQL schema file against MySQL.
- Verify the table structure with DESCRIBE.
Design your forecasts table
Your pipeline will extract hourly weather forecasts for multiple cities. Each forecast row needs to store which city it belongs to, the timestamp, and the weather measurements.
The critical design decision is a unique constraint on the combination of city name and forecast datetime. This prevents duplicate rows when you re-run the pipeline.
- In VS Code, create a new file called schema.sql inside your weather-etl folder.
- Add the following SQL to create the database and the forecasts table:
CREATE DATABASE IF NOT EXISTS weather_db;
USE weather_db;
CREATE TABLE IF NOT EXISTS forecasts (
id INT AUTO_INCREMENT PRIMARY KEY,
city_name VARCHAR(100) NOT NULL,
latitude DECIMAL(6,4) NOT NULL,
longitude DECIMAL(7,4) NOT NULL,
forecast_datetime DATETIME NOT NULL,
temperature_c DECIMAL(5,2) NOT NULL,
relative_humidity INT NOT NULL,
wind_speed_kmh DECIMAL(5,2) NOT NULL,
weather_code INT NOT NULL,
temperature_category VARCHAR(10) NOT NULL,
comfort_index DECIMAL(5,2) NOT NULL,
loaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uq_city_forecast (city_name, forecast_datetime)
);What does this schema do?
- CREATE DATABASE IF NOT EXISTS creates the weather_db database only if it does not already exist. This makes the script safe to re-run.
- The forecasts table stores one row per city per forecast hour. Columns cover location, weather measurements, and two derived fields (temperature_category and comfort_index) that your Python code will compute later.
- UNIQUE KEY uq_city_forecast (city_name, forecast_datetime) enforces that the same city and timestamp can only appear once. This is what enables idempotent loading.
- loaded_at automatically updates whenever a row is inserted or modified, giving you an audit trail of when data was last refreshed.
What is idempotent loading?
An idempotent operation produces the same result no matter how many times you run it. For a data pipeline, this means running it twice does not create duplicate rows. The unique constraint combined with INSERT ... ON DUPLICATE KEY UPDATE (which you will write later) achieves this.
- Save schema.sql.
File not saving?
Make sure you created the file inside the weather-etl folder, not at a higher level. Check the VS Code Explorer sidebar to confirm schema.sql appears at the top level of your project.
Help me find where my file was saved in VS Code.
Execute the schema against MySQL
The schema file is a reproducible record of your database design. Now you need to execute it so MySQL actually creates the database and table.
- In your VS Code terminal, make sure you are in the weather-etl folder (where schema.sql lives).
- Execute the schema file against MySQL by running this command:
mysql -u root -p < schema.sql- When prompted, enter the root password you set during MySQL installation.
If the command succeeds, you will see no output. MySQL only prints messages when something goes wrong.
Seeing 'mysql' is not recognized?
This means MySQL is not in your system PATH. You can fix this by using the full path to the mysql executable. The default location on Windows is C:\Program Files\MySQL\MySQL Server 9.7\bin\mysql.exe.
Help me add MySQL to my system PATH on Windows.
Verify the table was created
A quick check confirms the table exists with the correct columns and unique key.
- Connect to your weather_db database by running:
mysql -u root -p weather_db- Enter your root password when prompted.
- Run the following SQL command inside the MySQL shell:
DESCRIBE forecasts;You should see a table listing all 12 columns: id, city_name, latitude, longitude, forecast_datetime, temperature_c, relative_humidity, wind_speed_kmh, weather_code, temperature_category, comfort_index, and loaded_at. The Key column should show MUL next to city_name, confirming the unique key is in place.
- Exit the MySQL shell by typing:
EXIT;Table not showing up?
If you see ERROR 1146: Table 'weather_db.forecasts' doesn't exist, the schema file did not execute properly. Go back and re-run the mysql -u root -p < schema.sql command. Watch for any error messages this time.
Help me debug why my schema.sql did not create the forecasts table.
✔️ Awesome, I've got everything!
Great. Make sure you have saved schema.sql and confirmed the table exists in MySQL.
ⓧ I'd like to double check the full code
Here is the complete schema.sql file:
CREATE DATABASE IF NOT EXISTS weather_db;
USE weather_db;
CREATE TABLE IF NOT EXISTS forecasts (
id INT AUTO_INCREMENT PRIMARY KEY,
city_name VARCHAR(100) NOT NULL,
latitude DECIMAL(6,4) NOT NULL,
longitude DECIMAL(7,4) NOT NULL,
forecast_datetime DATETIME NOT NULL,
temperature_c DECIMAL(5,2) NOT NULL,
relative_humidity INT NOT NULL,
wind_speed_kmh DECIMAL(5,2) NOT NULL,
weather_code INT NOT NULL,
temperature_category VARCHAR(10) NOT NULL,
comfort_index DECIMAL(5,2) NOT NULL,
loaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uq_city_forecast (city_name, forecast_datetime)
);Your database is ready to receive weather data. Next up, you will write the Python code that pulls live forecast data from an API and feeds it into this table.
Extract Data from the Open-Meteo API
Your database is ready and waiting for data. Now you need to fill it with real weather forecasts from a live source.
The "E" in ETL stands for Extract. In this step, you will pull 7-day hourly forecasts from the Open-Meteo API for three cities. Open-Meteo is a free weather API that requires no signup and no API key.
In this step, get ready to:
- Create the ETL pipeline script with configuration and logging.
- Write the extract() function that calls the Open-Meteo API.
- Test the extraction by running the script and inspecting the response structure.
Create the pipeline script with configuration
Every ETL pipeline needs a central script to orchestrate the flow. You will create etl_pipeline.py and set up the imports, database credentials, city list, and API parameters that the rest of the pipeline depends on.
- In VS Code's Explorer sidebar, right-click inside the weather-etl folder and select New File.
- Name the file etl_pipeline.py and press Enter.
- Add the imports and configuration section by pasting this code:
import requests
import mysql.connector
import logging
from datetime import datetime
# --- Configuration ---
DB_CONFIG = {
"host": "localhost",
"user": "root",
"password": "your-mysql-password-here",
"database": "weather_db",
}
CITIES = [
{"name": "London", "latitude": 51.5074, "longitude": -0.1278},
{"name": "New York", "latitude": 40.7128, "longitude": -74.0060},
{"name": "Tokyo", "latitude": 35.6762, "longitude": 139.6503},
]
API_BASE_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_PARAMS = "temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code"
# --- Logging Setup ---
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)What does this code do?
- requests handles HTTP calls to the weather API. mysql.connector connects to your database. logging provides structured output instead of raw print statements.
- DB_CONFIG stores the MySQL connection details. Replace your-mysql-password-here with the root password you set during MySQL installation.
- CITIES is the list of cities to fetch forecasts for. Each entry includes the city name and its GPS coordinates.
- API_BASE_URL points to the Open-Meteo forecast endpoint. HOURLY_PARAMS tells the API which weather variables to return (temperature, humidity, wind speed, and weather code).
- The logging setup formats log messages with timestamps and severity levels so you can trace each pipeline run.
- Replace your-mysql-password-here in the DB_CONFIG dictionary with the root password you configured in Step 1.
- Save etl_pipeline.py.
You will not see any output yet. The next chunk adds the function that uses this configuration.
Seeing a red underline on the imports?
If VS Code shows an import error for requests or mysql.connector, make sure your virtual environment is selected as the Python interpreter. Press Ctrl+Shift+P, type Python: Select Interpreter, and choose the one that includes venv in its path.
Help me fix my import errors in VS Code
Write the extract function
The extract function is the first stage of your pipeline. It loops through the city list, calls the Open-Meteo API for each one, and collects the raw JSON responses.
- Add the following extract() function below the logging setup in etl_pipeline.py:
# --- Extract ---
def extract(cities):
"""Extract weather forecast data from Open-Meteo API for each city."""
logger.info("Starting extraction for %d cities", len(cities))
raw_data = []
for city in cities:
params = {
"latitude": city["latitude"],
"longitude": city["longitude"],
"hourly": HOURLY_PARAMS,
"timezone": "auto",
}
response = requests.get(API_BASE_URL, params=params)
response.raise_for_status()
data = response.json()
data["_city_name"] = city["name"]
raw_data.append(data)
logger.info("Extracted data for %s", city["name"])
logger.info("Extraction complete: %d API responses collected", len(raw_data))
return raw_dataWhat does this code do?
- The function accepts a list of city dictionaries and returns a list of raw API response dictionaries.
- For each city, it builds a params dictionary with the city's coordinates, the hourly variables to fetch, and timezone: "auto" which tells Open-Meteo to return times in each city's local timezone.
- requests.get() sends the HTTP request. raise_for_status() throws an error if the API returns a non-200 status code, so bad responses fail loudly instead of silently.
- data["_city_name"] = city["name"] attaches the city name directly to the response dictionary. The API does not include the city name in its response, so this makes it easy to track which response belongs to which city during the transform step.
- Save etl_pipeline.py.
You will not see output yet. In the next substep, you will test this function to confirm it pulls real data.
Not sure where to paste this code?
The extract() function goes directly after the logger = logging.getLogger(__name__) line. Leave one blank line between the logger line and the # --- Extract --- comment.
Help me understand where to place the extract function
✔️ Awesome, I've got everything!
Great. Double check you have saved etl_pipeline.py before moving on.
ⓧ I'd like to double check the full code
Here is the complete etl_pipeline.py so far:
import requests
import mysql.connector
import logging
from datetime import datetime
# --- Configuration ---
DB_CONFIG = {
"host": "localhost",
"user": "root",
"password": "your-mysql-password-here",
"database": "weather_db",
}
CITIES = [
{"name": "London", "latitude": 51.5074, "longitude": -0.1278},
{"name": "New York", "latitude": 40.7128, "longitude": -74.0060},
{"name": "Tokyo", "latitude": 35.6762, "longitude": 139.6503},
]
API_BASE_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_PARAMS = "temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code"
# --- Logging Setup ---
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# --- Extract ---
def extract(cities):
"""Extract weather forecast data from Open-Meteo API for each city."""
logger.info("Starting extraction for %d cities", len(cities))
raw_data = []
for city in cities:
params = {
"latitude": city["latitude"],
"longitude": city["longitude"],
"hourly": HOURLY_PARAMS,
"timezone": "auto",
}
response = requests.get(API_BASE_URL, params=params)
response.raise_for_status()
data = response.json()
data["_city_name"] = city["name"]
raw_data.append(data)
logger.info("Extracted data for %s", city["name"])
logger.info("Extraction complete: %d API responses collected", len(raw_data))
return raw_dataTest the extraction
Before building the rest of the pipeline, you should verify that the extraction actually pulls live data. You will add temporary test code to call the function and inspect what comes back.
- Add the following test code at the very bottom of etl_pipeline.py:
# --- Temporary test code (remove after testing) ---
if __name__ == "__main__":
raw_data = extract(CITIES)
print(f"\nReceived {len(raw_data)} responses")
print(f"Keys in first response: {list(raw_data[0].keys())}")
print(f"Number of hourly timestamps: {len(raw_data[0]['hourly']['time'])}")
print(f"First 3 timestamps: {raw_data[0]['hourly']['time'][:3]}")What does this code do?
The if __name__ == "__main__" guard ensures this code only runs when you execute the file directly. It calls extract() with the CITIES list and prints key details about the response structure so you can confirm the API is working.
- Save etl_pipeline.py.
- Run the script by executing this command in your VS Code terminal:
python etl_pipeline.pyYou should see log messages for each city extraction, followed by output showing 3 responses, a list of keys including hourly and _city_name, and 168 hourly timestamps (7 days of hourly data).
What should I see?
Your output will look similar to this (with today's dates):
2026-05-19 10:00:00 - INFO - Starting extraction for 3 cities
2026-05-19 10:00:01 - INFO - Extracted data for London
2026-05-19 10:00:01 - INFO - Extracted data for New York
2026-05-19 10:00:01 - INFO - Extracted data for Tokyo
2026-05-19 10:00:01 - INFO - Extraction complete: 3 API responses collected
Received 3 responses
Keys in first response: ['latitude', 'longitude', 'elevation', ..., 'hourly', '_city_name']
Number of hourly timestamps: 168
First 3 timestamps: ['2026-05-19T00:00', '2026-05-19T01:00', '2026-05-19T02:00']The 168 timestamps represent 7 days of hourly forecasts (24 hours x 7 days). The hourly object contains parallel arrays for time, temperature_2m, relative_humidity_2m, wind_speed_10m, and weather_code. Each array has the same length, so index 0 across all arrays describes the same hour.
Seeing a ConnectionError or timeout?
This means your machine cannot reach the Open-Meteo API. Check that you have an active internet connection. If you are behind a corporate firewall or VPN, try disconnecting it temporarily.
If you see an HTTP error with a 400 status, double-check that HOURLY_PARAMS matches the value in the code above exactly. A typo in a parameter name causes the API to reject the request.
Help me troubleshoot my connection error
- Now that you have confirmed extraction works, remove the temporary test block from the bottom of etl_pipeline.py. Delete everything from the # --- Temporary test code comment through the last print() line.
- Save etl_pipeline.py.
Why remove the test code?
You will add a proper main() function later that orchestrates the full ETL flow. Keeping temporary test code around clutters the file and can cause unexpected behavior when you run the script in later steps.
Your extraction stage is pulling live forecast data for three cities across the globe. Next up, you will transform that raw JSON into clean, structured records ready for your database.
Transform the Raw Data
Your extraction function is pulling live forecast data from the Open-Meteo API. The raw response comes back as nested parallel arrays, with one array for timestamps, another for temperatures, another for humidity, and so on.
The Transform stage reshapes this raw data into clean, flat records that map directly to your database columns. You will also add derived fields that enrich the data beyond what the API provides.
In this step, get ready to:
- Write helper functions that compute derived fields (temperature category and comfort index).
- Build a transform function that flattens nested API arrays into structured records.
- Add data validation to filter out records with missing values.
Add helper functions for derived fields
Before writing the main transform function, you need two small helpers it will call. These add enriched data that the raw API does not provide.
- In etl_pipeline.py, scroll below the extract() function.
- Add a # --- Transform --- section comment, then paste the following function below it:
# --- Transform ---
def categorize_temperature(temp_c):
"""Assign a temperature category based on Celsius value."""
if temp_c < 10:
return "cold"
elif temp_c < 20:
return "mild"
elif temp_c < 30:
return "warm"
else:
return "hot"What does this code do?
This helper takes a numeric temperature in Celsius and returns a human-readable category. The thresholds are 10, 20, and 30 degrees.
Your database schema has a temperature_category column. This function generates that value for every forecast record.
- Directly below categorize_temperature(), add the comfort index helper by pasting this function:
def compute_comfort_index(temp_c, humidity):
"""Compute a simple comfort index (0-100). Lower humidity and moderate temps score higher."""
temp_score = max(0, 50 - abs(temp_c - 22) * 2.5)
humidity_score = max(0, 50 - abs(humidity - 40) * 0.5)
return round(temp_score + humidity_score, 2)What does this code do?
The comfort index combines temperature and humidity into a single 0-100 score. A temperature of 22°C and humidity of 40% scores highest (100). Deviations in either direction reduce the score.
This is a derived field. The raw API gives you temperature and humidity separately. Your pipeline enriches the data by computing a value that neither source provides on its own.
- Save etl_pipeline.py.
Seeing a syntax error?
Check that both functions are at the top level of the file (no extra indentation). They should align with the def extract(cities): function above.
help me fix a syntax error in my categorize_temperature or compute_comfort_index function
Write the transform function
The Open-Meteo API returns hourly data as parallel arrays. The time array holds timestamps, temperature_2m holds temperature values, and so on. Each index position across all arrays represents the same hour.
Your transform function zips these arrays together and produces one flat dictionary per forecast hour. It also filters out records with missing data and calls your helper functions.
- Below compute_comfort_index(), add the transform function by pasting this code:
def transform(raw_data):
"""Transform raw API responses into clean, flat records."""
logger.info("Starting transformation")
records = []
for response in raw_data:
city_name = response["_city_name"]
latitude = response["latitude"]
longitude = response["longitude"]
hourly = response["hourly"]
times = hourly["time"]
temperatures = hourly["temperature_2m"]
humidities = hourly["relative_humidity_2m"]
wind_speeds = hourly["wind_speed_10m"]
weather_codes = hourly["weather_code"]
for i in range(len(times)):
temp = temperatures[i]
humidity = humidities[i]
wind = wind_speeds[i]
code = weather_codes[i]
# Skip records with missing data
if temp is None or humidity is None or wind is None:
continue
record = {
"city_name": city_name,
"latitude": latitude,
"longitude": longitude,
"forecast_datetime": datetime.fromisoformat(times[i]),
"temperature_c": temp,
"relative_humidity": humidity,
"wind_speed_kmh": wind,
"weather_code": code if code is not None else 0,
"temperature_category": categorize_temperature(temp),
"comfort_index": compute_comfort_index(temp, humidity),
}
records.append(record)
logger.info("Transformation complete: %d records produced", len(records))
return recordsWhat does this code do?
- The outer loop iterates over each city's API response. It pulls out the city metadata and the hourly object containing parallel arrays.
- The inner loop walks through each index position, pulling the corresponding value from each array to build one flat record per forecast hour.
- datetime.fromisoformat() parses the ISO 8601 time strings (like 2026-05-19T14:00) into Python datetime objects that MySQL can store directly.
- The if temp is None check skips any record where critical data is missing. APIs sometimes return None for hours where sensor data was unavailable.
- Each record calls categorize_temperature() and compute_comfort_index() to add derived fields before appending.
- Save etl_pipeline.py.
Seeing an indentation error?
Make sure the for i in range(len(times)): loop is indented inside the outer for response in raw_data: loop. The record = { dictionary must be indented inside the inner loop.
help me fix an indentation error in my transform function
Verify the data validation logic
The transform() function silently skips records where temperature, humidity, or wind speed is None. This is your data quality gate.
Why filter out None values?
Live APIs sometimes return null values for hours where weather station data was unavailable or a model failed to produce a prediction. Loading None into a NOT NULL database column would crash your pipeline.
Filtering early in the Transform stage means the Load stage can trust that every record is complete. This is a common pattern in production pipelines.
- Confirm the validation line in your transform() function reads:
# Skip records with missing data
if temp is None or humidity is None or wind is None:
continue- Run a quick test to confirm your transform function works. Add these lines at the bottom of etl_pipeline.py temporarily:
# Temporary test - remove after verifying
if __name__ == "__main__":
raw_data = extract(CITIES)
records = transform(raw_data)
print(f"Produced {len(records)} records")
print(f"Sample record: {records[0]}")
- Run your script by executing this command in the terminal:
python etl_pipeline.py- You should see log messages for extraction, then a Produced count (around 504 records for 3 cities with 7 days of hourly data) and a sample record dictionary with all your keys.
- After confirming the output, delete the temporary test lines you just added (the if __name__ block at the bottom).
- Save etl_pipeline.py.
Seeing a KeyError or AttributeError?
A KeyError means the API response structure does not match what your code expects. Check that hourly["temperature_2m"] matches the parameter name temperature_2m exactly.
help me debug a KeyError or AttributeError in my transform function when processing Open-Meteo API data
✔️ Awesome, I've got everything!
Great. Double check you have saved etl_pipeline.py with the temporary test code removed.
ⓧ I'd like to double check the full code
Here is the full etl_pipeline.py as it should look at the end of this step:
import requests
import mysql.connector
import logging
from datetime import datetime
# --- Configuration ---
DB_CONFIG = {
"host": "localhost",
"user": "root",
"password": "your-mysql-password-here",
"database": "weather_db",
}
CITIES = [
{"name": "London", "latitude": 51.5074, "longitude": -0.1278},
{"name": "New York", "latitude": 40.7128, "longitude": -74.0060},
{"name": "Tokyo", "latitude": 35.6762, "longitude": 139.6503},
]
API_BASE_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_PARAMS = "temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code"
# --- Logging Setup ---
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# --- Extract ---
def extract(cities):
"""Extract weather forecast data from Open-Meteo API for each city."""
logger.info("Starting extraction for %d cities", len(cities))
raw_data = []
for city in cities:
params = {
"latitude": city["latitude"],
"longitude": city["longitude"],
"hourly": HOURLY_PARAMS,
"timezone": "auto",
}
response = requests.get(API_BASE_URL, params=params)
response.raise_for_status()
data = response.json()
data["_city_name"] = city["name"]
raw_data.append(data)
logger.info("Extracted data for %s", city["name"])
logger.info("Extraction complete: %d API responses collected", len(raw_data))
return raw_data
# --- Transform ---
def categorize_temperature(temp_c):
"""Assign a temperature category based on Celsius value."""
if temp_c < 10:
return "cold"
elif temp_c < 20:
return "mild"
elif temp_c < 30:
return "warm"
else:
return "hot"
def compute_comfort_index(temp_c, humidity):
"""Compute a simple comfort index (0-100). Lower humidity and moderate temps score higher."""
temp_score = max(0, 50 - abs(temp_c - 22) * 2.5)
humidity_score = max(0, 50 - abs(humidity - 40) * 0.5)
return round(temp_score + humidity_score, 2)
def transform(raw_data):
"""Transform raw API responses into clean, flat records."""
logger.info("Starting transformation")
records = []
for response in raw_data:
city_name = response["_city_name"]
latitude = response["latitude"]
longitude = response["longitude"]
hourly = response["hourly"]
times = hourly["time"]
temperatures = hourly["temperature_2m"]
humidities = hourly["relative_humidity_2m"]
wind_speeds = hourly["wind_speed_10m"]
weather_codes = hourly["weather_code"]
for i in range(len(times)):
temp = temperatures[i]
humidity = humidities[i]
wind = wind_speeds[i]
code = weather_codes[i]
# Skip records with missing data
if temp is None or humidity is None or wind is None:
continue
record = {
"city_name": city_name,
"latitude": latitude,
"longitude": longitude,
"forecast_datetime": datetime.fromisoformat(times[i]),
"temperature_c": temp,
"relative_humidity": humidity,
"wind_speed_kmh": wind,
"weather_code": code if code is not None else 0,
"temperature_category": categorize_temperature(temp),
"comfort_index": compute_comfort_index(temp, humidity),
}
records.append(record)
logger.info("Transformation complete: %d records produced", len(records))
return recordsYour transformation layer is ready to clean and reshape raw API data. Next up, you will load these records into MySQL and see them appear in your database.
Load Data into MySQL
You have clean, structured records flowing out of your transform() function. Now you need somewhere to put them.
The "L" in ETL is where your data stops being transient Python objects and becomes durable, queryable rows in a database. In this step, you will write the load function, wire up the full pipeline, and run it end to end.
In this step, get ready to:
- Write a load function that upserts records into MySQL.
- Add a main function that orchestrates extract, transform, and load.
- Run the pipeline and verify data appears in the database.
Write the load function with upsert logic
A production pipeline must be safe to re-run. If you run it twice, you want updated data, not duplicate rows.
MySQL's INSERT ... ON DUPLICATE KEY UPDATE syntax handles this. It inserts a new row if the unique key does not exist, and overwrites the matching row if it does.
- In etl_pipeline.py, scroll to the bottom of the file (below the transform() function) and add the following load() function:
# --- Load ---
def load(records):
"""Load transformed records into MySQL using upsert logic."""
logger.info("Starting load of %d records", len(records))
insert_query = """
INSERT INTO forecasts (
city_name, latitude, longitude, forecast_datetime,
temperature_c, relative_humidity, wind_speed_kmh,
weather_code, temperature_category, comfort_index
) VALUES (
%(city_name)s, %(latitude)s, %(longitude)s, %(forecast_datetime)s,
%(temperature_c)s, %(relative_humidity)s, %(wind_speed_kmh)s,
%(weather_code)s, %(temperature_category)s, %(comfort_index)s
) ON DUPLICATE KEY UPDATE
temperature_c = VALUES(temperature_c),
relative_humidity = VALUES(relative_humidity),
wind_speed_kmh = VALUES(wind_speed_kmh),
weather_code = VALUES(weather_code),
temperature_category = VALUES(temperature_category),
comfort_index = VALUES(comfort_index)
"""
connection = None
try:
connection = mysql.connector.connect(**DB_CONFIG)
cursor = connection.cursor()
cursor.executemany(insert_query, records)
connection.commit()
logger.info("Load complete: %d rows affected", cursor.rowcount)
except mysql.connector.Error as err:
logger.error("Database error: %s", err)
raise
finally:
if connection and connection.is_connected():
cursor.close()
connection.close()What does this code do?
The insert_query uses %(key)s placeholders that map directly to the dictionary keys in each record. ON DUPLICATE KEY UPDATE tells MySQL to overwrite the existing row when the unique constraint (city_name, forecast_datetime) matches.
The cursor.executemany() method batches all records into a single network round-trip instead of sending one INSERT per row. This is significantly faster for hundreds of records.
The try/except/finally block guarantees the database connection is always closed, even if an error occurs mid-load. This prevents connection leaks that could exhaust your MySQL connection pool.
- Save etl_pipeline.py.
Why upsert instead of plain INSERT?
Weather forecasts update constantly. If you run this pipeline every hour, a plain INSERT would fail with a duplicate key error on the second run. The upsert pattern makes your pipeline idempotent. You can run it as many times as you want and always get the freshest data without duplicates.
Seeing a NameError on mysql.connector?
Make sure the import mysql.connector line is still at the top of your file. If you accidentally deleted it, add it back below import requests.
Also confirm your virtual environment is active. You should see (venv) at the start of your terminal prompt.
Help me debug a NameError in my ETL load function.
Add the main function to orchestrate the pipeline
With extract, transform, and load each written as separate functions, you need one function to call them in sequence. This is the orchestrator.
- Below the load() function in etl_pipeline.py, add the following code:
# --- Main ---
def main():
"""Orchestrate the ETL pipeline."""
logger.info("=== ETL Pipeline Started ===")
start_time = datetime.now()
raw_data = extract(CITIES)
records = transform(raw_data)
load(records)
elapsed = (datetime.now() - start_time).total_seconds()
logger.info("=== ETL Pipeline Finished in %.2f seconds ===", elapsed)
if __name__ == "__main__":
main()What does this code do?
The main() function calls extract(), transform(), and load() in order, passing each stage's output to the next. It also logs the total elapsed time so you can monitor performance.
The if __name__ == "__main__" guard ensures main() only runs when you execute the script directly. It will not run if you import this file as a module in another script later.
- Save etl_pipeline.py.
Not sure where to add main?
The main() function and the if __name__ guard should be the very last lines in your file. Nothing should come after them.
Help me figure out where main() goes in my ETL script.
Run the pipeline and verify data in MySQL
Your pipeline is complete. Time to run it and confirm data lands in the database.
- Make sure your DB_CONFIG password matches the root password you set during MySQL installation. Open the configuration section at the top of etl_pipeline.py and update your-mysql-password-here to your actual password.
- Run the pipeline by executing this command in your VS Code terminal:
python etl_pipeline.py- You should see log output showing each phase completing. Look for a line like Load complete: ... rows affected and a final ETL Pipeline Finished message with the elapsed time.
What should I see?
Your terminal will show timestamped log lines for extraction (one per city), transformation (total record count), and loading (rows affected). The entire pipeline should finish in under 10 seconds.
- Verify the data landed in MySQL by running this command:
mysql -u root -p -e "SELECT city_name, forecast_datetime, temperature_c, temperature_category FROM weather_db.forecasts LIMIT 10;"- Enter your MySQL root password when prompted.
- You should see a table with 10 rows showing city names, forecast datetimes, temperatures, and categories.
Getting Access denied or connection refused?
Double-check that the password value in DB_CONFIG matches the root password you set during MySQL Configurator setup.
Make sure the MySQL service is running. Open Services (press the Windows key, type Services) and confirm MySQL shows a status of Running.
Help me fix a MySQL connection error in my Python ETL pipeline.
✔️ Awesome, I've got everything!
Great. Double check you have saved etl_pipeline.py and confirmed data in MySQL before moving on.
ⓧ I'd like to double check the full code
Here is the complete etl_pipeline.py at this point:
import requests
import mysql.connector
import logging
from datetime import datetime
# --- Configuration ---
DB_CONFIG = {
"host": "localhost",
"user": "root",
"password": "your-mysql-password-here",
"database": "weather_db",
}
CITIES = [
{"name": "London", "latitude": 51.5074, "longitude": -0.1278},
{"name": "New York", "latitude": 40.7128, "longitude": -74.0060},
{"name": "Tokyo", "latitude": 35.6762, "longitude": 139.6503},
]
API_BASE_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_PARAMS = "temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code"
# --- Logging Setup ---
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# --- Extract ---
def extract(cities):
"""Extract weather forecast data from Open-Meteo API for each city."""
logger.info("Starting extraction for %d cities", len(cities))
raw_data = []
for city in cities:
params = {
"latitude": city["latitude"],
"longitude": city["longitude"],
"hourly": HOURLY_PARAMS,
"timezone": "auto",
}
response = requests.get(API_BASE_URL, params=params)
response.raise_for_status()
data = response.json()
data["_city_name"] = city["name"]
raw_data.append(data)
logger.info("Extracted data for %s", city["name"])
logger.info("Extraction complete: %d API responses collected", len(raw_data))
return raw_data
# --- Transform ---
def categorize_temperature(temp_c):
"""Assign a temperature category based on Celsius value."""
if temp_c < 10:
return "cold"
elif temp_c < 20:
return "mild"
elif temp_c < 30:
return "warm"
else:
return "hot"
def compute_comfort_index(temp_c, humidity):
"""Compute a simple comfort index (0-100). Lower humidity and moderate temps score higher."""
temp_score = max(0, 50 - abs(temp_c - 22) * 2.5)
humidity_score = max(0, 50 - abs(humidity - 40) * 0.5)
return round(temp_score + humidity_score, 2)
def transform(raw_data):
"""Transform raw API responses into clean, flat records."""
logger.info("Starting transformation")
records = []
for response in raw_data:
city_name = response["_city_name"]
latitude = response["latitude"]
longitude = response["longitude"]
hourly = response["hourly"]
times = hourly["time"]
temperatures = hourly["temperature_2m"]
humidities = hourly["relative_humidity_2m"]
wind_speeds = hourly["wind_speed_10m"]
weather_codes = hourly["weather_code"]
for i in range(len(times)):
temp = temperatures[i]
humidity = humidities[i]
wind = wind_speeds[i]
code = weather_codes[i]
# Skip records with missing data
if temp is None or humidity is None or wind is None:
continue
record = {
"city_name": city_name,
"latitude": latitude,
"longitude": longitude,
"forecast_datetime": datetime.fromisoformat(times[i]),
"temperature_c": temp,
"relative_humidity": humidity,
"wind_speed_kmh": wind,
"weather_code": code if code is not None else 0,
"temperature_category": categorize_temperature(temp),
"comfort_index": compute_comfort_index(temp, humidity),
}
records.append(record)
logger.info("Transformation complete: %d records produced", len(records))
return records
# --- Load ---
def load(records):
"""Load transformed records into MySQL using upsert logic."""
logger.info("Starting load of %d records", len(records))
insert_query = """
INSERT INTO forecasts (
city_name, latitude, longitude, forecast_datetime,
temperature_c, relative_humidity, wind_speed_kmh,
weather_code, temperature_category, comfort_index
) VALUES (
%(city_name)s, %(latitude)s, %(longitude)s, %(forecast_datetime)s,
%(temperature_c)s, %(relative_humidity)s, %(wind_speed_kmh)s,
%(weather_code)s, %(temperature_category)s, %(comfort_index)s
) ON DUPLICATE KEY UPDATE
temperature_c = VALUES(temperature_c),
relative_humidity = VALUES(relative_humidity),
wind_speed_kmh = VALUES(wind_speed_kmh),
weather_code = VALUES(weather_code),
temperature_category = VALUES(temperature_category),
comfort_index = VALUES(comfort_index)
"""
connection = None
try:
connection = mysql.connector.connect(**DB_CONFIG)
cursor = connection.cursor()
cursor.executemany(insert_query, records)
connection.commit()
logger.info("Load complete: %d rows affected", cursor.rowcount)
except mysql.connector.Error as err:
logger.error("Database error: %s", err)
raise
finally:
if connection and connection.is_connected():
cursor.close()
connection.close()
# --- Main ---
def main():
"""Orchestrate the ETL pipeline."""
logger.info("=== ETL Pipeline Started ===")
start_time = datetime.now()
raw_data = extract(CITIES)
records = transform(raw_data)
load(records)
elapsed = (datetime.now() - start_time).total_seconds()
logger.info("=== ETL Pipeline Finished in %.2f seconds ===", elapsed)
if __name__ == "__main__":
main()Your pipeline is alive. It pulls live weather data, cleans it, and stores it in MySQL. Next up, you will add logging and configuration improvements to make it production-ready.
Make It Production-Ready
Your pipeline works. It extracts, transforms, and loads weather data into MySQL. But a pipeline that only works once is not production-ready.
Production pipelines need two things: observability (knowing what happened and when) and safety (running repeatedly without breaking anything). In this step, you will review the logging and configuration patterns already in your code, then prove your pipeline is idempotent by running it a second time.
In this step, get ready to:
- Review the logging setup that gives your pipeline observability.
- Confirm the configuration section keeps credentials and parameters easy to change.
- Run the pipeline a second time and verify idempotent loading works.
Review your logging setup
When a pipeline runs in production, nobody is watching the terminal. Logs are the only way to find out what happened after the fact. Python's built-in logging module gives you timestamped, leveled messages that are far more useful than plain print() calls.
- Open etl_pipeline.py in VS Code and scroll to the top of the file.
- Find the logging setup section near the top:
# --- Logging Setup ---
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)What does this logging setup do?
logging.basicConfig() sets the minimum log level to INFO and defines a format that includes a timestamp, severity level, and message. Every logger.info() call throughout the pipeline uses this format automatically.
Unlike print(), the logging module lets you filter by severity (DEBUG, INFO, WARNING, ERROR). In production, you could switch to WARNING to silence routine messages and only surface problems.
- Scroll through extract(), transform(), and load() and confirm each function logs its start and record counts.
- Check the main() function at the bottom and confirm it logs the pipeline start, end, and elapsed time.
Why log at the start and end of each phase?
If a pipeline fails at 3 AM, the first question is always "which phase broke?" Logging at the boundary of each ETL phase gives you timestamps to calculate duration and pinpoint exactly where things went wrong.
Review the configuration section and verify idempotency
A clean configuration section makes it easy to change database credentials, swap cities, or adjust API parameters without hunting through business logic.
- Scroll to the top of etl_pipeline.py and confirm the configuration section is clearly separated:
# --- Configuration ---
DB_CONFIG = {
"host": "localhost",
"user": "root",
"password": "your-mysql-password-here",
"database": "weather_db",
}
CITIES = [
{"name": "London", "latitude": 51.5074, "longitude": -0.1278},
{"name": "New York", "latitude": 40.7128, "longitude": -74.0060},
{"name": "Tokyo", "latitude": 35.6762, "longitude": 139.6503},
]
API_BASE_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_PARAMS = "temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code"Why separate configuration from logic?
All the values someone might change between environments (database credentials, city list, API endpoint) live in one place at the top. The functions below never hardcode these values. This means you could add a fourth city or change the database host without touching any function logic.
Now for the real test. A production pipeline must be safe to run multiple times. Your ON DUPLICATE KEY UPDATE clause should update existing rows instead of creating duplicates.
- First, check how many rows are currently in the forecasts table by running this in your MySQL terminal:
SELECT COUNT(*) FROM forecasts;- Note the number. You should see around 500 rows (7 days x 24 hours x 3 cities).
- Now run the pipeline a second time from your VS Code terminal:
python etl_pipeline.py- You should see timestamped log output showing each phase completing successfully.
- Go back to your MySQL terminal and run the count query again:
SELECT COUNT(*) FROM forecasts;- The row count should be the same (or very close). The pipeline updated existing rows instead of inserting duplicates.
- Verify the data was actually refreshed by checking the loaded_at timestamps:
SELECT city_name, forecast_datetime, loaded_at
FROM forecasts
ORDER BY loaded_at DESC
LIMIT 5;- The loaded_at column should show timestamps from just now, proving the rows were updated during the second run.
What is idempotent loading?
An idempotent operation produces the same result no matter how many times you run it. Your INSERT ... ON DUPLICATE KEY UPDATE statement achieves this by using the unique constraint on (city_name, forecast_datetime) to detect existing rows.
If a row with the same city and datetime already exists, MySQL updates it with the fresh values. If it does not exist, MySQL inserts a new row. This means you can schedule this pipeline to run every hour without worrying about data corruption.
Row count doubled after the second run?
If your row count roughly doubled, the unique constraint may not be in place. Run SHOW INDEX FROM forecasts; and look for uq_city_forecast. If it is missing, re-run your schema.sql file or add the constraint manually.
Help me debug why my row count doubled after re-running the pipeline.
✔️ Awesome, I've got everything!
Great. Double check you have saved etl_pipeline.py and confirmed the row count stayed stable after the second run.
ⓧ I'd like to double check the full code
import requests
import mysql.connector
import logging
from datetime import datetime
# --- Configuration ---
DB_CONFIG = {
"host": "localhost",
"user": "root",
"password": "your-mysql-password-here",
"database": "weather_db",
}
CITIES = [
{"name": "London", "latitude": 51.5074, "longitude": -0.1278},
{"name": "New York", "latitude": 40.7128, "longitude": -74.0060},
{"name": "Tokyo", "latitude": 35.6762, "longitude": 139.6503},
]
API_BASE_URL = "https://api.open-meteo.com/v1/forecast"
HOURLY_PARAMS = "temperature_2m,relative_humidity_2m,wind_speed_10m,weather_code"
# --- Logging Setup ---
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# --- Extract ---
def extract(cities):
"""Extract weather forecast data from Open-Meteo API for each city."""
logger.info("Starting extraction for %d cities", len(cities))
raw_data = []
for city in cities:
params = {
"latitude": city["latitude"],
"longitude": city["longitude"],
"hourly": HOURLY_PARAMS,
"timezone": "auto",
}
response = requests.get(API_BASE_URL, params=params)
response.raise_for_status()
data = response.json()
data["_city_name"] = city["name"]
raw_data.append(data)
logger.info("Extracted data for %s", city["name"])
logger.info("Extraction complete: %d API responses collected", len(raw_data))
return raw_data
# --- Transform ---
def categorize_temperature(temp_c):
"""Assign a temperature category based on Celsius value."""
if temp_c < 10:
return "cold"
elif temp_c < 20:
return "mild"
elif temp_c < 30:
return "warm"
else:
return "hot"
def compute_comfort_index(temp_c, humidity):
"""Compute a simple comfort index (0-100). Lower humidity and moderate temps score higher."""
temp_score = max(0, 50 - abs(temp_c - 22) * 2.5)
humidity_score = max(0, 50 - abs(humidity - 40) * 0.5)
return round(temp_score + humidity_score, 2)
def transform(raw_data):
"""Transform raw API responses into clean, flat records."""
logger.info("Starting transformation")
records = []
for response in raw_data:
city_name = response["_city_name"]
latitude = response["latitude"]
longitude = response["longitude"]
hourly = response["hourly"]
times = hourly["time"]
temperatures = hourly["temperature_2m"]
humidities = hourly["relative_humidity_2m"]
wind_speeds = hourly["wind_speed_10m"]
weather_codes = hourly["weather_code"]
for i in range(len(times)):
temp = temperatures[i]
humidity = humidities[i]
wind = wind_speeds[i]
code = weather_codes[i]
# Skip records with missing data
if temp is None or humidity is None or wind is None:
continue
record = {
"city_name": city_name,
"latitude": latitude,
"longitude": longitude,
"forecast_datetime": datetime.fromisoformat(times[i]),
"temperature_c": temp,
"relative_humidity": humidity,
"wind_speed_kmh": wind,
"weather_code": code if code is not None else 0,
"temperature_category": categorize_temperature(temp),
"comfort_index": compute_comfort_index(temp, humidity),
}
records.append(record)
logger.info("Transformation complete: %d records produced", len(records))
return records
# --- Load ---
def load(records):
"""Load transformed records into MySQL using upsert logic."""
logger.info("Starting load of %d records", len(records))
insert_query = """
INSERT INTO forecasts (
city_name, latitude, longitude, forecast_datetime,
temperature_c, relative_humidity, wind_speed_kmh,
weather_code, temperature_category, comfort_index
) VALUES (
%(city_name)s, %(latitude)s, %(longitude)s, %(forecast_datetime)s,
%(temperature_c)s, %(relative_humidity)s, %(wind_speed_kmh)s,
%(weather_code)s, %(temperature_category)s, %(comfort_index)s
) ON DUPLICATE KEY UPDATE
temperature_c = VALUES(temperature_c),
relative_humidity = VALUES(relative_humidity),
wind_speed_kmh = VALUES(wind_speed_kmh),
weather_code = VALUES(weather_code),
temperature_category = VALUES(temperature_category),
comfort_index = VALUES(comfort_index)
"""
connection = None
try:
connection = mysql.connector.connect(**DB_CONFIG)
cursor = connection.cursor()
cursor.executemany(insert_query, records)
connection.commit()
logger.info("Load complete: %d rows affected", cursor.rowcount)
except mysql.connector.Error as err:
logger.error("Database error: %s", err)
raise
finally:
if connection and connection.is_connected():
cursor.close()
connection.close()
# --- Main ---
def main():
"""Orchestrate the ETL pipeline."""
logger.info("=== ETL Pipeline Started ===")
start_time = datetime.now()
raw_data = extract(CITIES)
records = transform(raw_data)
load(records)
elapsed = (datetime.now() - start_time).total_seconds()
logger.info("=== ETL Pipeline Finished in %.2f seconds ===", elapsed)
if __name__ == "__main__":
main()Your pipeline is production-ready. It logs every phase, keeps configuration in one place, and handles re-runs gracefully. This is the exact pattern used in professional data engineering workflows. Next up, the Secret Mission challenges you to build a second aggregation stage on top of this foundation.
Secret mission
Build a Daily Summary Aggregation Stage
Your pipeline loads raw hourly forecasts into MySQL. But what if you want to answer questions like 'which city has the highest average temperature this week?' Raw hourly data makes that query expensive. Build a second pipeline stage that computes daily aggregates and loads them into a summary table.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely on your local machine, so there are no ongoing cloud costs. The MySQL service does use a small amount of memory while running, but it will not charge you anything.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- MySQL database weather_db with tables forecasts and city_daily_summary.
- MySQL Community Server running as a Windows service.
- Project folder weather-etl on your Desktop (containing etl_pipeline.py, schema.sql, and the venv virtual environment).
Keep everything running
No action needed. Choose this if you are still actively building or plan to extend the pipeline further.
- The MySQL service will continue running in the background and start automatically when Windows boots.
- Your weather-etl folder stays on your Desktop, ready for the next run.
- You can re-run the pipeline anytime by activating the virtual environment and running python etl_pipeline.py.
Pause - I'll come back to this later
Stop the MySQL service to free up memory. Your data and project files stay intact for when you return.
- Deactivate your Python virtual environment by running:
deactivate- Stop the MySQL service by opening a Command Prompt as Administrator and running:
NET STOP MySQLWhat if NET STOP MySQL says the service name is not found?
Your MySQL service name depends on the version you installed. Open the Windows Services app (search for Services in the Start menu) and look for a service starting with MySQL. Use the exact name shown there.
- When you come back, restart the service with:
NET START MySQLDelete - I don't want to use this again
Remove all project resources and start fresh. This permanently deletes your database, project files, and optionally the MySQL server itself.
- Deactivate your virtual environment by running:
deactivate- Drop the weather_db database by running:
mysql -u root -p -e "DROP DATABASE weather_db;"- Enter your MySQL root password when prompted.
- Delete the project folder. Navigate to your Desktop and run:
cd %USERPROFILE%\Desktop
rmdir /s /q weather-etl- If you are on macOS or Linux, use this instead:
cd ~/Desktop
rm -rf weather-etl- (Optional) Uninstall MySQL Community Server through Windows Settings.
- Press the Windows key and search for Add or remove programs.
- Find MySQL Server in the list, click it, and select Uninstall.
- Follow the uninstaller prompts to remove MySQL from your machine.
Nice Work!
Nice Work!
You just built a real ETL pipeline from scratch. You now have a production-style Python script that extracts live weather data, transforms it into analysis-ready records, and loads it cleanly into a MySQL database.
You've learned how to:
- Extract live data from the Open-Meteo API using Python's requests library and handle JSON responses with nested arrays.
- Transform raw JSON into clean, structured records with derived fields (temperature category, comfort index) and data validation to filter out incomplete entries.
- Design a MySQL schema with a unique constraint and load data using INSERT ... ON DUPLICATE KEY UPDATE so your pipeline is idempotent and safe to re-run.
- Built a second aggregation stage that computes daily summaries from raw forecast data and loads them into a dedicated summary table.
Ready to quiz yourself?