This notebook creates a SageMaker Pipeline to build an end-to-end Machine Learning system to solve the problem of classifying penguin species. With a SageMaker Pipeline, you can create, automate, and manage end-to-end Machine Learning workflows at scale.
The machine learning system we’ll build during this program consists of four main pipelines: A training pipeline, an inference pipeline, a deployment pipeline, and a monitoring pipeline.
+
Here is an architectural diagram showing how the system is structured:
+
+
Throughout the sessions, we’ll build each of these pipelines. We’ll also build variations to show you different alternatives and best practices.
+
Let’s start by setting up the environment and preparing to run the notebook.
+
We can run this notebook in Local Mode to test some of the system components in your local environment. Unfortunately, not every component is supported in Local Mode.
+
Setting the LOCAL_MODE variable to True will run every supported pipeline component locally. Setting the variable to False will run the pipeline in SageMaker.
+
+
LOCAL_MODE =True
+
+
Let’s now load the environment variables we need to run the notebook.
If you are running the pipeline in Local Mode on an ARM64 machine (for example, on Apple Silicon), you will need to use a custom Docker image to train and evaluate the model. Let’s create a variable indicating if we are running on an ARM64 machine.
+
+
# We can retrieve the architecture of the local
+# computer using the `uname -m` command.
+architecture =!(uname -m)
+
+IS_ARM64_ARCHITECTURE = architecture[0] =="arm64"
+
+
Let’s create a configuration dictionary with different settings depending on whether we are running the pipeline in Local Mode. We’ll use this dictionary to configure the pipeline components.
+
+
import sagemaker
+from sagemaker.workflow.pipeline_context import LocalPipelineSession, PipelineSession
+
+pipeline_session = PipelineSession(default_bucket=bucket) ifnot LOCAL_MODE elseNone
+
+if LOCAL_MODE:
+ config = {
+"session": LocalPipelineSession(default_bucket=bucket),
+"instance_type": "local",
+# We need to use a custom Docker image when we run the pipeline
+# in Local Model on an ARM64 machine.
+"image": (
+"sagemaker-tensorflow-toolkit-local"if IS_ARM64_ARCHITECTURE elseNone
+ ),
+ }
+else:
+ config = {
+"session": pipeline_session,
+"instance_type": "ml.m5.xlarge",
+"image": None,
+ }
+
+# These specific settings refer to the SageMaker
+# TensorFlow container we'll use.
+config["framework_version"] ="2.12"
+config["py_version"] ="py310"
+
+
Let’s now initialize a few variables that we’ll need throughout the notebook:
Let’s run Exploratory Data Analysis on the Penguins dataset. The goal of this session is to understand the data and the problem we are trying to solve.
+
Let’s load the Penguins dataset:
+
+
import numpy as np
+import pandas as pd
+
+penguins = pd.read_csv(DATA_FILEPATH)
+penguins.head()
+
+
+
+
+
+
+
+
+
+
species
+
island
+
culmen_length_mm
+
culmen_depth_mm
+
flipper_length_mm
+
body_mass_g
+
sex
+
+
+
+
+
0
+
Adelie
+
Torgersen
+
39.1
+
18.7
+
181.0
+
3750.0
+
MALE
+
+
+
1
+
Adelie
+
Torgersen
+
39.5
+
17.4
+
186.0
+
3800.0
+
FEMALE
+
+
+
2
+
Adelie
+
Torgersen
+
40.3
+
18.0
+
195.0
+
3250.0
+
FEMALE
+
+
+
3
+
Adelie
+
Torgersen
+
NaN
+
NaN
+
NaN
+
NaN
+
NaN
+
+
+
4
+
Adelie
+
Torgersen
+
36.7
+
19.3
+
193.0
+
3450.0
+
FEMALE
+
+
+
+
+
+
+
+
+
We can see the dataset contains the following columns:
+
+
species: The species of a penguin. This is the column we want to predict.
+
island: The island where the penguin was found
+
culmen_length_mm: The length of the penguin’s culmen (bill) in millimeters
+
culmen_depth_mm: The depth of the penguin’s culmen in millimeters
+
flipper_length_mm: The length of the penguin’s flipper in millimeters
+
body_mass_g: The body mass of the penguin in grams
+
sex: The sex of the penguin
+
+
If you are curious, here is the description of a penguin’s culmen:
+
+
Now, let’s get the summary statistics for the features in our dataset.
+
+
penguins.describe(include="all")
+
+
+
+
+
+
+
+
+
+
species
+
island
+
culmen_length_mm
+
culmen_depth_mm
+
flipper_length_mm
+
body_mass_g
+
sex
+
+
+
+
+
count
+
344
+
344
+
342.000000
+
342.000000
+
342.000000
+
342.000000
+
334
+
+
+
unique
+
3
+
3
+
NaN
+
NaN
+
NaN
+
NaN
+
3
+
+
+
top
+
Adelie
+
Biscoe
+
NaN
+
NaN
+
NaN
+
NaN
+
MALE
+
+
+
freq
+
152
+
168
+
NaN
+
NaN
+
NaN
+
NaN
+
168
+
+
+
mean
+
NaN
+
NaN
+
43.921930
+
17.151170
+
200.915205
+
4201.754386
+
NaN
+
+
+
std
+
NaN
+
NaN
+
5.459584
+
1.974793
+
14.061714
+
801.954536
+
NaN
+
+
+
min
+
NaN
+
NaN
+
32.100000
+
13.100000
+
172.000000
+
2700.000000
+
NaN
+
+
+
25%
+
NaN
+
NaN
+
39.225000
+
15.600000
+
190.000000
+
3550.000000
+
NaN
+
+
+
50%
+
NaN
+
NaN
+
44.450000
+
17.300000
+
197.000000
+
4050.000000
+
NaN
+
+
+
75%
+
NaN
+
NaN
+
48.500000
+
18.700000
+
213.000000
+
4750.000000
+
NaN
+
+
+
max
+
NaN
+
NaN
+
59.600000
+
21.500000
+
231.000000
+
6300.000000
+
NaN
+
+
+
+
+
+
+
+
+
Let’s now display the distribution of values for the three categorical columns in our data:
The distribution of the categories in our data are:
+
+
species: There are 3 species of penguins in the dataset: Adelie (152), Gentoo (124), and Chinstrap (68).
+
island: Penguins are from 3 islands: Biscoe (168), Dream (124), and Torgersen (52).
+
sex: We have 168 male penguins, 165 female penguins, and 1 penguin with an ambiguous gender (.).
+
+
Let’s replace the ambiguous value in the sex column with a null value:
+
+
penguins["sex"] = penguins["sex"].replace(".", np.nan)
+
+# Let's display the new distribution of the column:
+sex_distribution = penguins["sex"].value_counts()
+sex_distribution
+
+
sex
+MALE 168
+FEMALE 165
+Name: count, dtype: int64
+
+
+
Next, let’s check for any missing values in the dataset.
Let’s get rid of the missing values. For now, we are going to replace the missing values with the most frequent value in the column. Later, we’ll use a different strategy to replace missing numeric values.
+
+
from sklearn.impute import SimpleImputer
+
+imputer = SimpleImputer(strategy="most_frequent")
+penguins.iloc[:, :] = imputer.fit_transform(penguins)
+
+# Let's display again the number of missing values:
+penguins.isna().sum()
Let’s display the covariance matrix of the dataset. The “covariance” measures how changes in one variable are associated with changes in a second variable. In other words, the covariance measures the degree to which two variables are linearly associated.
+
+
penguins.cov(numeric_only=True)
+
+
+
+
+
+
+
+
+
+
culmen_length_mm
+
culmen_depth_mm
+
flipper_length_mm
+
body_mass_g
+
+
+
+
+
culmen_length_mm
+
29.679415
+
-2.516984
+
50.260588
+
2596.971151
+
+
+
culmen_depth_mm
+
-2.516984
+
3.877201
+
-16.108849
+
-742.660180
+
+
+
flipper_length_mm
+
50.260588
+
-16.108849
+
197.269501
+
9792.552037
+
+
+
body_mass_g
+
2596.971151
+
-742.660180
+
9792.552037
+
640316.716388
+
+
+
+
+
+
+
+
+
Here are three examples of what we get from interpreting the covariance matrix below:
+
+
The positive covariance of 50.26 between culmen length and flippler length suggests that larger values of culmen length are associated with larger values of flipper length. As one increases, generally so does the other.
+
The positive covariance of 2596.97 between culmen length and body mass suggests that heavier penguins generally have longer culmens. There is a tendency for these two variables to increase together.
+
The negative covariance of -742.66 between culmen depth and body mass suggests a general tendency that penguins with deeper culmens weigh less.
+
+
Let’s now display the correlation matrix. “Correlation” measures both the strength and direction of the linear relationship between two variables:
+
+
penguins.corr(numeric_only=True)
+
+
+
+
+
+
+
+
+
+
culmen_length_mm
+
culmen_depth_mm
+
flipper_length_mm
+
body_mass_g
+
+
+
+
+
culmen_length_mm
+
1.000000
+
-0.234635
+
0.656856
+
0.595720
+
+
+
culmen_depth_mm
+
-0.234635
+
1.000000
+
-0.582472
+
-0.471339
+
+
+
flipper_length_mm
+
0.656856
+
-0.582472
+
1.000000
+
0.871302
+
+
+
body_mass_g
+
0.595720
+
-0.471339
+
0.871302
+
1.000000
+
+
+
+
+
+
+
+
+
Here are three examples of what we get from interpreting the correlation matrix below:
+
+
Penguins that weight more tend to have longer flippers.
+
Penguins with a shallower culmen tend to have longer flippers.
+
Penguins with longer culmens tend to have longer flippers.
+
+
Let’s display the distribution of species by island:
+
+
unique_species = penguins["species"].unique()
+
+fig, ax = plt.subplots(figsize=(6, 6))
+for species in unique_species:
+ data = penguins[penguins["species"] == species]
+ ax.hist(data["island"], bins=5, alpha=0.5, label=species)
+
+ax.set_xlabel("Island")
+ax.set_ylabel("Count")
+ax.set_title("Distribution of Species by Island")
+ax.legend()
+plt.show()
+
+
+
+
+
+
+
+
+
Let’s display the distribution of species by sex.
+
+
fig, ax = plt.subplots(figsize=(6, 6))
+
+for species in unique_species:
+ data = penguins[penguins["species"] == species]
+ ax.hist(data["sex"], bins=3, alpha=0.5, label=species)
+
+ax.set_xlabel("Sex")
+ax.set_ylabel("Count")
+ax.set_title("Distribution of Species by Sex")
+
+ax.legend()
+plt.show()
+
+
+
+
+
+
+
+
+
+
+
Session 3 - Splitting and Transforming the Data
+
In this session we’ll build a simple SageMaker Pipeline with one step to split and transform the data:
The first step we need in the pipeline is a Processing Step to run a script that will split and transform the data.
+
This Processing Step will create a SageMaker Processing Job in the background, run the script, and upload the output to S3. You can use Processing Jobs to perform data preprocessing, post-processing, feature engineering, data validation, and model evaluation. Check the ProcessingStep SageMaker’s SDK documentation for more information.
+
We will store the script in a folder called processing and add it to the system path so we can later import it as a module.
Several SageMaker Pipeline steps support caching. When a step runs, and dependending on the configured caching policy, SageMaker will try to reuse the result of a previous successful run of the same step. You can find more information about this topic in Caching Pipeline Steps.
+
Let’s define a caching policy that we’ll reuse on every step:
+
+
from sagemaker.workflow.steps import CacheConfig
+
+cache_config = CacheConfig(enable_caching=True, expire_after="15d")
+
+
+
+
Step 3 - Pipeline Configuration
+
We can parameterize a SageMaker Pipeline to make it more flexible. In this case, we’ll use a parameter to pass the location of the dataset we want to process. We can execute the pipeline with different datasets by changing the value of this parameter. Check Pipeline Parameters for more information.
Let’s now define the ProcessingStep that we’ll use in the pipeline to run the script that will split and transform the data.
+
A processor gives the Processing Step information about the hardware and software that SageMaker should use to launch a Processing Job. To run the script we created, we need access to Scikit-Learn, so we can use the SKLearnProcessor processor that comes out-of-the-box with the SageMaker’s Python SDK.
+
SageMaker manages the infrastructure of a Processing Job. It provisions resources for the duration of the job, and cleans up when it completes. The Processing Container image that SageMaker uses to run a Processing Job can either be a SageMaker built-in image or a custom image:
from sagemaker.sklearn.processing import SKLearnProcessor
+
+processor = SKLearnProcessor(
+ base_job_name="preprocess-data",
+ framework_version="1.2-1",
+# By default, a new account doesn't have access to `ml.m5.xlarge` instances.
+# If you haven't requested a quota increase yet, you can use an
+# `ml.t3.medium` instance type instead. This will work out of the box, but
+# the Processing Job will take significantly longer than it should have.
+# To get access to `ml.m5.xlarge` instances, you can request a quota
+# increase under the Service Quotas section in your AWS account.
+ instance_type=config["instance_type"],
+ instance_count=1,
+ role=role,
+ sagemaker_session=config["session"],
+)
+
+
Let’s now define the Processing Step that we’ll use in the pipeline.
+
This step will specify the list of inputs that we’ll access from the preprocessing script. In this case, the input is the dataset we stored in S3. We also have a few outputs that we want SageMaker to capture when the Processing Job finishes.
We can now create the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
Let’s create the training script. This script is responsible for training a neural network using the train data, validating the model, and saving it so we can later use it.
+
We will store the script in a folder called training and add it to the system path so we can later import it as a module.
import argparse
+import json
+import os
+import tarfile
+
+from pathlib import Path
+from comet_ml import Experiment
+
+import keras
+import numpy as np
+import pandas as pd
+from keras import Input
+from keras.layers import Dense
+from keras.models import Sequential
+from keras.optimizers import SGD
+from packaging import version
+from sklearn.metrics import accuracy_score
+
+
+def train(
+ model_directory,
+ train_path,
+ validation_path,
+ pipeline_path,
+ experiment,
+ epochs=50,
+ batch_size=32,
+):
+print(f"Keras version: {keras.__version__}")
+
+ X_train = pd.read_csv(Path(train_path) /"train.csv")
+ y_train = X_train[X_train.columns[-1]]
+ X_train = X_train.drop(X_train.columns[-1], axis=1)
+
+ X_validation = pd.read_csv(Path(validation_path) /"validation.csv")
+ y_validation = X_validation[X_validation.columns[-1]]
+ X_validation = X_validation.drop(X_validation.columns[-1], axis=1)
+
+ model = Sequential(
+ [
+ Input(shape=(X_train.shape[1],)),
+ Dense(10, activation="relu"),
+ Dense(8, activation="relu"),
+ Dense(3, activation="softmax"),
+ ]
+ )
+
+ model.compile(
+ optimizer=SGD(learning_rate=0.01),
+ loss="sparse_categorical_crossentropy",
+ metrics=["accuracy"],
+ )
+
+ model.fit(
+ X_train,
+ y_train,
+ validation_data=(X_validation, y_validation),
+ epochs=epochs,
+ batch_size=batch_size,
+ verbose=2,
+ )
+
+ predictions = np.argmax(model.predict(X_validation), axis=-1)
+ val_accuracy = accuracy_score(y_validation, predictions)
+print(f"Validation accuracy: {val_accuracy}")
+
+# Starting on version 3, Keras changed the model saving format.
+# Since we are running the training script using two different versions
+# of Keras, we need to check to see which version we are using and save
+# the model accordingly.
+ model_filepath = (
+ Path(model_directory) /"001"
+if version.parse(keras.__version__) < version.parse("3")
+else Path(model_directory) /"penguins.keras"
+ )
+
+ model.save(model_filepath)
+
+# Let's save the transformation pipelines inside the
+# model directory so they get bundled together.
+with tarfile.open(Path(pipeline_path) /"model.tar.gz", "r:gz") as tar:
+ tar.extractall(model_directory)
+
+if experiment:
+ experiment.log_parameters(
+ {
+"epochs": epochs,
+"batch_size": batch_size,
+ }
+ )
+ experiment.log_dataset_hash(X_train)
+ experiment.log_confusion_matrix(
+ y_validation.astype(int), predictions.astype(int)
+ )
+ experiment.log_model("penguins", model_filepath.as_posix())
+
+
+if__name__=="__main__":
+# Any hyperparameters provided by the training job are passed to
+# the entry point as script arguments.
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--epochs", type=int, default=50)
+ parser.add_argument("--batch_size", type=int, default=32)
+ args, _ = parser.parse_known_args()
+
+# Let's create a Comet experiment to log the metrics and parameters
+# of this training job.
+ comet_api_key = os.environ.get("COMET_API_KEY", None)
+ comet_project_name = os.environ.get("COMET_PROJECT_NAME", None)
+
+ experiment = (
+ Experiment(
+ project_name=comet_project_name,
+ api_key=comet_api_key,
+ auto_metric_logging=True,
+ auto_param_logging=True,
+ log_code=True,
+ )
+if comet_api_key and comet_project_name
+elseNone
+ )
+
+ training_env = json.loads(os.environ.get("SM_TRAINING_ENV", {}))
+ job_name = training_env.get("job_name", None) if training_env elseNone
+
+# We want to use the SageMaker's training job name as the name
+# of the experiment so we can easily recognize it.
+if job_name and experiment:
+ experiment.set_name(job_name)
+
+ train(
+# This is the location where we need to save our model.
+# SageMaker will create a model.tar.gz file with anything
+# inside this directory when the training script finishes.
+ model_directory=os.environ["SM_MODEL_DIR"],
+# SageMaker creates one channel for each one of the inputs
+# to the Training Step.
+ train_path=os.environ["SM_CHANNEL_TRAIN"],
+ validation_path=os.environ["SM_CHANNEL_VALIDATION"],
+ pipeline_path=os.environ["SM_CHANNEL_PIPELINE"],
+ experiment=experiment,
+ epochs=args.epochs,
+ batch_size=args.batch_size,
+ )
+
+
+
Let’s test the script to ensure everything is working as expected:
We can now create a Training Step that we can add to the pipeline. This Training Step will create a SageMaker Training Job in the background, run the training script, and upload the output to S3. Check the TrainingStep SageMaker’s SDK documentation for more information.
+
SageMaker manages the infrastructure of a Training Job. It provisions resources for the duration of the job, and cleans up when it completes. The Training Container image that SageMaker uses to run a Training Job can either be a SageMaker built-in image or a custom image.
Our training script uses Comet to track metrics from the Training Job. We need to create a requirements.txt file to install the Comet library in the training container.
+
+
+
+
requirements.txt
+
+
comet_ml
+
+
+
SageMaker uses the concept of an Estimator to handle end-to-end training and deployment tasks. For this example, we will use the built-in TensorFlow Estimator to run the training script we wrote before.
+
Notice the list of hyperparameters defined below. SageMaker will pass these hyperparameters as arguments to the entry point of the training script.
COMET_PROJECT_NAME: The name of the project where you want to track the experiments.
+
+
+
from sagemaker.tensorflow import TensorFlow
+
+estimator = TensorFlow(
+ base_job_name="training",
+ entry_point="script.py",
+ source_dir=f"{(CODE_FOLDER /'training').as_posix()}",
+# SageMaker will pass these hyperparameters as arguments
+# to the entry point of the training script.
+ hyperparameters={
+"epochs": 50,
+"batch_size": 32,
+ },
+# SageMaker will create these environment variables on the
+# Training Job instance.
+ environment={
+"COMET_API_KEY": COMET_API_KEY,
+"COMET_PROJECT_NAME": COMET_PROJECT_NAME,
+ },
+# SageMaker will track these metrics as part of the experiment
+# associated to this pipeline. The metric definitions tells
+# SageMaker how to parse the values from the Training Job logs.
+ metric_definitions=[
+ {"Name": "loss", "Regex": "loss: ([0-9\\.]+)"},
+ {"Name": "accuracy", "Regex": "accuracy: ([0-9\\.]+)"},
+ {"Name": "val_loss", "Regex": "val_loss: ([0-9\\.]+)"},
+ {"Name": "val_accuracy", "Regex": "val_accuracy: ([0-9\\.]+)"},
+ ],
+ image_uri=config["image"],
+ framework_version=config["framework_version"],
+ py_version=config["py_version"],
+ instance_type=config["instance_type"],
+ instance_count=1,
+ disable_profiler=True,
+ debugger_hook_config=False,
+ sagemaker_session=config["session"],
+ role=role,
+)
+
+
We can now create a Training Step. This Training Step will create a SageMaker Training Job in the background, run the training script, and upload the output to S3. Check the TrainingStep SageMaker’s SDK documentation for more information.
+
This step will receive the train and validation split from the preprocessing step as inputs.
+
Here, we are using three input channels, train, validation, and pipeline. SageMaker will automatically create an environment variable corresponding to each of these channels following the format SM_CHANNEL_[channel_name]:
+
+
SM_CHANNEL_TRAIN: This environment variable will contain the path to the training data coming from the preprocessing step.
+
SM_CHANNEL_VALIDATION: This environment variable will contain the path to the validation data comimng from the preprocessing step.
+
SM_CHANNEL_PIPELINE: This environment variable will contain the path to the transformation pipeline that we saved in the preprocessing step.
+
+
Notice that we are creating a function that we can later reuse to create a training step using a different estimator.
Let’s define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
The first step is to copy the training script to a folder where we’ll prepare the Docker image. We are going to reuse the training script we created before, since it’s compatible with the latest version of Keras.
Since we are creating a new Docker image, we need to install the libraries we need in the training container. We’ll use a requirements.txt file to install these libraries. Notice that we are installing jax to run it as our backend.
+
The sagemaker-training library contains the common functionality necessary to create a container compatible with SageMaker and its Python SDK.
We can now create the Dockerfile containing the instructions to build the training image. Notice how this image will automatically run the train.py script when it starts.
+
To use JAX as the backend for our model, we need to set the KERAS_BACKEND environment variable to jax.
+
+
+
+
Dockerfile
+
+
FROM python:3.10-slim
+
+RUN apt-get -y update && apt-get install -y --no-install-recommends \
+ python3 \
+ build-essential \
+ libssl-dev
+
+# Let's install the required Python packages from
+# the requirements.txt file.
+COPY requirements.txt .
+RUN pip install --user --upgrade pip
+RUN pip3 install -r requirements.txt
+
+# We are going to be running the training script
+# as the entrypoint of this container.
+COPY train.py /opt/ml/code/train.py
+ENV SAGEMAKER_PROGRAM train.py
+
+# We want to use JAX as the backend for Keras.
+ENV KERAS_BACKEND=jax
+
+
+
+
+
Step 2 - Building the Docker Image
+
We can now build the Docker image using the docker build command. We are going to define the name of this image using the IMAGE_NAME variable.
+
+
IMAGE_NAME ="keras-custom-training-container"
+
+ifnot LOCAL_MODE:
+# If we aren't running the code in Local Mode, we need
+# to specify we want to build the Docker image for the
+# linux/amd64 architecture before uploading it to ECR.
+print("Building Docker image for linux/amd64 architecture...")
+
+!docker build --platform="linux/amd64"-t $IMAGE_NAME \
+ $CODE_FOLDER/containers/training/
+else:
+# If we are running in Local Mode, we can use the
+# default Docker build command.
+print("Building Docker image for arm64 architecture...")
+
+!docker build -t $IMAGE_NAME \
+ $CODE_FOLDER/containers/training/
+
+
+
+
Step 3 - Pushing Docker Image to ECR
+
We can now push the Docker image to Amazon Elastic Container Registry (ECR). This is a fully-managed Docker container registry where we can manage Docker container images. This step is necessary to make the image available to SageMaker when running the pipeline.
+
+
algorithm_name=$2
+account=$(aws sts get-caller-identity --query Account --output text)
+
+# Get the region defined in the current configuration
+# (default to us-east-1 if none defined)
+region=$(aws configure get region)
+region=${region:-us-east-1}
+
+repository="${account}.dkr.ecr.${region}.amazonaws.com/${algorithm_name}:latest"
+
+# We only want to push the Docker image to ECR if
+# we are not running in Local Mode.
+if [ $1="False" ]
+then
+# Create the repository if it doesn't exist in ECR
+ aws ecr describe-repositories \
+--repository-names "${algorithm_name}">/dev/null 2>&1
+if [ $? -ne 0 ]
+ then
+ aws ecr create-repository \
+--repository-name "${algorithm_name}">/dev/null
+ fi
+
+# Get the login command from ECR to run the
+# Docker push command.
+ aws ecr get-login-password \
+--region ${region}|docker \
+ login --username AWS --password-stdin ${repository}
+
+# Push the Docker image to the ECR repository
+ docker tag ${algorithm_name} ${repository}
+ docker push ${repository}
+fi
+
+
+
+
Step 4 - Setting up the Training Step
+
Let’s now compute the name of the training image we’ll use to run the Training Job.
+
If we are running in LOCAL_MODE, we’ll use the name of the image we built before (IMAGE_NAME). Otherwise, we’ll use the name of the image we pushed to ECR.
Let’s define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
+
+
session5_pipeline = Pipeline(
+ name="session5-pipeline",
+ parameters=[dataset_location],
+ steps=[
+ preprocessing_step,
+# This time we want to use the new training step
+# we created using the custom Docker image.
+ keras_train_model_step,
+ ],
+ pipeline_definition_config=pipeline_definition_config,
+ sagemaker_session=config["session"],
+)
+
+session5_pipeline.upsert(role_arn=role)
Since we could use the Training of the Tuning Step to create a model, we’ll define a constant to indicate which approach we want to run. Notice that the Tuning Step is not supported in Local Mode.
+
+
USE_TUNING_STEP =False
+
+
+
+
Step 2 - Setting up a Tuning Step
+
Let’s now create a Tuning Step. This Tuning Step will create a SageMaker Hyperparameter Tuning Job in the background and use the training script to train different model variants and choose the best one. Check the TuningStep SageMaker’s SDK documentation for more information.
+
The Tuning Step requires a HyperparameterTuner reference to configure the Hyperparameter Tuning Job.
+
Here is the configuration that we’ll use to find the best model:
+
+
objective_metric_name: This is the name of the metric the tuner will use to determine the best model.
+
objective_type: This is the objective of the tuner. It specifies whether it should minimize the metric or maximize it. In this example, since we are using the validation accuracy of the model, we want the objective to be “Maximize.” If we were using the loss of the model, we would set the objective to “Minimize.”
+
metric_definitions: Defines how the tuner will determine the metric’s value by looking at the output logs of the training process.
+
+
The tuner expects the list of the hyperparameters you want to explore. You can use subclasses of the Parameter class to specify different types of hyperparameters. This example explores different values for the epochs hyperparameter.
+
Finally, you can control the number of jobs and how many of them will run in parallel using the following two arguments:
+
+
max_jobs: Defines the maximum total number of training jobs to start for the hyperparameter tuning job.
+
max_parallel_jobs: Defines the maximum number of parallel training jobs to start.
We can now create the Tuning Step using the tuner we configured before. SageMaker will create a Hyperparameter Tuning Job in the background and use the training script to train different model variants and choose the best one.
Let’s define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
This session extends the SageMaker Pipeline with a step to evaluate the model using the holdout set we created during the preprocessing step.
+
+
+
Step 1 - Creating the Evaluation Script
+
We’ll use a Processing Step to execute the evaluation script.
+
This script is responsible for loading the model we created and evaluating it on the test set. Before finishing, this script will generate an evaluation report of the model.
+
We will store the script in a folder called evaluation and add it to the system path so we can later import it as a module.
import json
+import tarfile
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+from sklearn.metrics import accuracy_score
+from tensorflow import keras
+
+
+def evaluate(model_path, test_path, output_path):
+ X_test = pd.read_csv(Path(test_path) /"test.csv")
+ y_test = X_test[X_test.columns[-1]]
+ X_test = X_test.drop(X_test.columns[-1], axis=1)
+
+# Let's now extract the model package so we can load
+# it in memory.
+with tarfile.open(Path(model_path) /"model.tar.gz") as tar:
+ tar.extractall(path=Path(model_path))
+
+ model = keras.models.load_model(Path(model_path) /"001")
+
+ predictions = np.argmax(model.predict(X_test), axis=-1)
+ accuracy = accuracy_score(y_test, predictions)
+print(f"Test accuracy: {accuracy}")
+
+# Let's create an evaluation report using the model accuracy.
+ evaluation_report = {
+"metrics": {
+"accuracy": {"value": accuracy},
+ },
+ }
+
+ Path(output_path).mkdir(parents=True, exist_ok=True)
+withopen(Path(output_path) /"evaluation.json", "w") as f:
+ f.write(json.dumps(evaluation_report))
+
+
+if__name__=="__main__":
+ evaluate(
+ model_path="/opt/ml/processing/model/",
+ test_path="/opt/ml/processing/test/",
+ output_path="/opt/ml/processing/evaluation/",
+ )
+
+
+
Let’s test the script to ensure everything is working as expected:
+
+
+Code
+
import os
+import shutil
+import tarfile
+import pytest
+import tempfile
+
+from processing.script import preprocess
+from training.script import train
+from evaluation.script import evaluate
+
+
+@pytest.fixture(scope="function", autouse=False)
+def directory():
+ directory = tempfile.mkdtemp()
+ input_directory = Path(directory) /"input"
+ input_directory.mkdir(parents=True, exist_ok=True)
+ shutil.copy2(DATA_FILEPATH, input_directory /"data.csv")
+
+ directory = Path(directory)
+
+ preprocess(base_directory=directory)
+
+ train(
+ model_directory=directory /"model",
+ train_path=directory /"train",
+ validation_path=directory /"validation",
+ pipeline_path=directory /"model",
+ experiment=None,
+ epochs=1,
+ )
+
+# After training a model, we need to prepare a package just like
+# SageMaker would. This package is what the evaluation script is
+# expecting as an input.
+with tarfile.open(directory /"model.tar.gz", "w:gz") as tar:
+ tar.add(directory /"model"/"001", arcname="001")
+
+ evaluate(
+ model_path=directory,
+ test_path=directory /"test",
+ output_path=directory /"evaluation",
+ )
+
+yield directory /"evaluation"
+
+ shutil.rmtree(directory)
+
+
+def test_evaluate_generates_evaluation_report(directory):
+ output = os.listdir(directory)
+assert"evaluation.json"in output
+
+
+def test_evaluation_report_contains_accuracy(directory):
+withopen(directory /"evaluation.json", "r") asfile:
+ report = json.load(file)
+
+assert"metrics"in report
+assert"accuracy"in report["metrics"]
+
+
+
+
+
Step 2 - Referencing the Model Assets
+
One of the inputs to the evaluation step is the model coming from the Training or the Tuning step. We can use the USE_TUNING_STEP flag to determine whether we created the model using a Training Step or a Tuning Step.
+
In case we are using the Tuning Step, we can use the TuningStep.get_top_model_s3_uri() function to get the model assets from the top performing training job of the Hyperparameter Tuning Job.
SageMaker supports mapping outputs from a Processing Step to property files. This is useful when we want to access a specific property from the pipeline.
We are now ready to define the Processing Step that will run the evaluation script:
+
+
evaluate_model_step = ProcessingStep(
+ name="evaluate-model",
+ step_args=evaluation_processor.run(
+ code=f"{(CODE_FOLDER /'evaluation'/'script.py').as_posix()}",
+ inputs=[
+# The first input is the test split that we generated on
+# the first step of the pipeline when we split and
+# transformed the data.
+ ProcessingInput(
+ source=preprocessing_step.properties.ProcessingOutputConfig.Outputs[
+"test"
+ ].S3Output.S3Uri,
+ destination="/opt/ml/processing/test",
+ ),
+# The second input is the model that we generated on
+# the Training or Tunning Step.
+ ProcessingInput(
+ source=model_assets,
+ destination="/opt/ml/processing/model",
+ ),
+ ],
+ outputs=[
+# The output is the evaluation report that we generated
+# in the evaluation script.
+ ProcessingOutput(
+ output_name="evaluation",
+ source="/opt/ml/processing/evaluation",
+ ),
+ ],
+ ),
+ property_files=[evaluation_report],
+ cache_config=cache_config,
+)
+
+
+
+
Step 5 - Creating the Pipeline
+
Let’s define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
First, let’s define the name of the group where we’ll register the model. The Model Registry uses groups to organize the versions of a model:
+
+
BASIC_MODEL_PACKAGE_GROUP ="basic-penguins"
+
+
+
+
Step 2 - Creating the Model
+
Let’s now create the model that we’ll register in the Model Registry. The model we trained uses TensorFlow, so we can use the built-in TensorFlowModel class to create an instance of the model:
When we register a model in the Model Registry, we can attach relevant metadata to it. We’ll use the evaluation report we generated during the evaluation step to populate the metrics of this model:
Let’s define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
This session extends the SageMaker Pipeline with a condition to register the model only if its accuracy is above a predefined threshold.
+
Here’s a high-level overview of the Condition Step:
+
+
+
Step 1 - Configuring the Accuracy Threshold
+
Let’s define a new Pipeline Parameter to specify the minimum accuracy that the model should reach for it to be registered.
+
+
from sagemaker.workflow.parameters import ParameterFloat
+
+accuracy_threshold = ParameterFloat(name="accuracy_threshold", default_value=0.70)
+
+
+
+
Step 2 - Setting up a Fail Step
+
If the model’s accuracy is not greater than or equal to our threshold, we will send the pipeline to a Fail Step with the appropriate error message. Check the FailStep SageMaker’s SDK documentation for more information.
+
+
from sagemaker.workflow.fail_step import FailStep
+
+fail_step = FailStep(
+ name="fail",
+ error_message=Join(
+ on=" ",
+ values=[
+"Execution failed because the model's accuracy was lower than",
+ accuracy_threshold,
+ ],
+ ),
+)
+
+
+
+
Step 3 - Defining the Condition
+
We can use a ConditionGreaterThanOrEqualTo condition to compare the model’s accuracy with the threshold. Look at the Conditions section in the documentation for more information about the types of supported conditions.
We can now define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
Let’s now write a simple Flask script to serve the model.
+
When this application receives the first request, it will unpack and load the model into memory. From there, it will use the model to make predictions on the incoming requests.
+
+
+
+
app.py
+
+
import tarfile
+import tempfile
+import numpy as np
+
+from flask import Flask, request, jsonify
+from pathlib import Path
+from tensorflow import keras
+
+
+MODEL_PATH = Path(__file__).parent
+
+
+class Model:
+ model =None
+
+def load(self):
+"""
+ Extracts the model package and loads the model in memory
+ if it hasn't been loaded yet.
+ """
+# We want to load the model only if it is not loaded yet.
+ifnot Model.model:
+# Before we load the model, we need to extract it in
+# a temporal directory.
+
+with tempfile.TemporaryDirectory() as directory:
+with tarfile.open(MODEL_PATH /"model.tar.gz") as tar:
+ tar.extractall(path=directory)
+
+ Model.model = keras.models.load_model(Path(directory) /"001")
+
+def predict(self, data):
+"""
+ Generates predictions for the supplied data.
+ """
+self.load()
+return Model.model.predict(data)
+
+
+app = Flask(__name__)
+model = Model()
+
+
+@app.route("/predict/", methods=["POST"])
+def predict():
+ data = request.data.decode("utf-8")
+
+ data = np.array(data.split(",")).astype(np.float32)
+ data = np.expand_dims(data, axis=0)
+
+ predictions = model.predict(data=[data])
+
+ prediction =int(np.argmax(predictions[0], axis=-1))
+ confidence =float(predictions[0][prediction])
+
+return jsonify({"prediction": prediction, "confidence": confidence})
+
+
+
+
+
Step 4 - Running the Flask Application
+
We can now run the Flask application to serve the model from a terminal using the following command:
+
$ flask --app program/code/serving/app.py --debug run --host=0.0.0.0 --port=4242
+
After the server is running, you can send a POST request to the server to get a prediction. Here is an example using the curl command:
This session deploys the model from the Model Registry to an endpoint. We’ll do it manually, using the SageMaker SDK. Check Deploy to a SageMaker Endpoint for more information about deploying a model to an endpoint.
+
+
+
Step 1 - Configuring the Endpoint Name
+
Let’s start by defining the name of the endpoint where we’ll deploy the model:
+
+
from sagemaker.predictor import Predictor
+
+ENDPOINT ="penguins-endpoint"
+
+
+
+
Step 2 - Creating a Model Package
+
To deploy a model using the SageMaker’s Python SDK, we need to create a Model Package using the ARN of the model from the Model Registry. Remember we got the ARN of the latest approved model in the previous section.
After deploying the model, we can test the endpoint to make sure it works.
+
Each line of the payload we’ll send to the endpoint contains the information of a penguin. Notice the model expects data that’s already transformed. We can’t provide the original data from our dataset because the model we registered will not work with it.
+
The endpoint will return the predictions for each of these lines.
An error occurred (ValidationError) when calling the InvokeEndpoint operation: Endpoint penguins-endpoint of account 325223348818 not found.
+
+
+
+
+
+
Session 12 - Deploying From the Pipeline
+
This session extends the SageMaker Pipeline with a step to automatically deploy the model to an endpoint.
+
We’ll use a Lambda Step to create an endpoint and deploy the model.
+
Here’s a high-level overview of the Deploy Step:
+
+
+
Step 1 - Configuring Data Capture Settings
+
We want to enable Data Capture as part of the endpoint configuration. With Data Capture we can record the inputs and outputs of the endpoint to use them later for monitoring the model. We need to configuration settings to enable Data Capture:
+
+
DATA_CAPTURE_PERCENTAGE: Represents the percentage of traffic that we want to capture.
+
DATA_CAPTURE_DESTINATION: Specifies the S3 location where we want to store the captured data.
import os
+import json
+import boto3
+import time
+
+sagemaker = boto3.client("sagemaker")
+
+
+def lambda_handler(event, context):
+# If we are calling this function from EventBridge,
+# we need to extract the model package ARN and the
+# approval status from the event details. If we are
+# calling this function from the pipeline, we can
+# assume the model is approved and we can get the
+# model package ARN as a direct parameter.
+if"detail"in event:
+ model_package_arn = event["detail"]["ModelPackageArn"]
+ approval_status = event["detail"]["ModelApprovalStatus"]
+else:
+ model_package_arn = event["model_package_arn"]
+ approval_status ="Approved"
+
+print(f"Model: {model_package_arn}")
+print(f"Approval status: {approval_status}")
+
+if approval_status !="Approved":
+ response = {
+"message": "Skipping deployment.",
+"approval_status": approval_status,
+ }
+
+print(response)
+return {"statusCode": 200, "body": json.dumps(response)}
+
+ endpoint_name = os.environ["ENDPOINT"]
+ data_capture_percentage =int(os.environ["DATA_CAPTURE_PERCENTAGE"])
+ data_capture_destination = os.environ["DATA_CAPTURE_DESTINATION"]
+ role = os.environ["ROLE"]
+
+ timestamp = time.strftime("%m%d%H%M%S", time.localtime())
+ model_name =f"{endpoint_name}-model-{timestamp}"
+ endpoint_config_name =f"{endpoint_name}-config-{timestamp}"
+
+ sagemaker.create_model(
+ ModelName=model_name,
+ ExecutionRoleArn=role,
+ Containers=[{"ModelPackageName": model_package_arn}],
+ )
+
+ sagemaker.create_endpoint_config(
+ EndpointConfigName=endpoint_config_name,
+ ProductionVariants=[
+ {
+"ModelName": model_name,
+"InstanceType": "ml.m5.xlarge",
+"InitialVariantWeight": 1,
+"InitialInstanceCount": 1,
+"VariantName": "AllTraffic",
+ }
+ ],
+# We can enable Data Capture to record the inputs and outputs
+# of the endpoint to use them later for monitoring the model.
+ DataCaptureConfig={
+"EnableCapture": True,
+"InitialSamplingPercentage": data_capture_percentage,
+"DestinationS3Uri": data_capture_destination,
+"CaptureOptions": [
+ {"CaptureMode": "Input"},
+ {"CaptureMode": "Output"},
+ ],
+"CaptureContentTypeHeader": {
+"CsvContentTypes": ["text/csv", "application/octect-stream"],
+"JsonContentTypes": ["application/json", "application/octect-stream"],
+ },
+ },
+ )
+
+ response = sagemaker.list_endpoints(NameContains=endpoint_name, MaxResults=1)
+
+iflen(response["Endpoints"]) ==0:
+# If the endpoint doesn't exist, let's create it.
+ sagemaker.create_endpoint(
+ EndpointName=endpoint_name,
+ EndpointConfigName=endpoint_config_name,
+ )
+else:
+# If the endpoint already exists, let's update it with the
+# new configuration.
+ sagemaker.update_endpoint(
+ EndpointName=endpoint_name,
+ EndpointConfigName=endpoint_config_name,
+ )
+
+return {"statusCode": 200, "body": json.dumps("Endpoint deployed successfully")}
+
+
+
+
+
Step 3 - Setting up Lambda Permissions
+
We need to ensure our Lambda function has permission to interact with SageMaker, so let’s create a new role with the appropriate permissions.
We can now define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
Waiter EndpointInService failed: Waiter encountered a terminal failure state: Matched expected service error code: ValidationException
+
+
+
+
+
+
Session 13 - Deploying From an Event
+
This session modifies the SageMaker Pipeline to register the model with PendingManualApproval status and deploys it whenever its status changes to Approved.
+
+
We will use Amazon EventBridge to trigger a Lambda function that will deploy the model whenever its status changes from “PendingManualApproval” to “Approved.”
+
+
Step 1 - Configuring the Model Package Group
+
We need to define the name of a new group where we’ll register models with PendingManualApproval status.
+
+
PENDING_MODEL_PACKAGE_GROUP ="pending-penguins"
+
+
+
+
Step 2 - Setting Up EventBridge
+
We can now create an EventBridge rule that triggers the deployment process whenever a model approval status becomes Approved. To do this, let’s define the event pattern that will trigger the deployment process. Check Model package state change for more information.
Now, we need to define the target of the rule. The target will trigger whenever the rule matches an event. In this case, we want to trigger the Lambda function we created before:
Let’s modify the Condition Step to include the new registration step. If the condition succeeds, we will register the model with PendingManualApproval status.
Let’s define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
This session creates an inference pipeline to control the data that goes in and comes out of the endpoint.
+
Deploying the model we trained directly to an endpoint doesn’t lets us control the data that goes in and comes out of the endpoint. The TensorFlow model we trained requires transformed data, which makes it useless to other applications:
+
+
To fix this, we can create an Inference Pipeline using SageMaker to control the data that goes in and comes out of the endpoint.
+
Our inference pipeline will have three components:
+
+
A preprocessing component that will transform the input data into the format the model expects.
+
The TensorFlow model.
+
A postprocessing component that will transform the output of the model into a human-readable format.
+
+
+
We want our endpoint to handle unprocessed data in CSV and JSON format and return the penguin’s species. Here is an example of the payload input we want the endpoint to support:
The first component of our inference pipeline will transform the input data into the format the model expects.
+
We’ll use the Scikit-Learn transformer we saved when we split and transformed the data. To deploy this component as part of an inference pipeline, we need to write a script that loads the transformer, uses it to modify the input data, and returns the output in the format the TensorFlow model expects.
+
We’ll store the scripts of every component in a folder called pipeline and add it to the system path so we can later import it as a module.
Let’s now create the script for the preprocessing component:
+
+
+
+
preprocessing_component.py
+
+
import os
+import pandas as pd
+import json
+import joblib
+
+from io import StringIO
+
+try:
+from sagemaker_containers.beta.framework import worker
+exceptImportError:
+# We don't have access to the `worker` package when testing locally.
+# We'll set it to None so we can change the way functions create
+# a response.
+ worker =None
+
+
+TARGET_COLUMN ="species"
+FEATURE_COLUMNS = [
+"island",
+"culmen_length_mm",
+"culmen_depth_mm",
+"flipper_length_mm",
+"body_mass_g",
+"sex",
+]
+
+
+def model_fn(model_dir):
+"""
+ Deserializes the model that will be used in this container.
+ """
+
+return joblib.load(os.path.join(model_dir, "features.joblib"))
+
+
+def input_fn(input_data, content_type):
+"""
+ Parses the input payload and creates a Pandas DataFrame.
+
+ This function will check whether the target column is present in the
+ input data and will remove it.
+ """
+
+if content_type =="text/csv":
+ df = pd.read_csv(StringIO(input_data), header=None, skipinitialspace=True)
+
+# If we find an extra column, it's probably the target
+# feature, so let's drop it. We'll assume the target
+# is always the first column,
+iflen(df.columns) ==len(FEATURE_COLUMNS) +1:
+ df = df.drop(df.columns[0], axis=1)
+
+ df.columns = FEATURE_COLUMNS
+return df
+
+if content_type =="application/json":
+ df = pd.DataFrame([json.loads(input_data)])
+
+if TARGET_COLUMN in df.columns:
+ df = df.drop(TARGET_COLUMN, axis=1)
+
+return df
+
+raiseValueError(f"{content_type} is not supported!")
+
+
+def predict_fn(input_data, model):
+"""
+ Preprocess the input using the transformer.
+ """
+
+try:
+return model.transform(input_data)
+exceptValueErroras e:
+print("Error transforming the input data", e)
+returnNone
+
+
+def output_fn(prediction, accept):
+"""
+ Formats the prediction output to generate a response.
+
+ The default accept/content-type between containers for serial inference
+ is JSON. Since this model preceeds a TensorFlow model, we want to
+ return a JSON object following TensorFlow's input requirements.
+ """
+
+if prediction isNone:
+raiseException("There was an error transforming the input data")
+
+ instances = [p for p in prediction.tolist()]
+ response = {"instances": instances}
+return (
+ worker.Response(json.dumps(response), mimetype=accept)
+if worker
+else (response, accept)
+ )
+
+
+
Let’s test the script to ensure everything is working as expected:
The final component of our inference pipeline will transform the output from the model into a human-readable format.
+
We’ll use the Scikit-Learn target transformer we saved when we split and transformed the data. To deploy this component as part of an inference pipeline, we need to write a script that loads the transformer, uses it to modify the output from the model, and returns a human-readable format.
+
+
+
+
postprocessing_component.py
+
+
import os
+import numpy as np
+import json
+import joblib
+
+
+try:
+from sagemaker_containers.beta.framework import encoders, worker
+exceptImportError:
+# We don't have access to the `worker` package when testing locally.
+# We'll set it to None so we can change the way functions create
+# a response.
+ worker =None
+
+
+def model_fn(model_dir):
+"""
+ Deserializes the target model and returns the list of fitted categories.
+ """
+
+ model = joblib.load(os.path.join(model_dir, "target.joblib"))
+return model.named_transformers_["species"].categories_[0]
+
+
+def input_fn(input_data, content_type):
+if content_type =="application/json":
+return json.loads(input_data)["predictions"]
+
+raiseValueError(f"{content_type} is not supported.")
+
+
+def predict_fn(input_data, model):
+"""
+ Transforms the prediction into its corresponding category.
+ """
+ predictions = np.argmax(input_data, axis=-1)
+ confidence = np.max(input_data, axis=-1)
+return [
+ (model[prediction], confidence)
+for confidence, prediction inzip(confidence, predictions)
+ ]
+
+def output_fn(prediction, accept):
+if accept =="text/csv":
+return (
+ worker.Response(encoders.encode(prediction, accept), mimetype=accept)
+if worker
+else (prediction, accept)
+ )
+
+if accept =="application/json":
+ response = []
+for p, c in prediction:
+ response.append({"prediction": p, "confidence": c})
+
+# If there's only one prediction, we'll return it
+# as a single object.
+iflen(response) ==1:
+ response = response[0]
+
+return (
+ worker.Response(json.dumps(response), mimetype=accept)
+if worker
+else (response, accept)
+ )
+
+raiseException(f"{accept} accept type is not supported.")
+
+
+
Let’s test the script to ensure everything is working as expected:
We can now create a PipelineModel to define our inference pipeline.
+
We’ll use the model we generated in the Split and Transform step as the input to the first and last components of the inference pipeline. This model.tar.gz file contains the two transformers we need to preprocess and postprocess the data.
+
Let’s create a variable with the URI to this file:
We can now define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
Let’s now test the endpoint. Notice that we can now send the raw data to the endpoint, and it will return the penguin’s species in a human-readable format.
An error occurred (ValidationError) when calling the InvokeEndpoint operation: Endpoint penguins-endpoint of account 325223348818 not found.
+
+
+
+
+
+
Session 15 - Custom Inference Script
+
This session creates a custom inference script to control the inference process in the SageMaker endpoint. This is an alternative to creating an inference pipeline to preprocess and postprocess the data that comes in and out of the model.
+
+
Step 1 - Creating the Inference Script
+
Let’s create a script where we’ll manage the inference process in the endpoint.
+
We’ll’ include this code as part of the model assets to control the inference process on the SageMaker endpoint. SageMaker will automatically call the handler() function for every request to the endpoint. Check How to implement the pre- and/or post-processing handler(s) for more information.
+
We can now create the script inside the folder.
+
+
+
+
inference.py
+
+
import os
+import json
+import requests
+import joblib
+import numpy as np
+import pandas as pd
+from pathlib import Path
+
+
+def handler(data, context, directory=Path("/opt/ml/model")):
+"""
+ This is the entrypoint that will be called by SageMaker
+ when the endpoint receives a request.
+ """
+print("Handling endpoint request")
+
+ processed_input = _process_input(data, context, directory)
+ output = _predict(processed_input, context, directory) if processed_input elseNone
+return _process_output(output, context, directory)
+
+
+def _process_input(data, context, directory):
+print("Processing input data...")
+
+if context isNone:
+# The context will be None when we are testing the code
+# directly from a notebook. In that case, we can use the
+# data directly.
+ endpoint_input = data
+elif context.request_content_type in (
+"application/json",
+"application/octet-stream",
+ ):
+# When the endpoint is running, we will receive a context
+# object. We need to parse the input and turn it into
+# JSON in that case.
+ endpoint_input = data.read().decode("utf-8")
+else:
+raiseValueError(
+f"Unsupported content type: {context.request_content_type or'unknown'}"
+ )
+
+# Let's now transform the input data using the features pipeline.
+try:
+ endpoint_input = json.loads(endpoint_input)
+ df = pd.json_normalize(endpoint_input)
+ features_pipeline = joblib.load(directory /"features.joblib")
+ result = features_pipeline.transform(df)
+exceptExceptionas e:
+print(f"There was an error processing the input data. {e}")
+returnNone
+
+return result[0].tolist()
+
+
+def _predict(instance, context, directory):
+print("Sending input data to model to make a prediction...")
+
+if context isNone:
+# The context will be None when we are testing the code
+# directly from a notebook. In that case, we want to load the
+# model we trained and make a prediction using it.
+import keras
+
+ model = keras.models.load_model(Path(directory) /"001")
+ predictions = model.predict(np.array([instance]))
+ result = {"predictions": predictions.tolist()}
+else:
+# When the endpoint is running, we will receive a context
+# object. In that case we need to send the instance to the
+# model to get a prediction back.
+ model_input = json.dumps({"instances": [instance]})
+ response = requests.post(context.rest_uri, data=model_input)
+
+if response.status_code !=200:
+raiseValueError(response.content.decode("utf-8"))
+
+ result = json.loads(response.content)
+
+print(f"Response: {result}")
+return result
+
+
+def _process_output(output, context, directory):
+print("Processing prediction received from the model...")
+
+if output:
+ prediction = np.argmax(output["predictions"][0])
+ confidence = output["predictions"][0][prediction]
+
+ target_pipeline = joblib.load(directory /"target.joblib")
+ classes = target_pipeline.named_transformers_["species"].categories_[0]
+
+ result = {
+"prediction": classes[prediction],
+"confidence": confidence,
+ }
+else:
+ result = {"prediction": None}
+
+print(result)
+
+ response_content_type = (
+"application/json"if context isNoneelse context.accept_header
+ )
+return json.dumps(result), response_content_type
+
+
+
Let’s test the script to ensure everything is working as expected:
We can now create a new TensorFlowModel including the inference.py file.
+
SageMaker triggers a repack operation whenever we specify the source_dir attribute in a model. We want that attribute to point to the local folder containing the inference.py file. SageMaker will automatically modify the original model.tar.gz package to include a /code folder containing the file.
+
Since we need access to Scikit-Learn in our script, we can include a requirements.txt file in the same location where the inference.py script is, and SageMaker will install everything in it.
+
To repack the model assets, SageMaker will automatically include a new step in the pipeline right before registering the model.
+
Here is what the new model.tar.gz package will look like:
We can now define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
This step will compute statistics and constraints from the data. We’ll’ later use this information as the baseline to detect data drift and other data quality issues.
+
+
Check Monitor data quality for more information about monitoring data quality in SageMaker.
+
+
Step 1 - Configuring Baseline Location
+
Let’s start by defining the location where SageMaker will store the baseline data:
Let’s configure a QualityCheck Step to compute the general statistics of the data we used to build our model.
+
We can configure the instance that will run the quality check using the CheckJobConfig class, and we can use the DataQualityCheckConfig class to configure the job.
+
We are running this step with the following configuration:
+
+
skip_check = True: This parameter controls whether the step should skip checking the data against a previous baseline. Since we want to generate the baseline for the first time, we set it to True. After running the pipeline once to generate the baseline, we can set this parameter to False to ensure any new data follows the same distribution as the baseline.
+
register_new_baseline = True: This parameter controls whether the new calculated baseline will be registered in the Model Registry.
We can now define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
Our pipeline generated data baseline statistics and constraints. We can take a look at what these values look like by downloading them from S3. You need to wait for the pipeline to finish running before these files are available.
This step will compute the baseline metrics we will later use as the baseline to detect model drift.
+
To create a baseline to compare the model performance, we must create predictions for the test set and compare the model’s metrics with the model performance on production data. We can do this by running a Batch Transform Job to predict every sample from the test set. We can use a Transform Step as part of the pipeline to run this job.
+
+
Check Monitor model quality for more information about monitoring model quality in SageMaker.
+
+
Step 1 - Configuring Baseline Location
+
Let’s start by defining the location where SageMaker will store the baseline data:
We are going to use a Batch Transform Job to generate predictions for every sample from the test set.
+
This Batch Transform Job will run every sample from the training dataset through the model so we can compute the baseline metrics. Check Run a Batch Transform Job for more information about running a Batch Transform Job.
+
Let’s start by configuring a Transformer instance:
We’ll generate predictions for the baseline test data that we generated when we split and transformed the data. This baseline is the same data we used to test the model, but it’s in raw format.
+
The output of this Batch Transform Job will have two fields. The first one will be the ground truth label, and the second one will be the prediction of the model.
+
+
+
from sagemaker.workflow.steps import TransformStep
+
+generate_test_predictions_step = TransformStep(
+ name="generate-test-predictions",
+ step_args=transformer.transform(
+# We will use the baseline set we generated when we split the data.
+# This set corresponds to the test split before the transformation step.
+ data=preprocessing_step.properties.ProcessingOutputConfig.Outputs[
+"test-baseline"
+ ].S3Output.S3Uri,
+ join_source="Input",
+ split_type="Line",
+ content_type="text/csv",
+# We want to output the first and the second to last field from
+# the joint set. The first field corresponds to the groundtruth,
+# and the second to last field corresponds to the prediction.
+#
+# Here is an example of the data the Transform Job will generate
+# after joining the input with the output from the model:
+#
+# Gentoo,39.1,18.7,181.0,3750.0,MALE,Gentoo,0.52
+#
+# Notice how the first field is the groundtruth coming from the
+# test set. The second to last field is the prediction coming the
+# model.
+ output_filter="$[0,-2]",
+ ),
+ cache_config=cache_config,
+)
+
+
+
+
Step 4 - Generating Model Quality Baseline
+
Let’s now configure the Quality Check Step and feed it the data we generated in the Transform Step. This step will automatically compute the performance metrics of the model on the test set.
+
We are running this step with the following configuration:
+
+
skip_check = True: This parameter controls whether the step should skip checking the data against a previous baseline. Since we want to generate the baseline for the first time, we set it to True. After running the pipeline once to generate the baseline, we can set this parameter to False to ensure any new data follows the same distribution as the baseline.
+
register_new_baseline = True: This parameter controls whether the new calculated baseline will be registered in the Model Registry.
+
+
+
from sagemaker.workflow.quality_check_step import ModelQualityCheckConfig
+
+model_quality_baseline_step = QualityCheckStep(
+ name="generate-model-quality-baseline",
+ check_job_config=CheckJobConfig(
+ instance_type="ml.c5.xlarge",
+ instance_count=1,
+ volume_size_in_gb=20,
+ sagemaker_session=config["session"],
+ role=role,
+ ),
+ quality_check_config=ModelQualityCheckConfig(
+# We are going to use the output of the Transform Step to generate
+# the model quality baseline.
+ baseline_dataset=generate_test_predictions_step.properties.TransformOutput.S3OutputPath,
+ dataset_format=DatasetFormat.csv(header=False),
+# We need to specify the problem type and the fields where the prediction
+# and groundtruth are so the process knows how to interpret the results.
+ problem_type="MulticlassClassification",
+# Since the data doesn't have headers, SageMaker will autocreate headers for it.
+# _c0 corresponds to the first column, and _c1 corresponds to the second column.
+ ground_truth_attribute="_c0",
+ inference_attribute="_c1",
+ output_s3_uri=MODEL_QUALITY_LOCATION,
+ ),
+ model_package_group_name=PIPELINE_MODEL_PACKAGE_GROUP,
+ skip_check=True,
+ register_new_baseline=True,
+ cache_config=cache_config,
+)
We can now define the SageMaker Pipeline and submit its definition to the SageMaker Pipelines service to create the pipeline if it doesn’t exist or update it if it does.
Our pipeline generated model baseline constraints. We can take a look at what these values look like by downloading them from S3. You need to wait for the pipeline to finish running before the file is available.
This session creates a Monitoring Job to monitor the quality of the data received by the endpoint. This schedule will run periodically and check the data that goes into the endpoint against the baseline we generated before.
Let’s deploy the latest approved model to an endpoint.
+
Since we need to do the same later, we can create a function to deploy the model. Notice how we need to enable Data Capture to monitor the data that goes in and out of the endpoint.
+
+
from sagemaker.model_monitor import DataCaptureConfig
+
+
+def deploy_model():
+"""Deploy the latest model registered in the Model Registry."""
+ response = sagemaker_client.list_model_packages(
+ ModelPackageGroupName=PIPELINE_MODEL_PACKAGE_GROUP,
+ ModelApprovalStatus="Approved",
+ SortBy="CreationTime",
+ MaxResults=1,
+ )
+
+ package = (
+ response["ModelPackageSummaryList"][0]
+if response["ModelPackageSummaryList"]
+elseNone
+ )
+
+if package:
+ model_package = ModelPackage(
+ model_package_arn=package["ModelPackageArn"],
+ sagemaker_session=sagemaker_session,
+ role=role,
+ )
+
+ model_package.deploy(
+ endpoint_name=ENDPOINT,
+ initial_instance_count=1,
+ instance_type=config["instance_type"],
+# We must enable Data Capture to monitor the model.
+ data_capture_config=DataCaptureConfig(
+ enable_capture=True,
+ sampling_percentage=100,
+ destination_s3_uri=DATA_CAPTURE_DESTINATION,
+ capture_options=["REQUEST", "RESPONSE"],
+ csv_content_types=["text/csv"],
+ json_content_types=["application/json"],
+ ),
+ )
+
+
+
deploy_model()
+
+
+
+
Step 2 - Generating Fake Traffic
+
To test the monitoring functionality, we need to generate traffic to the endpoint. To generate traffic, we will send every sample from the dataset to the endpoint to simulate real prediction requests:
+
+
from sagemaker.serializers import JSONSerializer
+
+
+def generate_fake_traffic():
+"""Generate fake traffic to the endpoint."""
+try:
+for index, row in data.iterrows():
+ payload =",".join([str(x) for x in row.to_list()])
+ predictor.predict(
+ payload,
+ initial_args={"ContentType": "text/csv", "Accept": "text/csv"},
+# The `inference_id` field is important to match
+# it later with a corresponding ground-truth label.
+ inference_id=str(index),
+ )
+exceptExceptionas e:
+print(e)
+
+
+generate_fake_traffic()
+
+
We can check the location where the endpoint stores the captured data, download a file, and display its content. It may take a few minutes for the first few files to show up in S3.
+
These files contain the data captured by the endpoint in a SageMaker-specific JSON-line format. Each inference request is captured in a single line in the jsonl file. The line contains both the input and output merged together:
SageMaker looks for violations in the data captured by the endpoint. By default, it combines the input data with the endpoint output and compares the result with the baseline we generated before. If we let SageMaker do this, we will get a few violations, for example an “extra column check” violation because the field confidence doesn’t exist in the baseline data.
+
We can fix these violations by creating a preprocessing script configuring the data we want the monitoring job to use. Check Preprocessing and Postprocessing for more information about how to configure these scripts.
+
We’ll store the script in a folder called monitoring:
We can now define the preprocessing script. Notice that this script will return a JSON object with a name for each feature and their value.
+
+
+
+
data_quality_preprocessor.py
+
+
import json
+
+
+def preprocess_handler(inference_record, logger):
+ input_data = inference_record.endpoint_input.data
+return {str(i).zfill(2): d for i, d inenumerate(input_data.split(","))}
+
+
+
+
+
Step 4 - Uploading Preprocessing Script
+
The monitoring schedule expects an S3 location pointing to the preprocessing script. Let’s upload the script to the default bucket.
INFO:sagemaker.image_uris:Defaulting to the only supported framework/algorithm version: .
+INFO:sagemaker.image_uris:Ignoring unnecessary instance type: None.
+
+
+
Let’s now create the monitoring schedule. Notice how we specify the record_preprocessor_script using the S3 location where we uploaded our script.
+
We are going to set up the monitoring schedule to run every hour. Keep in mind that SageMaker has a buffer period of 20 minutes to schedule an execution.
+
+
import time
+from sagemaker.model_monitor import CronExpressionGenerator
+
+data_monitor.create_monitoring_schedule(
+ monitor_schedule_name="penguins-data-monitoring-schedule",
+ endpoint_input=ENDPOINT,
+ record_preprocessor_script=data_quality_preprocessor,
+ statistics=f"{DATA_QUALITY_LOCATION}/statistics.json",
+ constraints=f"{DATA_QUALITY_LOCATION}/constraints.json",
+ schedule_cron_expression=CronExpressionGenerator.hourly(),
+ output_s3_uri=DATA_QUALITY_LOCATION,
+ enable_cloudwatch_metrics=True,
+)
+
+# Let's give SageMaker some time to process the
+# monitoring job before we start it.
+time.sleep(10)
+data_monitor.start_monitoring_schedule()
+
+
+
+
Step 6 - Checking Violations
+
After the monitoring schedule runs for the first time, we can check the results of the last execution. If the job completed successfully, we can check if there are any violations.
+
+
def check_execution(monitoring_schedule):
+"""Check the execution of the Monitoring Job.
+
+ This function checks the execution of the Monitoring
+ Job and prints out the list of violations if the job
+ completed.
+ """
+try:
+ executions = monitoring_schedule.list_executions()
+
+if executions:
+ execution = executions[-1].describe()
+print(f"Processing Job Status: {execution['ProcessingJobStatus']}")
+
+if execution["ProcessingJobStatus"] =="Completed":
+print(f"Exit Message: \"{execution['ExitMessage']}\"")
+print(
+f"Last Modified Time: {execution['LastModifiedTime']}",
+ end="\n\n",
+ )
+print("Execution:")
+print(json.dumps(execution, default=str, indent=2), end="\n\n")
+
+ latest_monitoring_violations = (
+ monitoring_schedule.latest_monitoring_constraint_violations()
+ )
+ response = json.loads(
+ S3Downloader.read_file(latest_monitoring_violations.file_s3_uri),
+ )
+print("Violations:")
+print(json.dumps(response, indent=2))
+exceptExceptionas e:
+print(e)
+
+
+check_execution(data_monitor)
This session creates a Monitoring Job to monitor the quality of the model outputs. This schedule will run periodically and check the data that goes into the endpoint against the baseline we generated before.
Let’s deploy the latest approved model to an endpoint.
+
Here, we can reuse the function we created before to deploy the model.
+
+
deploy_model()
+
+
+
+
Step 3 - Generating Fake Traffic
+
To test the monitoring functionality, we need to generate traffic to the endpoint. We can use the function we created before to generate fake traffic to the endpoint.
+
+
generate_fake_traffic()
+
+
We can check the location where the endpoint stores the captured data, download a file, and display its content. It may take a few minutes for the first few files to show up in S3.
+
These files contain the data captured by the endpoint in a SageMaker-specific JSON-line format. Each inference request is captured in a single line in the jsonl file. The line contains both the input and output merged together:
To test the performance of the model, we need to label the samples captured by the endpoint. We can simulate the labeling process by generating a random label for every sample. Check Ingest Ground Truth Labels and Merge Them With Predictions for more information about this.
+
+
import random
+from datetime import datetime, timezone
+
+from sagemaker.s3 import S3Uploader
+
+records = []
+for inference_id inrange(len(data)):
+ random.seed(inference_id)
+
+ records.append(
+ json.dumps(
+ {
+"groundTruthData": {
+# For testing purposes, we will generate a random
+# label for each request.
+"data": random.choice(["Adelie", "Chinstrap", "Gentoo"]),
+"encoding": "CSV",
+ },
+"eventMetadata": {
+# This value should match the id of the request
+# captured by the endpoint.
+"eventId": str(inference_id),
+ },
+"eventVersion": "0",
+ },
+ ),
+ )
+
+groundtruth_payload ="\n".join(records)
+upload_time = datetime.now(tz=timezone.utc)
+uri =f"{GROUND_TRUTH_LOCATION}/{upload_time:%Y/%m/%d/%H/%M%S}.jsonl"
+S3Uploader.upload_string_as_file_body(groundtruth_payload, uri)
+
+
+
+
Step 5 - Creating Monitoring Schedule
+
To set up a Model Quality Monitoring Job, we can use the ModelQualityMonitor class.
Let’s now create the monitoring schedule. The EndpointInput instance configures the attribute the monitoring job should use to determine the prediction from the model.
+
We are going to set up the monitoring schedule to run every hour. Keep in mind that SageMaker has a buffer period of 20 minutes to schedule an execution.
+
+
import time
+
+from sagemaker.model_monitor import CronExpressionGenerator, EndpointInput
+
+model_monitor.create_monitoring_schedule(
+ monitor_schedule_name="penguins-model-monitoring-schedule",
+ endpoint_input=EndpointInput(
+ endpoint_name=ENDPOINT,
+# The first attribute is the prediction made
+# by the model. For example, here is a
+# potential output from the model:
+# [Adelie,0.977324724\n]
+ inference_attribute="0",
+ destination="/opt/ml/processing/input_data",
+ ),
+ problem_type="MulticlassClassification",
+ ground_truth_input=GROUND_TRUTH_LOCATION,
+ constraints=f"{MODEL_QUALITY_LOCATION}/constraints.json",
+ schedule_cron_expression=CronExpressionGenerator.hourly(),
+ output_s3_uri=MODEL_QUALITY_LOCATION,
+ enable_cloudwatch_metrics=True,
+)
+
+# Let's give SageMaker some time to process the
+# monitoring job before we start it.
+time.sleep(10)
+model_monitor.start_monitoring_schedule()
+
+
+
+
Step 6 - Checking Violations
+
After the monitoring schedule runs for the first time, we can check the results of the last execution. If the job completed successfully, we can check if there are any violations.
We want to deploy the two latest approved models from the Model Registry to the same endpoint. The latest version of the model will act as the Shadow variant, and the previous version will act as the Production variant.
Production package: arn:aws:sagemaker:us-east-1:325223348818:model-package/basic-penguins/5
+Shadow package: arn:aws:sagemaker:us-east-1:325223348818:model-package/basic-penguins/6
+
+
+
+
+
Step 2 - Creating the Models
+
We want to deploy the two packages to a new endpoint. We’ll use the boto3 API to deploy these models.
+
Let’s start by creating the SageMaker Models.
+
+
import time
+
+# We'll use a different name for this endpoint.
+SHADOW_DEPLOYMENT_ENDPOINT ="shadow-penguins-endpoint"
+
+# The timestamp will help us create unique name for the
+# name of the models.
+timestamp = time.strftime("%m%d%H%M%S", time.localtime())
Let’s check the location where the endpoint stores the captured data, download a file, and display its content. It may take a few minutes for the first few files to show up in S3.
+
The endpoint will capture the data for both the Production and the Shadow variants.
