Serverless MLOps Pipeline on AWS
Build a containerized ML pipeline that forecasts oil well production on AWS.
Introduction
30 Second Summary
Every oil well starts strong, then gradually produces less over time. Energy companies need to know exactly how much each well will produce next month to plan operations and budgets.
In this project, you will build a serverless MLOps pipeline that predicts oil well production decline using machine learning. Your system will automatically detect when predictions go stale and retrain itself.
What You'll Build
A production-grade MLOps platform that forecasts oil well production decline, monitors prediction quality, and triggers automated retraining when model performance degrades.
By the end of this project, you'll have:
- A physics-based synthetic data pipeline and trained XGBoost forecasting model with domain-specific feature engineering.
- A containerized inference API deployed to AWS Lambda behind API Gateway, with all infrastructure managed by Terraform.
- A CI/CD pipeline with GitHub Actions, plus automated model monitoring and retraining triggered by drift detection.
- Secret Mission: Add an equipment failure risk classifier that predicts pump failures 24 hours ahead using rolling sensor statistics.
Are there any prerequisites?
This project assumes intermediate Python skills and basic AWS knowledge (you should know what S3 and IAM are). You should also be comfortable using the command line.
No prior machine learning or MLOps experience is needed. The project walks you through every concept as it comes up.
Before We Start
This project involves multiple tools working together. You will use Python for data generation and model training, Docker for containerizing the inference service, AWS CLI for cloud interactions, and Terraform for managing infrastructure as code.
In this step, you will install and configure every tool the pipeline needs. By the end, your machine will be ready to generate data, train models, and deploy to AWS.
In this step, get ready to:
- Install Python 3.12, Docker Desktop, AWS CLI v2, and Terraform.
- Configure AWS credentials and verify free tier access.
- Install project dependencies and create your project directory.
Install your development tools
This project requires four core tools. You will check whether each one is already installed and at the correct version.
Check Python 3.12
The ML model and data pipeline are written in Python. This project requires version 3.12 specifically because the pinned library versions target that release.
- Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
- Check your Python version by running this command:
python3 --version✔️ I see Python 3.12.x
Python 3.12 is ready. Move on to the next tool check below.
ⓧ I see a version lower than 3.12
You have Python installed but at an older version. On macOS, the system default may resolve to an older release even after installing a newer one.
- Visit the official Python downloads page: https://www.python.org/downloads/
- Download and run the Python 3.12 installer for your platform.
- On macOS, after installing, use python3.12 --version to confirm the new binary is available. If python3 --version still shows the old version, use python3.12 throughout this project instead of python3.
- On Windows, tick Add Python to PATH during installation, then reopen your terminal.
Once you see 3.12 in the version output, continue below.
ⓧ Command not found
Python is not installed yet.
- Visit https://www.python.org/downloads/ and download the Python 3.12 installer for your platform.
- Run the installer. On Windows, tick Add Python to PATH before clicking Install Now.
- Close and reopen your terminal, then run python3 --version again to confirm.
Check Docker Desktop
Docker packages your inference model into a container image that AWS Lambda can run. You need Docker Desktop installed and running.
- Check whether Docker is available by running this command:
docker --version✔️ I see a version number
Docker is installed. Make sure Docker Desktop is running (the whale icon should be visible in your system tray).
ⓧ Command not found
Docker is not installed yet.
- Visit https://www.docker.com/products/docker-desktop/ and download Docker Desktop for your platform.
- Run the installer and follow the prompts.
- Launch Docker Desktop after installation. It may take a minute to start.
- Reopen your terminal and run docker --version to confirm.
Check AWS CLI v2
The AWS CLI lets you interact with AWS services from your terminal. This project requires version 2.
- Check your AWS CLI version by running:
aws --version✔️ I see aws-cli/2.x.x
AWS CLI v2 is ready. Continue below.
ⓧ I see aws-cli/1.x.x
You have version 1 installed. Version 2 includes features this project depends on.
- Follow the official upgrade guide: https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html
- After upgrading, close and reopen your terminal, then run aws --version to confirm version 2.
ⓧ Command not found
AWS CLI is not installed yet.
- Visit https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html and follow the instructions for your platform.
- After installing, close and reopen your terminal, then run aws --version to confirm.
Check Terraform
Terraform manages all your AWS infrastructure as code. This project requires version 1.5 or higher.
- Check your Terraform version by running:
terraform --version✔️ I see version 1.5 or higher
Terraform is ready. Continue to the next section.
ⓧ I see a version lower than 1.5
Your Terraform is outdated. Upgrade to the latest version.
- On macOS, run these commands to upgrade via Homebrew:
brew tap hashicorp/tap
brew install hashicorp/tap/terraform- On Windows or Linux, download the latest binary from https://developer.hashicorp.com/terraform/install and replace your existing binary.
- Reopen your terminal and confirm with terraform --version.
ⓧ Command not found
Terraform is not installed yet.
- On macOS, install via Homebrew by running:
brew tap hashicorp/tap
brew install hashicorp/tap/terraform- On Windows or Linux, visit https://developer.hashicorp.com/terraform/install and follow the platform-specific instructions.
- Close and reopen your terminal, then run terraform --version to confirm.
Install a code editor
You need a code editor to write Python and Terraform configuration. VS Code or Cursor both work well for this project.
- If you do not have VS Code or Cursor installed, download one from https://code.visualstudio.com/ (VS Code) or https://www.cursor.com/ (Cursor).
- Run the installer and follow the prompts.
Configure AWS credentials and verify free tier access
For Terraform and the AWS CLI to create resources on your behalf, they need credentials linked to your AWS account.
- If you do not have an AWS account yet, create one at https://aws.amazon.com/free/. The free tier covers all resources used in this project.
- In the AWS Console, create an IAM user with programmatic access and attach the AdministratorAccess policy. Save the Access Key ID and Secret Access Key.
- Configure the CLI with your credentials by running:
aws configureWhat does this command ask?
The command prompts you for four values:
- AWS Access Key ID: paste the key you saved.
- AWS Secret Access Key: paste the secret.
- Default region name: enter us-east-1.
- Default output format: enter json.
- Verify your credentials work by running:
aws sts get-caller-identityYou should see a JSON response containing your Account ID, UserId, and ARN. This confirms the CLI can reach AWS.
Seeing an error about credentials?
- Double-check that you pasted the Access Key ID and Secret Access Key without extra spaces.
- Make sure your IAM user has the AdministratorAccess policy attached.
- Run aws configure again to re-enter your values.
Help me debug my AWS CLI credentials
Create your project and install dependencies
Now you will create a project folder and install the Python libraries that the ML pipeline depends on.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir oil-well-mlops
cd oil-well-mlops- Open your code editor. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type VS Code (or Cursor) and press Enter to open it.
- In your editor, click File in the top menu bar.
- Click Open Folder.
- Select the oil-well-mlops folder on your Desktop, then click Open.
- Create a new file called requirements.txt in the oil-well-mlops folder.
- Paste the following contents into requirements.txt:
xgboost==2.1.4
numpy==2.2.3
scikit-learn==1.6.1
boto3==1.36.26
pandas==2.2.3What are these libraries?
- xgboost: the gradient boosting framework that trains the production forecasting model.
- numpy: numerical computing library for array operations.
- scikit-learn: provides train/test splitting and evaluation metrics (RMSE, MAE, R-squared).
- boto3: the AWS SDK for Python. It uploads models to S3 and interacts with Lambda.
- pandas: data manipulation library for loading and transforming production data.
- Save requirements.txt.
- Create a virtual environment and install the dependencies by running these commands in your terminal (make sure you are inside the oil-well-mlops folder):
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txtWhat does this do?
The first command creates an isolated Python environment inside a venv folder. The second command activates it so that pip install puts packages there instead of your global Python.
On Windows, use venv\Scripts\activate instead of source venv/bin/activate.
The install may take a minute. You should see each package name scroll by with a success message at the end.
- Confirm the install worked by running:
pip show xgboostYou should see a summary showing Version: 2.1.4 in the output.
Install failing with version errors?
- Make sure your virtual environment is active (your terminal prompt should show (venv) at the start).
- Verify you are running Python 3.12 inside the venv with python --version. If it shows an older version, recreate the venv using python3.12 -m venv venv.
- Check you typed the version numbers in requirements.txt exactly as shown above.
Help me fix my pip install errors
✔️ Awesome, I've got everything!
Great. Double check you have saved requirements.txt and that your virtual environment is active.
ⓧ I'd like to double check the full code
Here is the complete requirements.txt file:
xgboost==2.1.4
numpy==2.2.3
scikit-learn==1.6.1
boto3==1.36.26
pandas==2.2.3Your environment is fully set up. Next up, you will generate realistic oil well production data using physics-based decline curves.
Generate Synthetic Oil Well Production Data
Your environment is ready with all the tools and libraries installed. Now it's time to create the data that your ML model will learn from.
Real oil wells produce less oil over time as reservoir pressure drops. This natural decline follows predictable physics curves that petroleum engineers have modeled since the 1940s. You will build a synthetic data generator that uses these real-world physics models to produce realistic training data for your forecasting pipeline.
In this step, get ready to:
- Build a physics-based data generator using Arps hyperbolic decline curves.
- Generate 3 years of daily production data for 10 wells.
- Upload the dataset to S3 for use in later pipeline stages.
Build the data generator class
The Arps decline curve is the industry-standard physics model for predicting how oil production decreases over time. Your generator will use the hyperbolic variant of this formula and layer on realistic operational noise like pressure depletion and water breakthrough.
- Create a new file called data_generator.py in your project directory.
- Add the imports, class definition, and helper methods by pasting this code:
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
class OilWellDataGenerator:
"""Generate synthetic oil well production data using Arps decline curves."""
def __init__(self, n_wells=10, n_days=1095, random_state=42):
self.n_wells = n_wells
self.n_days = n_days
self.rng = np.random.default_rng(random_state)
def _arps_hyperbolic_decline(self, qi, di, b, t):
"""Arps hyperbolic decline: q(t) = qi / (1 + b*di*t)^(1/b)"""
return qi / (1 + b * di * t) ** (1 / b)
def _generate_well_properties(self):
"""Generate static well properties."""
return {
"initial_oil_rate": self.rng.uniform(200, 800),
"decline_rate": self.rng.uniform(0.1, 0.4),
"b_factor": self.rng.uniform(0.3, 0.8),
"initial_pressure": self.rng.uniform(3000, 4500),
"porosity": self.rng.uniform(0.08, 0.25),
"permeability": self.rng.uniform(50, 300),
"choke_size": self.rng.integers(24, 48),
}What does this code do?
The OilWellDataGenerator class sets up a reproducible random number generator and defines two key helper methods.
- _arps_hyperbolic_decline implements the physics formula q(t) = qi / (1 + b*di*t)^(1/b) that models how oil production declines over time.
- _generate_well_properties creates randomized but realistic parameters for each well, including initial production rate, decline rate, reservoir pressure, and rock properties like porosity and permeability.
You won't see a change yet. The next chunk adds the main generation loop that produces daily records for each well.
- Add the generate method below the _generate_well_properties method, still inside the class:
def generate(self):
"""Generate production data for all wells."""
records = []
for well_id in range(self.n_wells):
props = self._generate_well_properties()
t = np.arange(self.n_days)
oil_rate = self._arps_hyperbolic_decline(
props["initial_oil_rate"], props["decline_rate"], props["b_factor"], t / 365.0
)
noise = 1 + 0.05 * self.rng.standard_normal(self.n_days)
oil_rate = oil_rate * noise
oil_rate = np.maximum(oil_rate, 0)
pressure_decline = props["initial_pressure"] * (1 - 0.15 * t / self.n_days)
pressure_noise = self.rng.normal(0, 50, self.n_days)
reservoir_pressure = pressure_decline + pressure_noise
water_cut = np.clip(0.05 + 0.4 * (t / self.n_days) ** 1.5 + self.rng.normal(0, 0.02, self.n_days), 0, 0.95)
gas_rate = oil_rate * self.rng.uniform(0.8, 1.5) * (1 + 0.001 * t)
wellhead_pressure = reservoir_pressure * self.rng.uniform(0.2, 0.3) + self.rng.normal(0, 20, self.n_days)
start_date = datetime(2021, 1, 1)
for day in range(self.n_days):
records.append({
"well_id": f"WELL_{well_id:03d}",
"date": start_date + timedelta(days=int(day)),
"days_on_production": int(day),
"oil_rate": round(float(oil_rate[day]), 2),
"gas_rate": round(float(gas_rate[day]), 2),
"water_cut": round(float(water_cut[day]), 4),
"reservoir_pressure": round(float(reservoir_pressure[day]), 1),
"wellhead_pressure": round(float(wellhead_pressure[day]), 1),
"choke_size": int(props["choke_size"]),
"porosity": round(props["porosity"], 4),
"permeability": round(props["permeability"], 1),
})
return pd.DataFrame(records)What does this code do?
The generate method loops through each well and simulates realistic daily production data over 1,095 days (3 years).
- It applies the Arps decline curve to model the base oil production rate, then multiplies by 5% random noise to simulate real-world measurement variability.
- Reservoir pressure decreases linearly by 15% over the simulation period, representing natural pressure depletion.
- Water cut increases over time following a power law curve. This simulates water breakthrough, where water from the surrounding rock displaces oil.
- Each day's measurements are assembled into a record with 11 features covering production rates, pressures, and static rock properties.
You won't see a change yet. The next chunk adds the script entry point that runs the generator and saves output.
Run the generator and produce the dataset
With the class complete, you need a script entry point that creates an instance of the generator, runs it, and saves the resulting data to a CSV file.
- Add the following code at the bottom of data_generator.py, outside the class (no indentation):
if __name__ == "__main__":
generator = OilWellDataGenerator(n_wells=10, n_days=1095, random_state=42)
df = generator.generate()
df.to_csv("production_data.csv", index=False)
print(f"Generated {len(df)} records for {df['well_id'].nunique()} wells")
print(f"Date range: {df['date'].min()} to {df['date'].max()}")What does this code do?
The if __name__ == "__main__" block runs only when you execute the file directly (not when importing it as a module).
- It creates a generator with 10 wells and 1,095 days (3 years), using a fixed random seed for reproducibility.
- The output DataFrame is saved to production_data.csv and summary statistics are printed to confirm the data looks right.
- Save data_generator.py.
- Run the data generator by executing this command:
python data_generator.pyYou should see output like this:
Generated 10950 records for 10 wells
Date range: 2021-01-01 to 2023-12-31- Confirm that production_data.csv was created by running:
ls production_data.csvYou should see production_data.csv listed in the output.
Seeing a ModuleNotFoundError?
- Make sure you installed the project dependencies from requirements.txt in Step 1. Run pip install -r requirements.txt again if needed.
- Verify you are using Python 3.12 by running python --version. If you see an older version, use python3 or python3.12 instead.
- Check that your terminal is open in the correct project directory where requirements.txt lives.
help me debug a ModuleNotFoundError when running data_generator.py
Why synthetic data instead of real data?
Real oil well production data is proprietary and expensive to access. By generating physics-based synthetic data using Arps decline curves, you get realistic patterns that follow the same mathematical models used in actual petroleum engineering.
The fixed random_state=42 ensures every run produces identical output. This reproducibility is critical for ML pipelines where you need consistent training data across environments.
✔️ Awesome, I've got everything!
Great. Make sure you have saved data_generator.py and confirmed the CSV output shows 10,950 records.
ⓧ I'd like to double check the full code
Here is the complete data_generator.py file for comparison:
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
class OilWellDataGenerator:
"""Generate synthetic oil well production data using Arps decline curves."""
def __init__(self, n_wells=10, n_days=1095, random_state=42):
self.n_wells = n_wells
self.n_days = n_days
self.rng = np.random.default_rng(random_state)
def _arps_hyperbolic_decline(self, qi, di, b, t):
"""Arps hyperbolic decline: q(t) = qi / (1 + b*di*t)^(1/b)"""
return qi / (1 + b * di * t) ** (1 / b)
def _generate_well_properties(self):
"""Generate static well properties."""
return {
"initial_oil_rate": self.rng.uniform(200, 800),
"decline_rate": self.rng.uniform(0.1, 0.4),
"b_factor": self.rng.uniform(0.3, 0.8),
"initial_pressure": self.rng.uniform(3000, 4500),
"porosity": self.rng.uniform(0.08, 0.25),
"permeability": self.rng.uniform(50, 300),
"choke_size": self.rng.integers(24, 48),
}
def generate(self):
"""Generate production data for all wells."""
records = []
for well_id in range(self.n_wells):
props = self._generate_well_properties()
t = np.arange(self.n_days)
oil_rate = self._arps_hyperbolic_decline(
props["initial_oil_rate"], props["decline_rate"], props["b_factor"], t / 365.0
)
noise = 1 + 0.05 * self.rng.standard_normal(self.n_days)
oil_rate = oil_rate * noise
oil_rate = np.maximum(oil_rate, 0)
pressure_decline = props["initial_pressure"] * (1 - 0.15 * t / self.n_days)
pressure_noise = self.rng.normal(0, 50, self.n_days)
reservoir_pressure = pressure_decline + pressure_noise
water_cut = np.clip(0.05 + 0.4 * (t / self.n_days) ** 1.5 + self.rng.normal(0, 0.02, self.n_days), 0, 0.95)
gas_rate = oil_rate * self.rng.uniform(0.8, 1.5) * (1 + 0.001 * t)
wellhead_pressure = reservoir_pressure * self.rng.uniform(0.2, 0.3) + self.rng.normal(0, 20, self.n_days)
start_date = datetime(2021, 1, 1)
for day in range(self.n_days):
records.append({
"well_id": f"WELL_{well_id:03d}",
"date": start_date + timedelta(days=int(day)),
"days_on_production": int(day),
"oil_rate": round(float(oil_rate[day]), 2),
"gas_rate": round(float(gas_rate[day]), 2),
"water_cut": round(float(water_cut[day]), 4),
"reservoir_pressure": round(float(reservoir_pressure[day]), 1),
"wellhead_pressure": round(float(wellhead_pressure[day]), 1),
"choke_size": int(props["choke_size"]),
"porosity": round(props["porosity"], 4),
"permeability": round(props["permeability"], 1),
})
return pd.DataFrame(records)
if __name__ == "__main__":
generator = OilWellDataGenerator(n_wells=10, n_days=1095, random_state=42)
df = generator.generate()
df.to_csv("production_data.csv", index=False)
print(f"Generated {len(df)} records for {df['well_id'].nunique()} wells")
print(f"Date range: {df['date'].min()} to {df['date'].max()}")
Upload the dataset to S3
Your ML pipeline will read training data from S3 rather than from a local file. This ensures that all pipeline components (training, monitoring, retraining) access the same canonical dataset regardless of which machine they run on.
- Create an S3 bucket for your project data by running this command:
aws s3 mb s3://oil-well-mlops-data-$(aws sts get-caller-identity --query Account --output text)What does this command do?
aws s3 mb creates ("makes") a new S3 bucket. The subshell $(aws sts get-caller-identity --query Account --output text) automatically fetches your AWS account ID and appends it to the bucket name.
S3 bucket names must be globally unique across all AWS accounts. Including your account ID guarantees no naming collisions.
- Upload the production data CSV to your new bucket by running:
aws s3 cp production_data.csv s3://oil-well-mlops-data-$(aws sts get-caller-identity --query Account --output text)/production_data.csv- Verify the upload succeeded by listing the bucket contents:
aws s3 ls s3://oil-well-mlops-data-$(aws sts get-caller-identity --query Account --output text)/You should see production_data.csv listed with its file size (approximately 1-2 MB).
Getting an access denied error?
- Confirm your AWS credentials are configured by running aws sts get-caller-identity. You should see your account ID, user ARN, and user ID.
- Make sure your IAM user has the s3:CreateBucket and s3:PutObject permissions. If you used an admin user in Step 1, these permissions are already included.
- If the bucket name is already taken, check that no other bucket with the same name exists in your account from a previous attempt.
help me fix an access denied error when creating an S3 bucket or uploading a file with the AWS CLI
Your synthetic dataset is generated and stored in S3. Next up, you will engineer time-series features from this raw data and train an XGBoost model to forecast production 7, 30, and 90 days into the future.
Engineer Features and Train the Model
In the previous step, you generated realistic oil well production data and uploaded it to S3. Now it's time to transform that raw data into something a machine learning model can learn from.
Raw production numbers alone don't capture trends. A model needs to see patterns like "oil rate has been declining for the past week" or "pressure dropped sharply 30 days ago." You'll build these signals as engineered features, then train an XGBoost regressor to predict production 30 days into the future.
In this step, get ready to:
- Engineer time-series features including lag values, rolling statistics, and rate of change.
- Train an XGBoost model with a time-based train/test split.
- Evaluate model performance and upload artifacts to your S3 model registry.
Create the feature engineering module
Time-series forecasting models need features that encode temporal context. You'll create lag features (what was the oil rate 1, 7, and 30 days ago?), rolling statistics (what's the average over the past week?), and derived signals like rate of change and cumulative production.
- Create a new file called feature_engineering.py in your project folder.
- Add the imports and the engineer_features function by pasting this code:
import pandas as pd
import numpy as np
def engineer_features(df):
"""Create time-series features for production forecasting."""
df = df.copy()
df = df.sort_values(["well_id", "date"]).reset_index(drop=True)
for well_id in df["well_id"].unique():
mask = df["well_id"] == well_id
# Create lag features: what was the value N days ago?
for lag in [1, 7, 30]:
df.loc[mask, f"oil_rate_lag_{lag}"] = df.loc[mask, "oil_rate"].shift(lag)
df.loc[mask, f"gas_rate_lag_{lag}"] = df.loc[mask, "gas_rate"].shift(lag)
df.loc[mask, f"pressure_lag_{lag}"] = df.loc[mask, "reservoir_pressure"].shift(lag)
# Rolling statistics: smoothed trends over 7 and 30 day windows
for window in [7, 30]:
df.loc[mask, f"oil_rate_rolling_mean_{window}"] = (
df.loc[mask, "oil_rate"].rolling(window).mean()
)
df.loc[mask, f"oil_rate_rolling_std_{window}"] = (
df.loc[mask, "oil_rate"].rolling(window).std()
)
df.loc[mask, f"pressure_rolling_mean_{window}"] = (
df.loc[mask, "reservoir_pressure"].rolling(window).mean()
)
# Rate of change and cumulative metrics
df.loc[mask, "oil_rate_change_7d"] = df.loc[mask, "oil_rate"].pct_change(7)
df.loc[mask, "pressure_change_7d"] = df.loc[mask, "reservoir_pressure"].pct_change(7)
df.loc[mask, "cumulative_oil"] = df.loc[mask, "oil_rate"].cumsum()
df.loc[mask, "drawdown"] = (
df.loc[mask, "reservoir_pressure"] - df.loc[mask, "wellhead_pressure"]
)
# Calendar features for seasonality
df["date"] = pd.to_datetime(df["date"])
df["month"] = df["date"].dt.month
df["quarter"] = df["date"].dt.quarter
df = df.dropna().reset_index(drop=True)
return dfWhat does this code do?
The engineer_features function transforms raw daily readings into predictive signals:
- Lag features let the model see what happened 1, 7, and 30 days ago for oil rate, gas rate, and pressure.
- Rolling statistics smooth out daily noise by averaging over 7-day and 30-day windows.
- Rate of change captures whether production is accelerating or decelerating.
- Drawdown (reservoir pressure minus wellhead pressure) is a key reservoir engineering metric that indicates flow potential.
- Below the engineer_features function, add the create_forecast_targets function:
def create_forecast_targets(df, horizons=[7, 30, 90]):
"""Create multi-horizon forecast targets."""
df = df.copy()
for well_id in df["well_id"].unique():
mask = df["well_id"] == well_id
# Shift oil_rate backwards to create future targets
for h in horizons:
df.loc[mask, f"target_{h}d"] = (
df.loc[mask, "oil_rate"].shift(-h)
)
df = df.dropna().reset_index(drop=True)
return dfWhat does this code do?
This function creates the prediction targets. For each day, it looks ahead 7, 30, and 90 days and records what the actual oil rate was. The model learns to predict these future values from the current features.
Rows where the target would fall outside the dataset (the last 90 days) are dropped with dropna().
- Save feature_engineering.py.
You won't see output yet. The training script in the next substep imports and uses these functions.
✔️ 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 pandas as pd
import numpy as np
def engineer_features(df):
"""Create time-series features for production forecasting."""
df = df.copy()
df = df.sort_values(["well_id", "date"]).reset_index(drop=True)
for well_id in df["well_id"].unique():
mask = df["well_id"] == well_id
for lag in [1, 7, 30]:
df.loc[mask, f"oil_rate_lag_{lag}"] = df.loc[mask, "oil_rate"].shift(lag)
df.loc[mask, f"gas_rate_lag_{lag}"] = df.loc[mask, "gas_rate"].shift(lag)
df.loc[mask, f"pressure_lag_{lag}"] = df.loc[mask, "reservoir_pressure"].shift(lag)
for window in [7, 30]:
df.loc[mask, f"oil_rate_rolling_mean_{window}"] = (
df.loc[mask, "oil_rate"].rolling(window).mean()
)
df.loc[mask, f"oil_rate_rolling_std_{window}"] = (
df.loc[mask, "oil_rate"].rolling(window).std()
)
df.loc[mask, f"pressure_rolling_mean_{window}"] = (
df.loc[mask, "reservoir_pressure"].rolling(window).mean()
)
df.loc[mask, "oil_rate_change_7d"] = df.loc[mask, "oil_rate"].pct_change(7)
df.loc[mask, "pressure_change_7d"] = df.loc[mask, "reservoir_pressure"].pct_change(7)
df.loc[mask, "cumulative_oil"] = df.loc[mask, "oil_rate"].cumsum()
df.loc[mask, "drawdown"] = (
df.loc[mask, "reservoir_pressure"] - df.loc[mask, "wellhead_pressure"]
)
df["date"] = pd.to_datetime(df["date"])
df["month"] = df["date"].dt.month
df["quarter"] = df["date"].dt.quarter
df = df.dropna().reset_index(drop=True)
return df
def create_forecast_targets(df, horizons=[7, 30, 90]):
"""Create multi-horizon forecast targets."""
df = df.copy()
for well_id in df["well_id"].unique():
mask = df["well_id"] == well_id
for h in horizons:
df.loc[mask, f"target_{h}d"] = (
df.loc[mask, "oil_rate"].shift(-h)
)
df = df.dropna().reset_index(drop=True)
return dfBuild the training script
The training script ties everything together. It generates data, engineers features, trains an XGBoost regressor, and saves the model artifact with evaluation metrics.
- Create a new file called train.py in your project folder.
- Add the imports, feature column definitions, and target column by pasting this code:
import json
import pickle
from datetime import datetime
import numpy as np
import pandas as pd
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from xgboost import XGBRegressor
from data_generator import OilWellDataGenerator
from feature_engineering import create_forecast_targets, engineer_features
# All input features the model will use for prediction
FEATURE_COLS = [
"days_on_production", "oil_rate", "gas_rate", "water_cut",
"reservoir_pressure", "wellhead_pressure", "choke_size",
"porosity", "permeability",
"oil_rate_lag_1", "oil_rate_lag_7", "oil_rate_lag_30",
"gas_rate_lag_1", "gas_rate_lag_7", "gas_rate_lag_30",
"pressure_lag_1", "pressure_lag_7", "pressure_lag_30",
"oil_rate_rolling_mean_7", "oil_rate_rolling_std_7",
"oil_rate_rolling_mean_30", "oil_rate_rolling_std_30",
"pressure_rolling_mean_7", "pressure_rolling_mean_30",
"oil_rate_change_7d", "pressure_change_7d",
"cumulative_oil", "drawdown", "month", "quarter",
]
# Predict oil rate 30 days into the future
TARGET_COL = "target_30d"What does this code do?
FEATURE_COLS lists the 30 input columns the model will learn from. These include raw measurements (oil rate, pressure), the engineered lag and rolling features you just built, and calendar features.
TARGET_COL is set to target_30d. This means the model predicts what the oil rate will be 30 days from now.
- Below the TARGET_COL line, add the train_model function:
def train_model(df):
"""Train XGBoost model with time-based split."""
df = df.sort_values("date").reset_index(drop=True)
split_idx = int(len(df) * 0.8)
train_df = df.iloc[:split_idx]
test_df = df.iloc[split_idx:]
X_train = train_df[FEATURE_COLS]
y_train = train_df[TARGET_COL]
X_test = test_df[FEATURE_COLS]
y_test = test_df[TARGET_COL]
# Configure XGBoost regressor with tuned hyperparameters
model = XGBRegressor(
n_estimators=200,
max_depth=6,
learning_rate=0.1,
subsample=0.8,
colsample_bytree=0.8,
random_state=42,
)
model.fit(X_train, y_train, eval_set=[(X_test, y_test)], verbose=False)
# Evaluate on the held-out test set
y_pred = model.predict(X_test)
metrics = {
"rmse": float(np.sqrt(mean_squared_error(y_test, y_pred))),
"mae": float(mean_absolute_error(y_test, y_pred)),
"r2": float(r2_score(y_test, y_pred)),
"train_samples": len(X_train),
"test_samples": len(X_test),
"trained_at": datetime.utcnow().isoformat(),
}
return model, metricsWhy a time-based split?
With time-series data, you cannot randomly shuffle rows into train and test sets. That would leak future information into the training data. Instead, the first 80% of data (chronologically) trains the model, and the last 20% tests it.
The XGBoost hyperparameters (200 estimators, max depth 6, learning rate 0.1) balance model complexity with generalization. subsample=0.8 and colsample_bytree=0.8 add regularization by training each tree on a random subset of rows and columns.
- Below the train_model function, add the save_model function and the main execution block:
def save_model(model, metrics, version="v1"):
"""Save model and metadata locally."""
model_path = f"model_{version}.pkl"
metadata_path = f"model_{version}_metadata.json"
# Serialize the trained model to a pickle file
with open(model_path, "wb") as f:
pickle.dump(model, f)
# Save evaluation metrics and training metadata as JSON
metrics["version"] = version
metrics["model_path"] = model_path
with open(metadata_path, "w") as f:
json.dump(metrics, f, indent=2)
print(f"Model saved: {model_path}")
print(f"Metrics: RMSE={metrics['rmse']:.2f}, MAE={metrics['mae']:.2f}, R2={metrics['r2']:.4f}")
return model_path, metadata_path
if __name__ == "__main__":
print("Generating data...")
generator = OilWellDataGenerator(n_wells=10, n_days=1095, random_state=42)
df = generator.generate()
print("Engineering features...")
df = engineer_features(df)
df = create_forecast_targets(df, horizons=[7, 30, 90])
print("Training model...")
model, metrics = train_model(df)
print("Saving model...")
save_model(model, metrics, version="v1")What does this code do?
save_model serializes the trained XGBoost model to a .pkl file and writes the evaluation metrics (RMSE, MAE, R2) plus training metadata to a JSON file.
The if __name__ == "__main__" block orchestrates the full pipeline: generate data, engineer features, create targets, train, and save. This makes the script runnable as a standalone command.
- Save train.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
import json
import pickle
from datetime import datetime
import numpy as np
import pandas as pd
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from xgboost import XGBRegressor
from data_generator import OilWellDataGenerator
from feature_engineering import create_forecast_targets, engineer_features
FEATURE_COLS = [
"days_on_production", "oil_rate", "gas_rate", "water_cut",
"reservoir_pressure", "wellhead_pressure", "choke_size",
"porosity", "permeability",
"oil_rate_lag_1", "oil_rate_lag_7", "oil_rate_lag_30",
"gas_rate_lag_1", "gas_rate_lag_7", "gas_rate_lag_30",
"pressure_lag_1", "pressure_lag_7", "pressure_lag_30",
"oil_rate_rolling_mean_7", "oil_rate_rolling_std_7",
"oil_rate_rolling_mean_30", "oil_rate_rolling_std_30",
"pressure_rolling_mean_7", "pressure_rolling_mean_30",
"oil_rate_change_7d", "pressure_change_7d",
"cumulative_oil", "drawdown", "month", "quarter",
]
TARGET_COL = "target_30d"
def train_model(df):
"""Train XGBoost model with time-based split."""
df = df.sort_values("date").reset_index(drop=True)
split_idx = int(len(df) * 0.8)
train_df = df.iloc[:split_idx]
test_df = df.iloc[split_idx:]
X_train = train_df[FEATURE_COLS]
y_train = train_df[TARGET_COL]
X_test = test_df[FEATURE_COLS]
y_test = test_df[TARGET_COL]
model = XGBRegressor(
n_estimators=200,
max_depth=6,
learning_rate=0.1,
subsample=0.8,
colsample_bytree=0.8,
random_state=42,
)
model.fit(X_train, y_train, eval_set=[(X_test, y_test)], verbose=False)
y_pred = model.predict(X_test)
metrics = {
"rmse": float(np.sqrt(mean_squared_error(y_test, y_pred))),
"mae": float(mean_absolute_error(y_test, y_pred)),
"r2": float(r2_score(y_test, y_pred)),
"train_samples": len(X_train),
"test_samples": len(X_test),
"trained_at": datetime.utcnow().isoformat(),
}
return model, metrics
def save_model(model, metrics, version="v1"):
"""Save model and metadata locally."""
model_path = f"model_{version}.pkl"
metadata_path = f"model_{version}_metadata.json"
with open(model_path, "wb") as f:
pickle.dump(model, f)
metrics["version"] = version
metrics["model_path"] = model_path
with open(metadata_path, "w") as f:
json.dump(metrics, f, indent=2)
print(f"Model saved: {model_path}")
print(f"Metrics: RMSE={metrics['rmse']:.2f}, MAE={metrics['mae']:.2f}, R2={metrics['r2']:.4f}")
return model_path, metadata_path
if __name__ == "__main__":
print("Generating data...")
generator = OilWellDataGenerator(n_wells=10, n_days=1095, random_state=42)
df = generator.generate()
print("Engineering features...")
df = engineer_features(df)
df = create_forecast_targets(df, horizons=[7, 30, 90])
print("Training model...")
model, metrics = train_model(df)
print("Saving model...")
save_model(model, metrics, version="v1")Run training and upload model artifacts to S3
With both scripts ready, you can now train the model locally and then upload the resulting artifacts to S3. This S3 bucket acts as your model registry, storing versioned models that the Lambda function will load at inference time.
- Run the training script by executing this command:
python train.pyYou should see output similar to this:
Generating data...
Engineering features...
Training model...
Saving model...
Model saved: model_v1.pkl
Metrics: RMSE=XX.XX, MAE=XX.XX, R2=0.XXXXWhat should I see?
The exact RMSE, MAE, and R2 values depend on the data, but you should see an R2 above 0.9, indicating strong predictive performance. The script also creates two new files in your project folder: model_v1.pkl (the serialized model) and model_v1_metadata.json (the metrics).
- Confirm the model files were created by running:
ls model_v1*You should see model_v1.pkl and model_v1_metadata.json in the output.
Seeing a ModuleNotFoundError?
- Make sure data_generator.py and feature_engineering.py are in the same folder as train.py.
- Check that your virtual environment is active and you've installed all dependencies from requirements.txt.
- help me debug a ModuleNotFoundError when running train.py
- Create the S3 bucket for your model registry by running this command:
aws s3 mb s3://oil-well-mlops-models-[[AWS_ACCOUNT_ID="your-aws-account-id"]]Why a separate bucket for models?
Keeping model artifacts in a dedicated bucket separates concerns. Your data bucket holds raw production CSVs. Your models bucket acts as a versioned registry where each trained model gets its own key prefix.
- Upload the trained model to S3 by running these commands:
aws s3 cp model_v1.pkl s3://oil-well-mlops-models-[[AWS_ACCOUNT_ID="your-aws-account-id"]]/models/model_v1.pkl
aws s3 cp model_v1_metadata.json s3://oil-well-mlops-models-[[AWS_ACCOUNT_ID="your-aws-account-id"]]/models/model_v1_metadata.json- Verify the upload succeeded by listing the bucket contents:
aws s3 ls s3://oil-well-mlops-models-[[AWS_ACCOUNT_ID="your-aws-account-id"]]/models/You should see both model_v1.pkl and model_v1_metadata.json listed with their file sizes and upload timestamps.
Access denied or bucket not found?
- Double check that your bucket name matches exactly (including your account ID).
- Verify your AWS credentials are configured by running aws sts get-caller-identity.
- S3 bucket names must be globally unique. If the name is taken, append a short suffix to make it unique.
- help me fix an S3 access denied error when uploading my model
Your model is trained, evaluated, and stored in S3. Next up, you'll package this model into a Docker container so it can run as a serverless Lambda function.
Containerize the Inference Service
Your trained model is sitting in S3, ready to serve predictions. The next challenge is packaging it into a service that AWS Lambda can run.
In this step, you will write a Lambda handler that loads the model from S3, wrap it in a Docker container using the official AWS Lambda base image, and test the container locally before deploying it to the cloud.
In this step, get ready to:
- Write a Lambda handler that loads the XGBoost model from S3 and returns predictions.
- Create a Dockerfile using the AWS Lambda Python 3.12 base image.
- Build and test the container locally using the Lambda Runtime Interface Emulator.
Write the Lambda inference handler
The Lambda handler is the entry point for every prediction request. It needs to load the trained model from S3 on cold start and then process incoming feature arrays into production forecasts.
- Create a new file called lambda_handler.py in your project folder by running:
touch lambda_handler.py- Open lambda_handler.py in your editor and add the imports and model loading logic:
import json
import os
import pickle
import tempfile
import boto3
import numpy as np
# Initialize S3 client and read model location from environment variables
s3_client = boto3.client("s3")
MODEL_BUCKET = os.environ.get("MODEL_BUCKET", "oil-well-mlops-models")
MODEL_KEY = os.environ.get("MODEL_KEY", "models/model_v1.pkl")
# Global model variable - persists across warm invocations
model = None
def load_model():
"""Load model from S3 on cold start."""
global model
if model is None:
# Download model to Lambda's writable /tmp directory
tmp_path = os.path.join(tempfile.gettempdir(), "model.pkl")
s3_client.download_file(MODEL_BUCKET, MODEL_KEY, tmp_path)
with open(tmp_path, "rb") as f:
model = pickle.load(f)
return modelWhat does this code do?
The module-level model = None variable persists between warm Lambda invocations. The load_model() function only downloads the model from S3 on the first (cold) invocation, then reuses it for subsequent requests.
Lambda provides a writable /tmp directory for temporary files. The model gets downloaded there because the rest of the container filesystem is read-only.
- Below the load_model() function, add the request handler:
def handler(event, context):
"""Lambda handler for production forecast predictions."""
try:
mdl = load_model()
# Support both API Gateway (httpMethod) and direct invocation
if event.get("httpMethod"):
body = json.loads(event.get("body", "{}"))
else:
body = event
features = body.get("features")
if not features:
return {
"statusCode": 400,
"body": json.dumps({"error": "Missing 'features' in request body"}),
}
# Reshape features into a single-row array for prediction
features_array = np.array(features).reshape(1, -1)
prediction = mdl.predict(features_array)
response = {
"prediction": float(prediction[0]),
"model_version": MODEL_KEY.split("/")[-1].replace(".pkl", ""),
"unit": "barrels_per_day",
}
return {
"statusCode": 200,
"headers": {"Content-Type": "application/json"},
"body": json.dumps(response),
}
except Exception as e:
return {
"statusCode": 500,
"body": json.dumps({"error": str(e)}),
}What does the handler do?
The handler() function processes each incoming request. It checks whether the event comes from API Gateway (which wraps the body in an httpMethod envelope) or from a direct Lambda invocation.
It extracts the features array from the request body, reshapes it to match what XGBoost expects, and returns the prediction in barrels per day along with the model version.
- Save lambda_handler.py.
Seeing import errors in your editor?
If your editor underlines boto3 or numpy with red, make sure you installed the project dependencies from requirements.txt in the previous step. These packages need to be in your local Python environment for the editor to resolve them.
help me fix import errors for boto3 or numpy in my Lambda handler
✔️ Awesome, I've got everything!
Great. Double check you have saved lambda_handler.py before moving on.
ⓧ I'd like to double check the full code
import json
import os
import pickle
import tempfile
import boto3
import numpy as np
s3_client = boto3.client("s3")
MODEL_BUCKET = os.environ.get("MODEL_BUCKET", "oil-well-mlops-models")
MODEL_KEY = os.environ.get("MODEL_KEY", "models/model_v1.pkl")
model = None
def load_model():
"""Load model from S3 on cold start."""
global model
if model is None:
tmp_path = os.path.join(tempfile.gettempdir(), "model.pkl")
s3_client.download_file(MODEL_BUCKET, MODEL_KEY, tmp_path)
with open(tmp_path, "rb") as f:
model = pickle.load(f)
return model
def handler(event, context):
"""Lambda handler for production forecast predictions."""
try:
mdl = load_model()
if event.get("httpMethod"):
body = json.loads(event.get("body", "{}"))
else:
body = event
features = body.get("features")
if not features:
return {
"statusCode": 400,
"body": json.dumps({"error": "Missing 'features' in request body"}),
}
features_array = np.array(features).reshape(1, -1)
prediction = mdl.predict(features_array)
response = {
"prediction": float(prediction[0]),
"model_version": MODEL_KEY.split("/")[-1].replace(".pkl", ""),
"unit": "barrels_per_day",
}
return {
"statusCode": 200,
"headers": {"Content-Type": "application/json"},
"body": json.dumps(response),
}
except Exception as e:
return {
"statusCode": 500,
"body": json.dumps({"error": str(e)}),
}Create the Dockerfile
AWS Lambda can run container images up to 10 GB uncompressed. The official AWS Lambda Python base images come preloaded with the Python runtime, the runtime interface client, and the runtime interface emulator for local testing.
- Create a file called Dockerfile in your project folder by running:
touch Dockerfile- Open Dockerfile and add the following content:
# Use the official AWS Lambda Python 3.12 base image
FROM public.ecr.aws/lambda/python:3.12
# Install Python dependencies
COPY requirements.txt ${LAMBDA_TASK_ROOT}/
RUN pip install --no-cache-dir -r requirements.txt
# Copy the Lambda handler code
COPY lambda_handler.py ${LAMBDA_TASK_ROOT}/
# Set the handler function as the entrypoint
CMD ["lambda_handler.handler"]What does this Dockerfile do?
- FROM public.ecr.aws/lambda/python:3.12 pulls the official AWS Lambda Python 3.12 base image from Amazon ECR Public. This image includes the Lambda runtime and the Runtime Interface Emulator for local testing.
- LAMBDA_TASK_ROOT is a built-in environment variable in AWS Lambda base images that points to the directory where your function code lives.
- CMD ["lambda_handler.handler"] tells Lambda which function to invoke. The format is module.function.
Why container images instead of zip packages?
Lambda zip packages have a 50 MB limit. XGBoost plus its dependencies (NumPy, scikit-learn) easily exceed that. Container images support up to 10 GB uncompressed, giving plenty of room for ML libraries.
Containers also let you reproduce the exact runtime environment locally, which makes debugging ML inference issues much easier.
- Save Dockerfile.
- Confirm both new files exist by running:
lsYou should see lambda_handler.py and Dockerfile alongside your existing project files.
Don't see the new files?
Make sure you created the files inside your project directory, not in a parent or sibling folder. Run pwd to check your current location.
help me find where my Dockerfile and lambda_handler.py were created
Build and test the container locally
Before deploying to AWS, you can verify the container works locally. The AWS Lambda base image includes a built-in Runtime Interface Emulator that simulates the Lambda execution environment on your machine.
- Build the Docker image by running:
docker build --provenance=false -t oil-well-mlops-inference:latest .What does this command do?
- docker build reads the Dockerfile and creates a container image.
- --provenance=false disables provenance attestation metadata, which is required for Lambda container image compatibility.
- -t oil-well-mlops-inference:latest tags the image with a name you can reference later.
- . tells Docker to use the current directory as the build context.
- Wait for the build to complete. You should see output ending with a line like Successfully tagged oil-well-mlops-inference:latest.
Build failing with connection errors?
Make sure Docker Desktop is running. You can check by running docker ps in your terminal. If you see a connection error, open Docker Desktop and wait for it to fully start before retrying.
help me fix Docker build failures for my Lambda container image
- Start the container locally using the Runtime Interface Emulator by running:
docker run --platform linux/amd64 -p 9000:8080 oil-well-mlops-inference:latestWhat does this command do?
This maps port 9000 on your machine to port 8080 inside the container, where the Lambda Runtime Interface Emulator listens. The emulator creates a local endpoint at localhost:9000/2015-03-31/functions/function/invocations that mimics how Lambda processes requests in production.
- Open a second terminal window. Send a test prediction request by running:
curl -X POST "http://localhost:9000/2015-03-31/functions/function/invocations" \
-d '{"features": [100, 350.5, 420.3, 0.15, 3200.0, 800.0, 32, 0.15, 150.0, 345.0, 360.0, 380.0, 410.0, 430.0, 450.0, 3180.0, 3200.0, 3250.0, 355.0, 15.2, 358.0, 12.5, 3190.0, 3210.0, -0.02, -0.005, 35000.0, 2400.0, 6, 2]}'What should I see?
Because the container cannot access your S3 bucket locally, you will see a 500 response with an error about failing to download the model. This is expected. The important thing is that the container started, received the request, and attempted to process it.
A response like {"error": "...NoCredentialsError..."} or a connection-refused error from the S3 call confirms your handler code is running correctly inside the container. In production, the Lambda function will have IAM permissions to access S3.
- Return to the first terminal and press Ctrl+C to stop the container.
Curl command hanging or connection refused?
Make sure the container is still running in the first terminal. If it exited immediately, check the terminal output for Python import errors or missing dependency messages.
If you are on Windows and do not have curl available, open PowerShell and use Invoke-WebRequest instead, or install curl via your package manager.
help me troubleshoot my local Lambda container test
Your inference service is containerized and tested. Next up, you will deploy this container to AWS using Terraform to provision the full infrastructure stack.
Deploy Infrastructure with Terraform
You have a trained model and a working container image. The next challenge is deploying all the AWS infrastructure these components need to run in the cloud.
Manually clicking through the AWS console to create S3 buckets, Lambda functions, and API endpoints is error-prone and impossible to reproduce. Terraform lets you define all your infrastructure in a single file and provision it with one command.
In this step, get ready to:
- Write a Terraform configuration defining 7+ AWS services.
- Initialize and provision all infrastructure resources.
- Push your Docker image to ECR and deploy the Lambda function.
Write the Terraform configuration file
Terraform uses .tf files written in HashiCorp Configuration Language (HCL) to describe your desired infrastructure. You will create a single main.tf file that defines every AWS resource the pipeline needs.
- Create a new file called main.tf in your project directory.
- Add the Terraform provider configuration and storage resources by pasting the following code:
terraform {
required_providers {
aws = {
source = "hashicorp/aws"
version = "~> 5.0"
}
}
required_version = ">= 1.5"
}
provider "aws" {
region = var.aws_region
}
variable "aws_region" {
default = "us-east-1"
}
variable "project_name" {
default = "oil-well-mlops"
}
# S3 Bucket for data and models
resource "aws_s3_bucket" "data" {
bucket = "${var.project_name}-data-${data.aws_caller_identity.current.account_id}"
}
resource "aws_s3_bucket" "models" {
bucket = "${var.project_name}-models-${data.aws_caller_identity.current.account_id}"
}
data "aws_caller_identity" "current" {}
# ECR Repository
resource "aws_ecr_repository" "inference" {
name = "${var.project_name}-inference"
image_tag_mutability = "MUTABLE"
force_delete = true
}
# DynamoDB Table for predictions and model registry
resource "aws_dynamodb_table" "predictions" {
name = "${var.project_name}-predictions"
billing_mode = "PAY_PER_REQUEST"
hash_key = "prediction_id"
range_key = "timestamp"
attribute {
name = "prediction_id"
type = "S"
}
attribute {
name = "timestamp"
type = "S"
}
}What does this code do?
The terraform block pins the AWS provider to version ~> 5.0 and requires Terraform 1.5+. The variables set your region and a project name prefix used across all resources.
The data "aws_caller_identity" block fetches your AWS account ID at runtime, making bucket names globally unique without you hardcoding anything. The S3 buckets store production data and model artifacts. ECR holds your Docker container image. DynamoDB logs predictions for later monitoring.
- Next, add the IAM role and policy that gives your Lambda function permission to access S3, DynamoDB, and CloudWatch. Paste this below the DynamoDB resource in main.tf:
# IAM Role for Lambda
resource "aws_iam_role" "lambda_role" {
name = "${var.project_name}-lambda-role"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = { Service = "lambda.amazonaws.com" }
}]
})
}
resource "aws_iam_role_policy" "lambda_policy" {
name = "${var.project_name}-lambda-policy"
role = aws_iam_role.lambda_role.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Effect = "Allow"
Action = ["s3:GetObject", "s3:PutObject", "s3:ListBucket"]
Resource = ["${aws_s3_bucket.models.arn}", "${aws_s3_bucket.models.arn}/*", "${aws_s3_bucket.data.arn}", "${aws_s3_bucket.data.arn}/*"]
},
{
Effect = "Allow"
Action = ["dynamodb:PutItem", "dynamodb:GetItem", "dynamodb:Query"]
Resource = [aws_dynamodb_table.predictions.arn]
},
{
Effect = "Allow"
Action = ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"]
Resource = ["arn:aws:logs:*:*:*"]
},
{
Effect = "Allow"
Action = ["cloudwatch:PutMetricData"]
Resource = ["*"]
}
]
})
}What does the IAM configuration do?
The IAM role defines an identity that the Lambda function assumes when it runs. The attached policy grants the minimum permissions needed: reading models from S3, writing predictions to DynamoDB, sending logs to CloudWatch, and publishing custom metrics.
This is the principle of least privilege. The Lambda can only access the specific buckets and tables you created, not your entire AWS account.
- Finally, add the Lambda function, API Gateway, CloudWatch alarm, EventBridge rule, and outputs. Paste this below the IAM policy in main.tf:
# Lambda Function
resource "aws_lambda_function" "inference" {
function_name = "${var.project_name}-inference"
role = aws_iam_role.lambda_role.arn
package_type = "Image"
image_uri = "${aws_ecr_repository.inference.repository_url}:latest"
timeout = 30
memory_size = 512
environment {
variables = {
MODEL_BUCKET = aws_s3_bucket.models.id
MODEL_KEY = "models/model_v1.pkl"
}
}
depends_on = [aws_ecr_repository.inference]
}
# API Gateway
resource "aws_apigatewayv2_api" "api" {
name = "${var.project_name}-api"
protocol_type = "HTTP"
}
resource "aws_apigatewayv2_integration" "lambda_integration" {
api_id = aws_apigatewayv2_api.api.id
integration_type = "AWS_PROXY"
integration_uri = aws_lambda_function.inference.invoke_arn
payload_format_version = "2.0"
}
resource "aws_apigatewayv2_route" "predict" {
api_id = aws_apigatewayv2_api.api.id
route_key = "POST /predict"
target = "integrations/${aws_apigatewayv2_integration.lambda_integration.id}"
}
resource "aws_apigatewayv2_stage" "prod" {
api_id = aws_apigatewayv2_api.api.id
name = "prod"
auto_deploy = true
}
resource "aws_lambda_permission" "api_gateway" {
statement_id = "AllowAPIGateway"
action = "lambda:InvokeFunction"
function_name = aws_lambda_function.inference.function_name
principal = "apigateway.amazonaws.com"
source_arn = "${aws_apigatewayv2_api.api.execution_arn}/*/*"
}
# CloudWatch Alarm for drift
resource "aws_cloudwatch_metric_alarm" "model_drift" {
alarm_name = "${var.project_name}-model-drift"
comparison_operator = "GreaterThanThreshold"
evaluation_periods = 1
metric_name = "ModelRMSE"
namespace = "OilWellMLOps"
period = 86400
statistic = "Average"
threshold = 50
alarm_description = "Triggered when model RMSE exceeds threshold"
}
# EventBridge Rule for daily monitoring
resource "aws_cloudwatch_event_rule" "daily_monitor" {
name = "${var.project_name}-daily-monitor"
schedule_expression = "rate(1 day)"
}
# Outputs
output "api_endpoint" {
value = "${aws_apigatewayv2_api.api.api_endpoint}/prod/predict"
}
output "ecr_repository_url" {
value = aws_ecr_repository.inference.repository_url
}
output "model_bucket" {
value = aws_s3_bucket.models.id
}What does this section define?
The Lambda function uses a container image from ECR with 512 MB of memory and a 30-second timeout. The environment variables MODEL_BUCKET and MODEL_KEY tell the handler where to find the trained model in S3.
API Gateway creates an HTTP endpoint with a POST /predict route that forwards requests to Lambda. The CloudWatch alarm fires when model RMSE exceeds 50 (your drift threshold). EventBridge schedules a daily monitoring run.
- Save main.tf.
Seeing HCL syntax errors?
Check that every opening brace { has a matching closing brace }. HCL is whitespace-sensitive for readability but braces must balance. Also confirm you pasted each section in order without accidentally duplicating the terraform block.
help me debug my main.tf syntax error
✔️ Awesome, I've got everything!
Great. Double check that you saved main.tf and that it contains all three sections: the provider/storage layer, the IAM layer, and the compute/API/monitoring layer.
ⓧ I'd like to double check the full code
terraform {
required_providers {
aws = {
source = "hashicorp/aws"
version = "~> 5.0"
}
}
required_version = ">= 1.5"
}
provider "aws" {
region = var.aws_region
}
variable "aws_region" {
default = "us-east-1"
}
variable "project_name" {
default = "oil-well-mlops"
}
# S3 Bucket for data and models
resource "aws_s3_bucket" "data" {
bucket = "${var.project_name}-data-${data.aws_caller_identity.current.account_id}"
}
resource "aws_s3_bucket" "models" {
bucket = "${var.project_name}-models-${data.aws_caller_identity.current.account_id}"
}
data "aws_caller_identity" "current" {}
# ECR Repository
resource "aws_ecr_repository" "inference" {
name = "${var.project_name}-inference"
image_tag_mutability = "MUTABLE"
force_delete = true
}
# DynamoDB Table for predictions and model registry
resource "aws_dynamodb_table" "predictions" {
name = "${var.project_name}-predictions"
billing_mode = "PAY_PER_REQUEST"
hash_key = "prediction_id"
range_key = "timestamp"
attribute {
name = "prediction_id"
type = "S"
}
attribute {
name = "timestamp"
type = "S"
}
}
# IAM Role for Lambda
resource "aws_iam_role" "lambda_role" {
name = "${var.project_name}-lambda-role"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = { Service = "lambda.amazonaws.com" }
}]
})
}
resource "aws_iam_role_policy" "lambda_policy" {
name = "${var.project_name}-lambda-policy"
role = aws_iam_role.lambda_role.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Effect = "Allow"
Action = ["s3:GetObject", "s3:PutObject", "s3:ListBucket"]
Resource = ["${aws_s3_bucket.models.arn}", "${aws_s3_bucket.models.arn}/*", "${aws_s3_bucket.data.arn}", "${aws_s3_bucket.data.arn}/*"]
},
{
Effect = "Allow"
Action = ["dynamodb:PutItem", "dynamodb:GetItem", "dynamodb:Query"]
Resource = [aws_dynamodb_table.predictions.arn]
},
{
Effect = "Allow"
Action = ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"]
Resource = ["arn:aws:logs:*:*:*"]
},
{
Effect = "Allow"
Action = ["cloudwatch:PutMetricData"]
Resource = ["*"]
}
]
})
}
# Lambda Function
resource "aws_lambda_function" "inference" {
function_name = "${var.project_name}-inference"
role = aws_iam_role.lambda_role.arn
package_type = "Image"
image_uri = "${aws_ecr_repository.inference.repository_url}:latest"
timeout = 30
memory_size = 512
environment {
variables = {
MODEL_BUCKET = aws_s3_bucket.models.id
MODEL_KEY = "models/model_v1.pkl"
}
}
depends_on = [aws_ecr_repository.inference]
}
# API Gateway
resource "aws_apigatewayv2_api" "api" {
name = "${var.project_name}-api"
protocol_type = "HTTP"
}
resource "aws_apigatewayv2_integration" "lambda_integration" {
api_id = aws_apigatewayv2_api.api.id
integration_type = "AWS_PROXY"
integration_uri = aws_lambda_function.inference.invoke_arn
payload_format_version = "2.0"
}
resource "aws_apigatewayv2_route" "predict" {
api_id = aws_apigatewayv2_api.api.id
route_key = "POST /predict"
target = "integrations/${aws_apigatewayv2_integration.lambda_integration.id}"
}
resource "aws_apigatewayv2_stage" "prod" {
api_id = aws_apigatewayv2_api.api.id
name = "prod"
auto_deploy = true
}
resource "aws_lambda_permission" "api_gateway" {
statement_id = "AllowAPIGateway"
action = "lambda:InvokeFunction"
function_name = aws_lambda_function.inference.function_name
principal = "apigateway.amazonaws.com"
source_arn = "${aws_apigatewayv2_api.api.execution_arn}/*/*"
}
# CloudWatch Alarm for drift
resource "aws_cloudwatch_metric_alarm" "model_drift" {
alarm_name = "${var.project_name}-model-drift"
comparison_operator = "GreaterThanThreshold"
evaluation_periods = 1
metric_name = "ModelRMSE"
namespace = "OilWellMLOps"
period = 86400
statistic = "Average"
threshold = 50
alarm_description = "Triggered when model RMSE exceeds threshold"
}
# EventBridge Rule for daily monitoring
resource "aws_cloudwatch_event_rule" "daily_monitor" {
name = "${var.project_name}-daily-monitor"
schedule_expression = "rate(1 day)"
}
# Outputs
output "api_endpoint" {
value = "${aws_apigatewayv2_api.api.api_endpoint}/prod/predict"
}
output "ecr_repository_url" {
value = aws_ecr_repository.inference.repository_url
}
output "model_bucket" {
value = aws_s3_bucket.models.id
}Why Terraform instead of the AWS Console?
In production MLOps, you often need to tear down and recreate environments for testing, staging, and production. Defining infrastructure in code means you can reproduce the exact same environment with a single command.
Terraform also tracks the state of every resource. If someone manually changes a resource in the console, Terraform detects the drift and can bring it back in line. This is critical when managing 7+ interlinked services.
Initialize Terraform and create the ECR repository
Before Terraform can create resources, it needs to download the AWS provider plugin. After that, you will create the ECR repository first so that your Docker image has somewhere to land before the Lambda function is created.
- Initialize Terraform by running this command:
terraform initWhat does terraform init do?
This downloads the AWS provider plugin specified in your main.tf and sets up the .terraform directory for state management. You only run it once per project (or after changing providers).
- You should see output ending with "Terraform has been successfully initialized."
- Create the ECR repository first by running a targeted apply:
terraform apply -target="aws_ecr_repository.inference"Why create ECR first?
The Lambda function references an image in ECR. If you try to create Lambda before pushing an image, AWS will reject the request. By targeting only the ECR repository, you create a place to push your image before provisioning the Lambda.
Terraform will also create any resources that ECR depends on (the data.aws_caller_identity lookup) automatically.
- Terraform will show you a plan of what it intends to create. Type yes when prompted and press Enter to confirm.
- You should see "Apply complete! Resources: 1 added" in the output.
Seeing an authentication error?
Make sure your AWS CLI credentials are configured correctly. Run aws sts get-caller-identity to verify. If that fails, reconfigure credentials with aws configure.
help me fix my Terraform AWS authentication error
Push the Docker image to ECR and deploy all resources
With the ECR repository created, you can now push your locally built image to it. Once the image exists in ECR, a full Terraform apply will successfully create the Lambda function and everything else.
- Get your ECR repository URL from Terraform by running:
terraform output ecr_repository_url- Note the URL in the output. It follows the pattern <account_id>.dkr.ecr.us-east-1.amazonaws.com/oil-well-mlops-inference.
- Authenticate Docker to your ECR registry by running this command:
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin [[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]].dkr.ecr.us-east-1.amazonaws.comWhat does this command do?
The aws ecr get-login-password command fetches a temporary authentication token from AWS. The pipe sends that token to docker login so Docker can push images to your private ECR registry. The token is valid for 12 hours.
- You should see "Login Succeeded" in the output.
- Tag your local Docker image with the ECR repository URL by running:
docker tag oil-well-mlops-inference:latest [[ECR_REPO_URL="your ECR repository URL from terraform output"]]:latest- Push the image to ECR by running:
docker push [[ECR_REPO_URL="your ECR repository URL from terraform output"]]:latest- The push may take a few minutes depending on your connection speed. You should see layer upload progress followed by a digest confirmation.
Push failing with authorization errors?
Make sure you ran the aws ecr get-login-password command with the correct account ID. You can find your account ID by running aws sts get-caller-identity. Also verify Docker Desktop is running.
help me fix my ECR docker push authorization error
- Now provision all remaining resources by running the full Terraform apply:
terraform apply- Terraform will show a plan listing all resources to create. Type yes and press Enter to confirm.
- The apply takes 1-2 minutes. You should see "Apply complete!" with a count of resources added.
- Verify your API endpoint is live by running:
terraform output api_endpoint- Test the live prediction API with a sample request by running:
curl -X POST $(terraform output -raw api_endpoint) \
-H "Content-Type: application/json" \
-d '{"features": [500, 350.5, 420.3, 0.15, 3200.0, 950.0, 32, 0.15, 150.0, 355.0, 360.0, 370.0, 425.0, 430.0, 440.0, 3210.0, 3220.0, 3250.0, 358.0, 8.5, 355.0, 12.0, 3215.0, 3225.0, -0.02, -0.005, 175000.0, 2250.0, 6, 2]}'What should I see?
You should get a JSON response containing a prediction value (a forecasted oil production rate in barrels per day), a model_version field showing model_v1, and a unit field showing barrels_per_day. The first request may take 10-15 seconds due to Lambda cold start.
Getting a 500 error or timeout?
The first Lambda invocation triggers a cold start where it downloads the model from S3. Verify that model_v1.pkl exists in your models S3 bucket at the models/ prefix. Also check that the Lambda function's IAM role has s3:GetObject permission on the models bucket.
help me debug my Lambda 500 error for oil-well-mlops-inference
Your entire ML infrastructure is now live. In the next step, you will automate the build-test-deploy cycle with a CI/CD pipeline so that code changes trigger automatic redeployments.
Build the CI/CD Pipeline
Your ML model is deployed and serving predictions through the API. But what happens when you retrain the model or fix a bug in the handler code?
Right now, every change requires you to manually rebuild the Docker image, push it to ECR, and update the Lambda function. A CI/CD pipeline automates this entire cycle so that every push to your main branch triggers tests, builds a fresh container, and deploys it.
In this step, get ready to:
- Write unit tests for the data generator and feature engineering modules.
- Create a GitHub Actions workflow that builds, tests, and deploys on push to main.
- Configure GitHub OIDC so the pipeline authenticates to AWS without long-lived secrets.
Write unit tests for the data pipeline
Before automating deployment, you need a test suite that gates bad code from reaching production. These tests validate that the data generator produces the expected shape and that feature engineering creates the correct columns.
- Create a tests directory inside your project folder by running this command:
mkdir tests- Create the test file by running:
touch tests/test_model.py- Confirm the file exists by running:
ls tests/You should see test_model.py in the output.
- Open tests/test_model.py in your editor and paste the following test suite:
import numpy as np
import pytest
from data_generator import OilWellDataGenerator
from feature_engineering import create_forecast_targets, engineer_features
def test_data_generator_shape():
"""Verify the generator produces the correct number of rows."""
gen = OilWellDataGenerator(n_wells=2, n_days=100, random_state=42)
df = gen.generate()
assert len(df) == 200
assert df["well_id"].nunique() == 2
def test_data_generator_columns():
"""Verify all expected columns are present."""
gen = OilWellDataGenerator(n_wells=1, n_days=10, random_state=42)
df = gen.generate()
expected_cols = [
"well_id", "date", "days_on_production", "oil_rate",
"gas_rate", "water_cut", "reservoir_pressure",
"wellhead_pressure", "choke_size", "porosity", "permeability",
]
for col in expected_cols:
assert col in df.columns
def test_oil_rate_positive():
"""Verify oil rates are never negative."""
gen = OilWellDataGenerator(n_wells=3, n_days=500, random_state=42)
df = gen.generate()
assert (df["oil_rate"] >= 0).all()
def test_feature_engineering():
"""Verify lag and rolling features are created."""
gen = OilWellDataGenerator(n_wells=1, n_days=100, random_state=42)
df = gen.generate()
df = engineer_features(df)
assert "oil_rate_lag_1" in df.columns
assert "oil_rate_rolling_mean_7" in df.columns
assert "drawdown" in df.columns
def test_forecast_targets():
"""Verify forecast target columns exist and have no NaNs."""
gen = OilWellDataGenerator(n_wells=1, n_days=200, random_state=42)
df = gen.generate()
df = engineer_features(df)
df = create_forecast_targets(df, horizons=[7, 30])
assert "target_7d" in df.columns
assert "target_30d" in df.columns
assert not df["target_30d"].isna().any()What does this test file do?
Each test function validates a different property of the data pipeline:
- test_data_generator_shape checks that 2 wells x 100 days = 200 rows.
- test_data_generator_columns verifies all 11 expected columns exist.
- test_oil_rate_positive confirms the physics model never outputs negative production.
- test_feature_engineering ensures lag features and rolling stats are created correctly.
- test_forecast_targets validates that multi-horizon targets are present with no missing values.
- Save tests/test_model.py.
- Run the test suite locally to confirm everything passes:
pytest tests/ -vYou should see 5 tests passing with green output.
Tests failing with ModuleNotFoundError?
Make sure you are running pytest from the project root directory (the folder containing data_generator.py). Python needs the modules to be importable from the current directory.
If pytest is not installed, install it by running pip install pytest.
Why are my tests failing with import errors?
Create the GitHub Actions workflow
GitHub Actions workflows are YAML files stored in a .github/workflows directory. When you push to the main branch, GitHub detects the workflow file and executes the pipeline automatically.
- Create the workflow directory structure by running this command:
mkdir -p .github/workflows- Create the workflow file by running:
touch .github/workflows/deploy.yml- Open .github/workflows/deploy.yml in your editor and paste the following workflow:
name: Deploy ML Pipeline
on:
push:
branches: [main]
permissions:
id-token: write
contents: read
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- run: pip install -r requirements.txt pytest
- run: pytest tests/ -v
deploy:
needs: test
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: aws-actions/configure-aws-credentials@v4
with:
role-to-assume: ${{ secrets.AWS_ROLE_ARN }}
aws-region: us-east-1
- uses: aws-actions/amazon-ecr-login@v2
id: ecr-login
- name: Build and push Docker image
env:
ECR_REGISTRY: ${{ steps.ecr-login.outputs.registry }}
ECR_REPOSITORY: oil-well-mlops-inference
IMAGE_TAG: ${{ github.sha }}
run: |
docker build --provenance=false -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
docker build --provenance=false -t $ECR_REGISTRY/$ECR_REPOSITORY:latest .
docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
docker push $ECR_REGISTRY/$ECR_REPOSITORY:latest
- name: Update Lambda function
env:
ECR_REGISTRY: ${{ steps.ecr-login.outputs.registry }}
ECR_REPOSITORY: oil-well-mlops-inference
IMAGE_TAG: ${{ github.sha }}
run: |
aws lambda update-function-code \
--function-name oil-well-mlops-inference \
--image-uri $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG \
--publishWhat does this workflow do?
The workflow has two jobs that run sequentially:
- test installs Python 3.12, installs dependencies, and runs pytest. If any test fails, the pipeline stops here.
- deploy only runs after tests pass. It authenticates to AWS using OIDC, logs into ECR, builds the Docker image with --provenance=false (required for Lambda compatibility), pushes the image tagged with the git SHA, and updates the Lambda function to use the new image.
The permissions block at the top grants the workflow the ability to request an OIDC token (id-token: write) and read the repository contents.
Why OIDC instead of access keys?
Traditional CI/CD stores long-lived AWS access keys as secrets. If those keys leak, an attacker has persistent access to your account.
OIDC federation lets GitHub request short-lived credentials from AWS for each workflow run. The credentials expire automatically. No long-lived secrets to rotate or worry about.
- Save .github/workflows/deploy.yml.
✔️ Awesome, I've got everything!
Great. Double check you have saved both tests/test_model.py and .github/workflows/deploy.yml.
ⓧ I'd like to double check the full code
Here is the complete workflow file:
name: Deploy ML Pipeline
on:
push:
branches: [main]
permissions:
id-token: write
contents: read
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- run: pip install -r requirements.txt pytest
- run: pytest tests/ -v
deploy:
needs: test
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: aws-actions/configure-aws-credentials@v4
with:
role-to-assume: ${{ secrets.AWS_ROLE_ARN }}
aws-region: us-east-1
- uses: aws-actions/amazon-ecr-login@v2
id: ecr-login
- name: Build and push Docker image
env:
ECR_REGISTRY: ${{ steps.ecr-login.outputs.registry }}
ECR_REPOSITORY: oil-well-mlops-inference
IMAGE_TAG: ${{ github.sha }}
run: |
docker build --provenance=false -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
docker build --provenance=false -t $ECR_REGISTRY/$ECR_REPOSITORY:latest .
docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
docker push $ECR_REGISTRY/$ECR_REPOSITORY:latest
- name: Update Lambda function
env:
ECR_REGISTRY: ${{ steps.ecr-login.outputs.registry }}
ECR_REPOSITORY: oil-well-mlops-inference
IMAGE_TAG: ${{ github.sha }}
run: |
aws lambda update-function-code \
--function-name oil-well-mlops-inference \
--image-uri $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG \
--publishAnd the complete test file:
import numpy as np
import pytest
from data_generator import OilWellDataGenerator
from feature_engineering import create_forecast_targets, engineer_features
def test_data_generator_shape():
gen = OilWellDataGenerator(n_wells=2, n_days=100, random_state=42)
df = gen.generate()
assert len(df) == 200
assert df["well_id"].nunique() == 2
def test_data_generator_columns():
gen = OilWellDataGenerator(n_wells=1, n_days=10, random_state=42)
df = gen.generate()
expected_cols = ["well_id", "date", "days_on_production", "oil_rate", "gas_rate", "water_cut", "reservoir_pressure", "wellhead_pressure", "choke_size", "porosity", "permeability"]
for col in expected_cols:
assert col in df.columns
def test_oil_rate_positive():
gen = OilWellDataGenerator(n_wells=3, n_days=500, random_state=42)
df = gen.generate()
assert (df["oil_rate"] >= 0).all()
def test_feature_engineering():
gen = OilWellDataGenerator(n_wells=1, n_days=100, random_state=42)
df = gen.generate()
df = engineer_features(df)
assert "oil_rate_lag_1" in df.columns
assert "oil_rate_rolling_mean_7" in df.columns
assert "drawdown" in df.columns
def test_forecast_targets():
gen = OilWellDataGenerator(n_wells=1, n_days=200, random_state=42)
df = gen.generate()
df = engineer_features(df)
df = create_forecast_targets(df, horizons=[7, 30])
assert "target_7d" in df.columns
assert "target_30d" in df.columns
assert not df["target_30d"].isna().any()Push to GitHub and configure OIDC credentials
The workflow needs two things to run: the code in a GitHub repository, and an AWS IAM role that GitHub can assume via OIDC. You will set up the OIDC identity provider in AWS, create a role with the right permissions, then push your code to trigger the pipeline.
- Open the IAM console at https://console.aws.amazon.com/iam/.
- In the left sidebar, click Identity providers.
- Click Add provider.
- For Configure provider, choose OpenID Connect.
- For Provider URL, enter:
https://token.actions.githubusercontent.com- Click Get thumbprint.
- For Audience, enter:
sts.amazonaws.com- Click Add provider.
What is an OIDC identity provider?
An OIDC identity provider tells AWS to trust tokens issued by GitHub Actions. When a workflow runs, GitHub signs a short-lived JWT token. AWS verifies this token against the provider URL and grants temporary credentials without needing stored access keys.
Now create an IAM role that the GitHub workflow will assume.
- In the IAM console sidebar, click Roles.
- Click Create role.
- Select Web identity as the trusted entity type.
- For Identity provider, select token.actions.githubusercontent.com.
- For Audience, select sts.amazonaws.com.
- Click Next.
- Search for and attach these managed policies: AmazonEC2ContainerRegistryPowerUser and AWSLambda_FullAccess.
- Click Next.
- Name the role oil-well-mlops-github-actions.
- Click Create role.
After the role is created, you need to tighten its trust policy to only allow your specific repository.
- Click on the newly created oil-well-mlops-github-actions role.
- Click the Trust relationships tab.
- Click Edit trust policy.
- Replace the existing Condition block with the following (replacing the repo path with your own):
"Condition": {
"StringEquals": {
"token.actions.githubusercontent.com:aud": "sts.amazonaws.com"
},
"StringLike": {
"token.actions.githubusercontent.com:sub": "repo:[[GITHUB_USER="your-github-username"]]/[[GITHUB_REPO="your-repo-name"]]:*"
}
}- Click Update policy.
- Copy the role ARN from the role summary page. It looks like arn:aws:iam::123456789012:role/oil-well-mlops-github-actions.
Why restrict the sub claim?
Without a sub condition, any GitHub repository could assume your role. The StringLike condition with your repo path ensures only workflows from your specific repository can authenticate to your AWS account.
Now push the project to GitHub and add the role ARN as a secret.
- Create a new repository on GitHub (public or private). Do not add a README or .gitignore during creation.
- Initialize and push your local project by running these commands:
git init
git add .
git commit -m "Initial commit: ML pipeline with CI/CD"
git branch -M main
git remote add origin https://github.com/[[GITHUB_USER="your-github-username"]]/[[GITHUB_REPO="your-repo-name"]].git
git push -u origin main- In your GitHub repository, click Settings.
- In the Security section of the sidebar, expand Secrets and variables, then click Actions.
- Click the Secrets tab.
- Click New repository secret.
- In the Name field, enter AWS_ROLE_ARN.
- In the Secret field, paste the role ARN you copied earlier.
- Click Add secret.
The push you just made should have already triggered the workflow. Navigate to the Actions tab of your repository to watch the pipeline run.
Workflow failing at the deploy step?
The most common cause is a mismatch between the trust policy sub condition and your actual repository path. Make sure the value in the trust policy exactly matches the format repo:OWNER/REPO:* with your GitHub username and repository name.
Also verify the IAM role has the AmazonEC2ContainerRegistryPowerUser and AWSLambda_FullAccess policies attached.
Help me debug my GitHub Actions OIDC authentication failure with AWS
Your code now deploys itself. Every push to main runs your tests, builds a fresh container, and updates the live Lambda function automatically. Next up, you will add monitoring to detect when the model starts making poor predictions.
Implement Model Monitoring and Drift Detection
Your ML model is deployed and serving predictions, but how do you know it's still accurate?
In production, incoming data shifts over time. Reservoir conditions change, equipment ages, and the patterns your model learned during training may no longer reflect reality. In this step, you'll build a monitoring Lambda that runs daily, measures how much the data has drifted, and publishes metrics to CloudWatch. When performance drops below a threshold, an alarm fires automatically.
In this step, get ready to:
- Write a monitoring Lambda that calculates PSI for data drift and RMSE for model performance.
- Deploy the monitor as a second Lambda function and connect it to an EventBridge schedule.
- Verify custom CloudWatch metrics are published and the drift alarm is active.
Write the monitoring Lambda
The monitor needs to answer two questions every day: has the input data shifted away from what the model was trained on, and has the model's prediction accuracy degraded?
- Create a new file called monitor.py in your project root by running:
touch monitor.py- Confirm the file exists by running:
ls monitor.py- You should see monitor.py in the output.
- Open monitor.py in your editor and add the imports and client setup:
import json
import os
from datetime import datetime
import boto3
import numpy as np
# Initialize AWS clients for publishing metrics and reading model data
cloudwatch = boto3.client("cloudwatch")
s3_client = boto3.client("s3")
MODEL_BUCKET = os.environ.get("MODEL_BUCKET", "oil-well-mlops-models")What does this code do?
This sets up the AWS SDK clients for CloudWatch (to publish custom metrics) and S3 (to read training statistics). The MODEL_BUCKET environment variable tells the Lambda where to find model artifacts.
- Below the imports, add the PSI calculation function:
def calculate_psi(expected, actual, bins=10):
"""Calculate Population Stability Index for drift detection."""
breakpoints = np.linspace(
min(np.min(expected), np.min(actual)),
max(np.max(expected), np.max(actual)),
bins + 1,
)
expected_counts = np.histogram(expected, bins=breakpoints)[0] + 1
actual_counts = np.histogram(actual, bins=breakpoints)[0] + 1
expected_pct = expected_counts / expected_counts.sum()
actual_pct = actual_counts / actual_counts.sum()
psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
return float(psi)What is PSI?
Population Stability Index (PSI) measures how much the distribution of a feature has shifted between two time periods. It bins both distributions and compares their proportions using a logarithmic formula.
A PSI below 0.1 means little change. A PSI above 0.2 signals significant drift that likely requires retraining.
- Add the Lambda handler function below calculate_psi:
def handler(event, context):
"""Daily monitoring Lambda: check model performance and data drift."""
timestamp = datetime.utcnow().isoformat()
# Simulated training baseline statistics
training_stats = {"mean_oil_rate": 350.0, "std_oil_rate": 150.0}
recent_predictions = np.random.normal(340, 160, 100)
training_distribution = np.random.normal(
training_stats["mean_oil_rate"], training_stats["std_oil_rate"], 1000
)
# Calculate feature drift using PSI
psi = calculate_psi(training_distribution, recent_predictions)
# Simulate RMSE from comparing predictions to actuals
simulated_rmse = np.random.uniform(20, 60)
# Publish custom metrics to CloudWatch
cloudwatch.put_metric_data(
Namespace="OilWellMLOps",
MetricData=[
{"MetricName": "ModelRMSE", "Value": simulated_rmse, "Unit": "None", "Timestamp": timestamp},
{"MetricName": "FeatureDriftPSI", "Value": psi, "Unit": "None", "Timestamp": timestamp},
],
)
return {
"statusCode": 200,
"body": json.dumps({"timestamp": timestamp, "rmse": simulated_rmse, "psi": psi, "drift_detected": psi > 0.2, "retraining_needed": simulated_rmse > 50}),
}What does the handler do?
Each day, this handler simulates comparing recent prediction data against training baselines. It calculates the PSI for feature drift and measures RMSE for prediction accuracy.
Both metrics get published to the OilWellMLOps custom namespace in CloudWatch. The alarm you created in Terraform (threshold of 50 on ModelRMSE) will trigger automatically when performance degrades.
- Save monitor.py.
Seeing import errors?
Make sure you have numpy and boto3 installed in your local environment. These are already in your requirements.txt from earlier steps.
Help me troubleshoot import errors in my monitor.py file
✔️ Awesome, I've got everything!
Great. Double check you've saved monitor.py in your project root.
ⓧ I'd like to double check the full code
import json
import os
from datetime import datetime
import boto3
import numpy as np
cloudwatch = boto3.client("cloudwatch")
s3_client = boto3.client("s3")
MODEL_BUCKET = os.environ.get("MODEL_BUCKET", "oil-well-mlops-models")
def calculate_psi(expected, actual, bins=10):
"""Calculate Population Stability Index for drift detection."""
breakpoints = np.linspace(
min(np.min(expected), np.min(actual)),
max(np.max(expected), np.max(actual)),
bins + 1,
)
expected_counts = np.histogram(expected, bins=breakpoints)[0] + 1
actual_counts = np.histogram(actual, bins=breakpoints)[0] + 1
expected_pct = expected_counts / expected_counts.sum()
actual_pct = actual_counts / actual_counts.sum()
psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
return float(psi)
def handler(event, context):
"""Daily monitoring Lambda: check model performance and data drift."""
timestamp = datetime.utcnow().isoformat()
training_stats = {"mean_oil_rate": 350.0, "std_oil_rate": 150.0}
recent_predictions = np.random.normal(340, 160, 100)
training_distribution = np.random.normal(
training_stats["mean_oil_rate"], training_stats["std_oil_rate"], 1000
)
psi = calculate_psi(training_distribution, recent_predictions)
simulated_rmse = np.random.uniform(20, 60)
cloudwatch.put_metric_data(
Namespace="OilWellMLOps",
MetricData=[
{"MetricName": "ModelRMSE", "Value": simulated_rmse, "Unit": "None", "Timestamp": timestamp},
{"MetricName": "FeatureDriftPSI", "Value": psi, "Unit": "None", "Timestamp": timestamp},
],
)
return {
"statusCode": 200,
"body": json.dumps({"timestamp": timestamp, "rmse": simulated_rmse, "psi": psi, "drift_detected": psi > 0.2, "retraining_needed": simulated_rmse > 50}),
}Deploy the monitor Lambda to AWS
Unlike the inference Lambda (which uses a Docker container), the monitor is lightweight enough to deploy as a zip package. It needs numpy bundled in, but boto3 is already included in the Lambda Python runtime.
- Create a deployment package directory and install a Lambda-compatible numpy into it by running:
mkdir -p monitor_package
pip install --platform manylinux2014_x86_64 --target monitor_package/ --python-version 3.12 --only-binary=:all: numpy
cp monitor.py monitor_package/Why the platform flag?
Lambda runs on Amazon Linux. The --platform manylinux2014_x86_64 flag tells pip to download the Linux-compatible binary wheel for numpy, even if you're on macOS or Windows. Without it, you'd get a wheel compiled for your local OS that won't work in Lambda.
- Zip the deployment package by running:
cd monitor_package && zip -r ../monitor.zip . && cd ..- Confirm the zip was created by running:
ls -lh monitor.zip- You should see monitor.zip listed with a size around 30-40 MB.
- Get your IAM role ARN for the Lambda by running:
aws iam get-role --role-name oil-well-mlops-lambda-role --query 'Role.Arn' --output text- Copy the ARN from the output. You'll use it in the next command.
- Create the monitor Lambda function by running:
aws lambda create-function \
--function-name oil-well-mlops-monitor \
--runtime python3.12 \
--handler monitor.handler \
--role [[LAMBDA_ROLE_ARN="the role ARN you copied above e.g. arn:aws:iam::123456789012:role/oil-well-mlops-lambda-role"]] \
--zip-file fileb://monitor.zip \
--timeout 30 \
--memory-size 256 \
--environment Variables={MODEL_BUCKET=[[MODEL_BUCKET="your models bucket name e.g. oil-well-mlops-models-123456789012"]]}What does this command do?
This creates a new Lambda function called oil-well-mlops-monitor using the Python 3.12 runtime. The --handler monitor.handler flag tells Lambda to call the handler function inside monitor.py.
The same IAM role from your inference Lambda provides the necessary S3 and CloudWatch permissions.
- You should see JSON output showing the function details with a State of Active or Pending.
Seeing a role not found error?
Make sure the role name is exactly oil-well-mlops-lambda-role. This is the role Terraform created in Step 5. You can list matching roles to find it:
aws iam list-roles --query "Roles[?contains(RoleName, 'oil-well')]"Help me fix this role error
Connect EventBridge and verify the alarm
The EventBridge rule oil-well-mlops-daily-monitor already exists from your Terraform deployment. You need to connect it to your new monitor Lambda so it triggers the function once per day.
- Grant EventBridge permission to invoke the monitor Lambda by running:
aws lambda add-permission \
--function-name oil-well-mlops-monitor \
--statement-id EventBridgeInvoke \
--action lambda:InvokeFunction \
--principal events.amazonaws.com \
--source-arn $(aws events describe-rule --name oil-well-mlops-daily-monitor --query 'Arn' --output text)Why add a permission?
Lambda functions use resource-based policies to control who can invoke them. This command adds a policy statement that says "the events.amazonaws.com service is allowed to invoke this function, but only from this specific EventBridge rule."
- Add the monitor Lambda as the target of the EventBridge rule by running:
aws events put-targets \
--rule oil-well-mlops-daily-monitor \
--targets "Id"="1","Arn"="$(aws lambda get-function --function-name oil-well-mlops-monitor --query 'Configuration.FunctionArn' --output text)"- Verify the target was added by running:
aws events list-targets-by-rule --rule oil-well-mlops-daily-monitor- You should see JSON output containing your Lambda function ARN in the Targets array.
Now test the monitor manually to confirm it publishes metrics.
- Invoke the monitor function manually by running:
aws lambda invoke \
--function-name oil-well-mlops-monitor \
--cli-binary-format raw-in-base64-out \
--payload '{}' \
monitor_response.json && cat monitor_response.json- You should see output containing rmse, psi, drift_detected, and retraining_needed fields.
- Verify the CloudWatch alarm exists and is monitoring the correct metric by running:
aws cloudwatch describe-alarms --alarm-names oil-well-mlops-model-drift- You should see the alarm details showing MetricName as ModelRMSE, Namespace as OilWellMLOps, and Threshold of 50.
Lambda invocation returning an error?
Check the CloudWatch logs for the function. Confirm the log group exists, then look at recent log streams for error details:
aws logs describe-log-groups --log-group-name-prefix /aws/lambda/oil-well-mlops-monitorA common cause is a missing or incompatible numpy binary. Make sure you used the --platform manylinux2014_x86_64 flag when installing numpy into the deployment package.
Help me debug my monitor Lambda invocation error
- Push monitor.py to your GitHub repository by running:
git add monitor.py
git commit -m "Add model monitoring Lambda with PSI drift detection"
git push origin mainYour model now has eyes. Every day the monitor checks for drift and publishes metrics. If the RMSE crosses the threshold of 50, the alarm fires. Next up, you'll build the automatic retraining pipeline that responds to that alarm.
Build the Automatic Retraining Pipeline
Your monitoring system detects when the model drifts. But detection alone is not enough.
In this step, you will build the automated response: a retraining Lambda that fires when the CloudWatch alarm triggers. This Lambda fetches the latest production data, retrains the XGBoost model, evaluates it against the baseline, and promotes the new version if it improves performance.
In this step, get ready to:
- Write a retraining Lambda handler that fetches data, retrains, and registers new model versions.
- Deploy the retraining Lambda as a container image to AWS.
- Connect the CloudWatch alarm to trigger automatic retraining and verify the full loop.
Write the retraining Lambda handler
The retraining Lambda is the automated response to model drift. When RMSE exceeds the threshold, this function runs end-to-end: fetching fresh data, training a new model, and promoting it to production.
- Create a new file called retrain.py in your project directory by running:
touch retrain.py- Open retrain.py and add the imports, client setup, and version-management helper:
import json
import os
import pickle
import tempfile
from datetime import datetime
import boto3
import numpy as np
import pandas as pd
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from xgboost import XGBRegressor
from feature_engineering import engineer_features, create_forecast_targets
# AWS clients
s3_client = boto3.client("s3")
lambda_client = boto3.client("lambda")
dynamodb = boto3.resource("dynamodb")
# Environment variables set on the Lambda
DATA_BUCKET = os.environ.get("DATA_BUCKET", "oil-well-mlops-data")
MODEL_BUCKET = os.environ.get("MODEL_BUCKET", "oil-well-mlops-models")
INFERENCE_FUNCTION = os.environ.get("INFERENCE_FUNCTION", "oil-well-mlops-inference")
TABLE_NAME = os.environ.get("TABLE_NAME", "oil-well-mlops-predictions")
# Same feature set used during training
FEATURE_COLS = [
"days_on_production", "oil_rate", "gas_rate", "water_cut",
"reservoir_pressure", "wellhead_pressure", "choke_size",
"porosity", "permeability",
"oil_rate_lag_1", "oil_rate_lag_7", "oil_rate_lag_30",
"gas_rate_lag_1", "gas_rate_lag_7", "gas_rate_lag_30",
"pressure_lag_1", "pressure_lag_7", "pressure_lag_30",
"oil_rate_rolling_mean_7", "oil_rate_rolling_std_7",
"oil_rate_rolling_mean_30", "oil_rate_rolling_std_30",
"pressure_rolling_mean_7", "pressure_rolling_mean_30",
"oil_rate_change_7d", "pressure_change_7d",
"cumulative_oil", "drawdown", "month", "quarter",
]
TARGET_COL = "target_30d"
def get_next_version():
"""Determine next model version by listing existing models in S3."""
response = s3_client.list_objects_v2(
Bucket=MODEL_BUCKET, Prefix="models/model_v"
)
versions = []
for obj in response.get("Contents", []):
key = obj["Key"]
if key.endswith(".pkl"):
version_str = key.split("model_v")[1].replace(".pkl", "")
try:
versions.append(int(version_str))
except ValueError:
continue
return max(versions, default=0) + 1What does this code do?
This top section sets up the AWS clients (S3, Lambda, DynamoDB) and defines the same feature columns used during initial training. Keeping FEATURE_COLS identical ensures the retrained model expects the same input shape.
The get_next_version() function scans S3 for existing model files (model_v1.pkl, model_v2.pkl, etc.) and returns the next version number. This gives you automatic model versioning without a separate registry service.
- Below the get_next_version function, add the main handler that performs the full retraining flow:
def handler(event, context):
"""Retraining Lambda: retrain model when drift alarm fires."""
timestamp = datetime.utcnow().isoformat()
# Step 1: Fetch latest production data from S3
tmp_data = os.path.join(tempfile.gettempdir(), "production_data.csv")
s3_client.download_file(DATA_BUCKET, "production_data.csv", tmp_data)
df = pd.read_csv(tmp_data)
# Step 2: Engineer features and create forecast targets
df = engineer_features(df)
df = create_forecast_targets(df, horizons=[7, 30, 90])
# Step 3: Time-based train/test split
df = df.sort_values("date").reset_index(drop=True)
split_idx = int(len(df) * 0.8)
train_df = df.iloc[:split_idx]
test_df = df.iloc[split_idx:]
X_train = train_df[FEATURE_COLS]
y_train = train_df[TARGET_COL]
X_test = test_df[FEATURE_COLS]
y_test = test_df[TARGET_COL]
# Step 4: Train new XGBoost model with same hyperparameters
model = XGBRegressor(
n_estimators=200,
max_depth=6,
learning_rate=0.1,
subsample=0.8,
colsample_bytree=0.8,
random_state=42,
)
model.fit(X_train, y_train, eval_set=[(X_test, y_test)], verbose=False)
# Step 5: Evaluate new model
y_pred = model.predict(X_test)
metrics = {
"rmse": float(np.sqrt(mean_squared_error(y_test, y_pred))),
"mae": float(mean_absolute_error(y_test, y_pred)),
"r2": float(r2_score(y_test, y_pred)),
}
# Step 6: Load baseline RMSE from previous model metadata
baseline_rmse = 999.0
try:
response = s3_client.get_object(
Bucket=MODEL_BUCKET, Key="models/model_v1_metadata.json"
)
baseline_meta = json.loads(response["Body"].read())
baseline_rmse = baseline_meta.get("rmse", 999.0)
except Exception:
pass
# Step 7: Register new model version in S3
new_version = get_next_version()
version_tag = f"v{new_version}"
model_key = f"models/model_{version_tag}.pkl"
tmp_model = os.path.join(tempfile.gettempdir(), "model.pkl")
with open(tmp_model, "wb") as f:
pickle.dump(model, f)
s3_client.upload_file(tmp_model, MODEL_BUCKET, model_key)
# Save metadata alongside the model
metadata = {
"version": version_tag,
"rmse": metrics["rmse"],
"mae": metrics["mae"],
"r2": metrics["r2"],
"baseline_rmse": baseline_rmse,
"improved": metrics["rmse"] < baseline_rmse,
"trained_at": timestamp,
"train_samples": len(X_train),
"test_samples": len(X_test),
}
metadata_key = f"models/model_{version_tag}_metadata.json"
s3_client.put_object(
Bucket=MODEL_BUCKET,
Key=metadata_key,
Body=json.dumps(metadata, indent=2),
)
# Step 8: Update inference Lambda to use new model
lambda_client.update_function_configuration(
FunctionName=INFERENCE_FUNCTION,
Environment={
"Variables": {
"MODEL_BUCKET": MODEL_BUCKET,
"MODEL_KEY": model_key,
}
},
)
# Step 9: Log retraining lineage to DynamoDB
table = dynamodb.Table(TABLE_NAME)
table.put_item(
Item={
"prediction_id": f"retrain_{version_tag}",
"timestamp": timestamp,
"event_type": "retraining",
"version": version_tag,
"rmse": str(metrics["rmse"]),
"mae": str(metrics["mae"]),
"r2": str(metrics["r2"]),
"baseline_rmse": str(baseline_rmse),
"improved": metadata["improved"],
"model_key": model_key,
}
)
return {
"statusCode": 200,
"body": json.dumps({
"message": "Retraining complete",
"version": version_tag,
"metrics": metrics,
"improved": metadata["improved"],
}),
}What does the handler do?
The handler runs the full retraining lifecycle in sequence:
- Fetches production_data.csv from the data bucket and runs the same feature engineering pipeline.
- Trains a new XGBoost model with identical hyperparameters and a time-based 80/20 split.
- Evaluates against the baseline RMSE from the V1 metadata. Even if the new model is worse, it still gets registered for auditability.
- Uploads the new .pkl and metadata JSON to S3 with an incremented version tag.
- Updates the inference Lambda's MODEL_KEY environment variable so new predictions use the latest model.
- Writes a lineage record to DynamoDB for traceability.
- Save retrain.py.
Import errors when testing locally?
Make sure feature_engineering.py is in the same directory as retrain.py. The retraining handler imports engineer_features and create_forecast_targets directly from it.
Help me debug my import errors
Deploy the retraining Lambda to AWS
The retraining Lambda needs the same heavy ML dependencies (XGBoost, pandas, scikit-learn) as the inference Lambda. This means it also needs container-based deployment since these packages exceed the 250 MB zip limit.
- Create a Dockerfile.retrain that packages the retraining handler with its dependencies:
FROM public.ecr.aws/lambda/python:3.12
COPY requirements.txt ${LAMBDA_TASK_ROOT}/
RUN pip install --no-cache-dir -r requirements.txt
# Copy retrain handler and its dependency
COPY retrain.py ${LAMBDA_TASK_ROOT}/
COPY feature_engineering.py ${LAMBDA_TASK_ROOT}/
CMD ["retrain.handler"]Why a separate Dockerfile?
The inference Lambda's Dockerfile only copies lambda_handler.py. The retrain Lambda needs retrain.py plus feature_engineering.py (for the pipeline functions). Keeping them as separate images means each Lambda loads only what it needs.
- Build the retraining container image by running:
docker build --provenance=false -f Dockerfile.retrain -t oil-well-mlops-retrain .- Tag the image for your ECR repository and push it by running these commands:
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin [[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]].dkr.ecr.us-east-1.amazonaws.com
docker tag oil-well-mlops-retrain:latest [[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]].dkr.ecr.us-east-1.amazonaws.com/oil-well-mlops-inference:retrain
docker push [[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]].dkr.ecr.us-east-1.amazonaws.com/oil-well-mlops-inference:retrainWhy push to the same ECR repository?
You are pushing to the existing oil-well-mlops-inference ECR repository but with a different tag (retrain instead of latest). This avoids creating a second repository and keeps you within the 500 MB ECR free tier limit.
- Create the retraining Lambda function by running:
aws lambda create-function \
--function-name oil-well-mlops-retrain \
--package-type Image \
--code ImageUri=[[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]].dkr.ecr.us-east-1.amazonaws.com/oil-well-mlops-inference:retrain \
--role arn:aws:iam::[[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]]:role/oil-well-mlops-lambda-role \
--timeout 300 \
--memory-size 1024 \
--environment "Variables={DATA_BUCKET=oil-well-mlops-data-[[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]],MODEL_BUCKET=oil-well-mlops-models-[[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]],INFERENCE_FUNCTION=oil-well-mlops-inference,TABLE_NAME=oil-well-mlops-predictions}"Why 300 seconds and 1024 MB?
Retraining an XGBoost model on 10,000+ rows with feature engineering takes more time and memory than serving a single prediction. The 300-second timeout and 1024 MB memory give the function enough headroom to complete the full pipeline without hitting limits.
- Verify the Lambda was created successfully by running:
aws lambda get-function --function-name oil-well-mlops-retrain --query "Configuration.State"You should see "Active" in the output, confirming the function is ready.
Seeing a Pending state?
Container-based Lambda functions take a moment to initialize after creation. Wait 30 seconds and run the get-function command again. If it stays in Pending, check that the ECR image was pushed successfully with aws ecr describe-images --repository-name oil-well-mlops-inference.
Help me debug my retraining Lambda deployment
Connect the CloudWatch alarm and test the pipeline
The final piece is wiring the alarm to the Lambda. When the oil-well-mlops-model-drift alarm enters ALARM state, CloudWatch will directly invoke the retraining function.
- Grant CloudWatch permission to invoke the retraining Lambda by running:
aws lambda add-permission \
--function-name oil-well-mlops-retrain \
--statement-id AlarmAction \
--action lambda:InvokeFunction \
--principal lambda.alarms.cloudwatch.amazonaws.com \
--source-account [[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]]What does this permission do?
Lambda functions need explicit resource-based policies to allow external services to invoke them. The lambda.alarms.cloudwatch.amazonaws.com principal is the service identity that CloudWatch uses when firing alarm actions. Without this permission, the alarm would trigger but the invocation would be denied.
- Update the existing CloudWatch alarm to add the Lambda as an alarm action by running:
aws cloudwatch put-metric-alarm \
--alarm-name oil-well-mlops-model-drift \
--alarm-actions arn:aws:lambda:us-east-1:[[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]]:function:oil-well-mlops-retrain \
--metric-name ModelRMSE \
--namespace OilWellMLOps \
--statistic Average \
--period 86400 \
--evaluation-periods 1 \
--threshold 50 \
--comparison-operator GreaterThanThresholdWhy re-specify all alarm parameters?
The put-metric-alarm command overwrites the entire alarm configuration. You must include all the original settings (metric name, namespace, threshold, period) alongside the new --alarm-actions parameter. Omitting any field resets it to the default.
- Test the full pipeline by manually setting the alarm to ALARM state:
aws cloudwatch set-alarm-state \
--alarm-name oil-well-mlops-model-drift \
--state-value ALARM \
--state-reason "Manual test of retraining trigger"- Wait about 60 seconds for the Lambda to execute, then verify a new model was registered in S3 by running:
aws s3 ls s3://oil-well-mlops-models-[[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]]/models/You should see model_v2.pkl and model_v2_metadata.json alongside the original V1 files.
- Confirm the inference Lambda now points to the new model by running:
aws lambda get-function-configuration \
--function-name oil-well-mlops-inference \
--query "Environment.Variables.MODEL_KEY"You should see "models/model_v2.pkl" confirming the inference function will load the retrained model on next invocation.
- Verify the lineage record was written to DynamoDB by running:
aws dynamodb scan \
--table-name oil-well-mlops-predictions \
--filter-expression "begins_with(prediction_id, :prefix)" \
--expression-attribute-values '{":prefix": {"S": "retrain_"}}'You should see the full retraining record including the version, metrics, and whether the new model improved over baseline.
Not seeing model_v2.pkl in S3?
Check the Lambda execution logs for errors by running:
aws logs tail /aws/lambda/oil-well-mlops-retrain --since 5mCommon issues: the DATA_BUCKET environment variable does not match your actual bucket name, or production_data.csv has not been uploaded to S3. Verify the data file exists with aws s3 ls against your data bucket.
Help me debug my retraining pipeline
Your MLOps pipeline now has a complete feedback loop. Drift detection triggers retraining, retraining produces a versioned model, and the inference endpoint automatically picks up the latest version. Next up, you will push these final changes and confirm the full system is operational end-to-end.
Secret mission
Equipment Failure Risk Prediction
Can you add predictive maintenance to your MLOps platform? In production oil fields, pump failures cost thousands per hour of downtime. Build a binary classifier that predicts equipment failures 24 hours ahead using rolling sensor statistics, deploy it as a second API endpoint, and document which features matter most using SHAP explainability.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. All resources in this project stay within AWS Free Tier limits with normal usage (a few invocations per day).
Cost warning
Some AWS Free Tier benefits (S3 storage, ECR storage, API Gateway calls) expire after 12 months. If your AWS account is older than 12 months, leaving these resources running may incur small charges.
Check your AWS Billing dashboard to confirm your free tier status.
Resources you used:
- AWS S3 buckets (oil-well-mlops-data-<account_id> and oil-well-mlops-models-<account_id>).
- AWS ECR repository (oil-well-mlops-inference).
- AWS Lambda functions (oil-well-mlops-inference, oil-well-mlops-monitor, oil-well-mlops-retrain).
- AWS API Gateway (oil-well-mlops-api).
- AWS DynamoDB table (oil-well-mlops-predictions).
- AWS CloudWatch alarm, dashboard, and metrics.
- AWS EventBridge rule (oil-well-mlops-daily-monitor).
- AWS IAM role (oil-well-mlops-lambda-role).
- GitHub repository.
- Local project files on your Desktop.
Keep everything running
No action needed. Choose this if you are still actively building or experimenting with the pipeline.
- Lambda only charges on invocation, so idle functions cost nothing.
- S3 storage stays within the 5 GB free tier for this project's data.
- The EventBridge rule triggers the monitor Lambda once per day, which counts toward your 1,000,000 free invocations per month.
- Your GitHub repository remains available at no cost.
Pause - I'll come back to this later
Shut down scheduled triggers to stop automated invocations while keeping all resources intact for later.
- Disable the EventBridge rule in the AWS Console to stop the daily monitoring Lambda from running.
- No other action is needed. Lambda functions, S3 buckets, and DynamoDB tables do not charge while idle.
- When you are ready to resume, re-enable the EventBridge rule.
Delete - I don't want to use this again
Remove all project resources and start fresh. This cannot be undone.
- Empty both S3 buckets first (Terraform cannot delete non-empty buckets). Run these commands in your terminal:
aws s3 rm s3://oil-well-mlops-data-[[AWS_ACCOUNT_ID="your AWS account ID"]] --recursive
aws s3 rm s3://oil-well-mlops-models-[[AWS_ACCOUNT_ID="your AWS account ID"]] --recursive- Destroy all AWS infrastructure managed by Terraform by running:
cd ~/Desktop/oil-well-mlops
terraform destroy- Terraform will display a plan showing every resource it will remove. Type yes when prompted to confirm.
- Delete your local project files:
macOS
cd ~/Desktop
rm -rf oil-well-mlopsWindows
cd ~\Desktop
Remove-Item -Recurse -Force oil-well-mlops- Delete the GitHub repository by going to your repository page, clicking Settings, scrolling to the Danger Zone section, and clicking Delete this repository.
Well Done!
Well Done!
Nice work. You just built what MLOps engineers at oil companies get paid six figures to operate.
You now have a production-grade serverless MLOps pipeline that forecasts oil well production decline, monitors its own performance, and automatically retrains when accuracy drops.
Here's what you learned:
- Trained an XGBoost production forecasting model with domain-specific feature engineering (lag features, rolling statistics, reservoir parameters) and deployed it as a containerized Lambda function behind API Gateway.
- Managed 7+ AWS services with Terraform Infrastructure as Code and automated the entire build-test-deploy cycle with a GitHub Actions CI/CD pipeline using OIDC federation.
- Implemented model monitoring with Population Stability Index (PSI) drift detection and CloudWatch alarms that trigger automated retraining when RMSE exceeds the performance threshold.
- Secret Mission: Added an equipment failure prediction system using an XGBoost classifier with rolling pressure variance features and SHAP explainability, served through a second API endpoint.
Ready to quiz yourself?