Creating a Pipeline¶
In the previous labs, you used the Azure Machine Learning SDK v2 to explore the entire model training process from accessing data through to running training experiments and registering machine learning models. Up until now, you have performed the various steps required to create a machine learning solution interactively. In this lab, you'll explore automation of these steps using pipelines.
Connect to Your Workspace¶
The first thing you need to do is to connect to your workspace using the Azure ML SDK v2.
Note: If the authenticated session with your Azure subscription has expired since you completed the previous exercise, you'll be prompted to reauthenticate.
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential
credential = DefaultAzureCredential()
ml_client = MLClient.from_config(credential=credential)
print(f"Ready to work with {ml_client.workspace_name}")
Prepare Data for a Pipeline¶
In this lab, you'll use the diabetes data. The pipeline's training step reads it as a single CSV file (uri_file), so rather than depending on the exact type/version of the shared diabetes_mltable data asset from earlier labs (which is registered as an mltable), you'll just pass the local data/diabetes.csv file directly as the pipeline input - the SDK uploads it automatically when the pipeline job is submitted.
Create Scripts for Pipeline Steps¶
Pipelines in the Azure ML SDK v2 consist of one or more components. Each component wraps a script (or other executable) together with its inputs, outputs, and environment, and can run in its own compute context. A pipeline is built by connecting components together - the output of one component becomes the input of the next.
In this exercise, you'll build a simple pipeline that contains a component to train a model, and a component to register the trained model.
import os
# Create a folder for the pipeline step files
experiment_folder = 'diabetes_pipeline'
os.makedirs(experiment_folder, exist_ok=True)
print(experiment_folder)
Now you can create the script for the first step, which will train a model. The script includes parameters for training_data (the input data asset) and output_folder (the folder where the trained model should be saved). Because Azure ML jobs configure MLflow tracking automatically, the script can use mlflow to log metrics and save the model.
%%writefile $experiment_folder/train_diabetes.py
# Import libraries
import argparse
import pandas as pd
import numpy as np
import mlflow
import mlflow.sklearn
from sklearn.model_selection import train_test_split
from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import roc_auc_score
from sklearn.metrics import roc_curve
import matplotlib.pyplot as plt
# Get parameters
parser = argparse.ArgumentParser()
parser.add_argument('--training_data', type=str, dest='training_data', help='training data')
parser.add_argument('--output_folder', type=str, dest='output_folder', default="diabetes_model", help='output folder')
args = parser.parse_args()
output_folder = args.output_folder
# load the diabetes data (passed as a component input)
print("Loading Data...")
diabetes = pd.read_csv(args.training_data)
# Separate features and labels
X, y = diabetes[['Pregnancies','PlasmaGlucose','DiastolicBloodPressure','TricepsThickness','SerumInsulin','BMI','DiabetesPedigree','Age']].values, diabetes['Diabetic'].values
# Split data into training set and test set
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.30, random_state=0)
# Train a decision tree model
print('Training a decision tree model')
model = DecisionTreeClassifier().fit(X_train, y_train)
# calculate accuracy
y_hat = model.predict(X_test)
acc = np.average(y_hat == y_test)
print('Accuracy:', acc)
mlflow.log_metric('Accuracy', float(acc))
# calculate AUC
y_scores = model.predict_proba(X_test)
auc = roc_auc_score(y_test,y_scores[:,1])
print('AUC: ' + str(auc))
mlflow.log_metric('AUC', float(auc))
# plot ROC curve
fpr, tpr, thresholds = roc_curve(y_test, y_scores[:,1])
fig = plt.figure(figsize=(6, 4))
# Plot the diagonal 50% line
plt.plot([0, 1], [0, 1], 'k--')
# Plot the FPR and TPR achieved by our model
plt.plot(fpr, tpr)
plt.xlabel('False Positive Rate')
plt.ylabel('True Positive Rate')
plt.title('ROC Curve')
mlflow.log_figure(fig, "ROC.png")
plt.show()
# Save the trained model in MLflow format, ready to be picked up by the next step
print("Saving model to", output_folder)
# cloudpickle format: the default skops refuses to serialise a decision tree.
mlflow.sklearn.save_model(model, path=output_folder,
serialization_format="cloudpickle")
The script for the second step registers the model saved by the first step in the workspace. It includes a single model_folder parameter that contains the path where the model was saved. The whole MLflow-format folder is registered as it stands - the model is not loaded into memory or saved a second time along the way.
%%writefile $experiment_folder/register_diabetes.py
# Import libraries
import argparse
import os
import mlflow
# Get parameters
parser = argparse.ArgumentParser()
parser.add_argument('--model_folder', type=str, dest='model_folder', default="diabetes_model", help='model location')
args = parser.parse_args()
model_folder = args.model_folder
# Register the model saved by the training step. The folder is registered as it
# stands, without loading the model into memory and saving it a second time.
print("Registering model from " + model_folder)
mlflow.register_model(f"file://{os.path.abspath(model_folder)}", "diabetes_model")
print("Model registered as diabetes_model")
Prepare a Compute Target for the Pipeline Steps¶
In this exercise, you'll use the same compute cluster for both steps, but it's important to realize that each step runs independently; so you could specify different compute targets for each step if appropriate.
First, get (or create) the compute cluster.
from azure.ai.ml.entities import AmlCompute
cluster_name = "aml-cluster"
try:
pipeline_cluster = ml_client.compute.get(cluster_name)
print('Found existing cluster, use it.')
except Exception:
# If not, create it
pipeline_cluster = AmlCompute(
name=cluster_name,
type="amlcompute",
size="Standard_D2as_v4",
min_instances=0,
max_instances=2,
idle_time_before_scale_down=300,
)
ml_client.compute.begin_create_or_update(pipeline_cluster).result()
print(f"Compute target '{pipeline_cluster.name}' ready to use.")
The compute will require a Python environment with the necessary package dependencies installed, so we'll create one and register it.
from azure.ai.ml.entities import Environment
# Define the conda dependencies for the pipeline steps.
# mlflow is pinned: a newer one breaks artifact logging through azureml-mlflow.
conda_dependencies = {
"name": "diabetes-pipeline-env",
"channels": ["conda-forge"],
"dependencies": [
"python=3.10",
"scikit-learn",
"pandas",
"matplotlib",
"pip",
{"pip": ["mlflow<=3.15.0", "azureml-mlflow"]},
],
}
pipeline_env = Environment(
name="diabetes-pipeline-env",
description="Training environment for the diabetes pipeline",
conda_file=conda_dependencies,
image="mcr.microsoft.com/azureml/openmpi5.0-ubuntu24.04",
)
pipeline_env = ml_client.environments.create_or_update(pipeline_env)
print(f"Environment '{pipeline_env.name}' registered, version {pipeline_env.version}.")
Create and Run a Pipeline¶
Now we're ready to create and run a pipeline.
First we need to define the components for the pipeline. In this case, the first component must write the trained model to a folder that can be read from by the second component. Since each component runs in its own container (and could even run on different compute), the outputs and inputs of a component are declared explicitly, and Azure ML takes care of passing the data between the steps' storage locations. We'll define an Output for the training component's model folder, and use it as the Input to the registration component.
from azure.ai.ml import command, Input, Output
from azure.ai.ml.constants import AssetTypes
# Step 1: train the model
train_step = command(
name="train_diabetes_model",
display_name="Train Model",
description="Trains a diabetes classification model",
inputs={"training_data": Input(type=AssetTypes.URI_FILE)},
outputs={"model_output": Output(type=AssetTypes.MLFLOW_MODEL)},
code=experiment_folder,
command="python train_diabetes.py --training_data ${{inputs.training_data}} --output_folder ${{outputs.model_output}}",
environment=f"{pipeline_env.name}:{pipeline_env.version}",
)
# Step 2: register the trained model
register_step = command(
name="register_diabetes_model",
display_name="Register Model",
description="Registers the trained model in the workspace",
inputs={"model_folder": Input(type=AssetTypes.MLFLOW_MODEL)},
code=experiment_folder,
command="python register_diabetes.py --model_folder ${{inputs.model_folder}}",
environment=f"{pipeline_env.name}:{pipeline_env.version}",
)
print("Pipeline steps defined")
OK, we're ready to go. Let's build the pipeline from the components we've defined, using the @dsl.pipeline decorator, and run it as a job.
from azure.ai.ml import dsl
# Construct the pipeline
@dsl.pipeline(description="Trains and registers a diabetes classification model")
def diabetes_training_pipeline(pipeline_input_data: Input(type=AssetTypes.URI_FILE)):
train_job = train_step(training_data=pipeline_input_data)
register_step(model_folder=train_job.outputs.model_output)
return {"trained_model": train_job.outputs.model_output}
pipeline_job = diabetes_training_pipeline(
pipeline_input_data=Input(type=AssetTypes.URI_FILE, path="data/diabetes.csv")
)
# Set the default compute for all steps that don't specify their own
pipeline_job.settings.default_compute = "aml-cluster"
print("Pipeline is built.")
# Submit the pipeline as a job
pipeline_job = ml_client.jobs.create_or_update(
pipeline_job, experiment_name="diabetes-training-pipeline"
)
print("Pipeline submitted for execution.")
# Stream the job logs until the pipeline completes
ml_client.jobs.stream(pipeline_job.name)
The stream call above shows the pipeline's progress until it completes. You can also monitor pipeline jobs on the Jobs page in Azure Machine Learning studio.
When the pipeline has finished, a new version of the diabetes_model should be registered. Run the following code to verify this.
for model in ml_client.models.list(name="diabetes_model"):
print(f"{model.name} version: {model.version}")
for tag_name, tag in (model.tags or {}).items():
print(f"\t{tag_name} : {tag}")
This is a simple example, designed to demonstrate the principle. In reality, you could build more sophisticated logic into the pipeline steps - for example, evaluating the model against some test data to calculate a performance metric like AUC or accuracy, comparing the metric to that of any previously registered versions of the model, and only registering the new model if it performs better.
You can use the Azure Machine Learning extension for Azure DevOps to combine Azure ML pipelines with Azure DevOps pipelines (yes, it is confusing that they have the same name!) and integrate model retraining into a continuous integration/continuous deployment (CI/CD) process. For example you could use an Azure DevOps build pipeline to trigger an Azure ML pipeline job that trains and registers a model, and when the model is registered it could trigger an Azure DevOps release pipeline that deploys the model as an endpoint, along with the application or service that consumes the model.