Publishing a Pipeline¶
In the previous lab, you created a pipeline. Now you're going to deploy it behind a batch endpoint, so it can be run on-demand or on a schedule without needing to open a notebook.
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}")
Find the Pipeline Job to Deploy¶
To make a pipeline reusable outside a notebook, you deploy a pipeline job behind a batch endpoint: Azure ML turns the job into a reusable pipeline component automatically, and the endpoint gives you a stable REST URI you (or another application) can call to start new runs.
Before you can deploy a pipeline, it must have been run at least once. You ran the pipeline in the previous lab, so now you just need a reference to that job.
# Get the most recent job for the pipeline experiment
experiment_name = "diabetes-training-pipeline"
pipeline_jobs = [job for job in ml_client.jobs.list() if job.experiment_name == experiment_name]
pipeline_jobs.sort(key=lambda job: job.creation_context.created_at, reverse=True)
pipeline_job_run = ml_client.jobs.get(pipeline_jobs[0].name)
print(f"Using pipeline job: {pipeline_job_run.name} (status: {pipeline_job_run.status})")
Create a Batch Endpoint¶
A batch endpoint provides a stable REST URI that stays the same even if you later change or redeploy the pipeline behind it. You'll see the endpoint on the Endpoints page (on the Batch endpoints tab) in Azure Machine Learning studio once it's created.
from azure.ai.ml.entities import BatchEndpoint
endpoint_name = "diabetes-pipeline-endpoint"
endpoint = BatchEndpoint(
name=endpoint_name,
description="Batch endpoint for the diabetes training pipeline",
)
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()
print(f"Endpoint '{endpoint_name}' created.")
You can find the endpoint's invocation URI as a property of the endpoint object:
endpoint = ml_client.batch_endpoints.get(name=endpoint_name)
print(endpoint.scoring_uri)
Deploy the Pipeline to the Endpoint¶
Now you can create a deployment under the endpoint from the existing pipeline job. Azure ML uses the job you found earlier as the definition of the pipeline component to run - you don't need to redefine the components or the @dsl.pipeline function again.
from azure.ai.ml.entities import PipelineComponentBatchDeployment
deployment_name = "diabetes-pipeline-deployment"
deployment = PipelineComponentBatchDeployment(
name=deployment_name,
description="Deployment of the diabetes training pipeline",
endpoint_name=endpoint_name,
job_definition=pipeline_job_run,
settings={"continue_on_step_failure": False, "default_compute": "aml-cluster"},
)
ml_client.batch_deployments.begin_create_or_update(deployment).result()
# Make this the default deployment for the endpoint
endpoint = ml_client.batch_endpoints.get(name=endpoint_name)
endpoint.defaults.deployment_name = deployment_name
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()
print(f"Deployment '{deployment_name}' is now the default deployment for '{endpoint_name}'.")
Use the Batch Endpoint¶
To use the endpoint, client applications call invoke (or the equivalent REST call). Because the MLClient you created is already authenticated with your Azure credentials, there's no need to acquire and pass an authorization header manually.
The pipeline runs asynchronously, so you'll get a job back immediately, which you can use to track the pipeline run as it executes.
from azure.ai.ml import Input
from azure.ai.ml.constants import AssetTypes
# Pass the local CSV file directly as the pipeline input - the SDK uploads it
# automatically. (This pipeline's training step reads a single uri_file, so we
# don't use the shared diabetes_mltable data asset, which is registered as an
# mltable.)
job = ml_client.batch_endpoints.invoke(
endpoint_name=endpoint_name,
inputs={"pipeline_input_data": Input(type=AssetTypes.URI_FILE, path="data/diabetes.csv")},
)
print(job.name)
Since we have the job, we can stream its logs to watch the pipeline run.
Note: The pipeline should complete quickly if the underlying steps are configured to allow output reuse. In reality, you'd likely want the training step to run every time in case the data has changed.
ml_client.jobs.stream(name=job.name)
Schedule the Pipeline¶
Suppose the clinic for the diabetes patients collects new data each week, and adds it to the dataset. You could run the pipeline every week to retrain the model with the new data, using a schedule to trigger a new run of the pipeline job automatically.
from azure.ai.ml.entities import JobSchedule, RecurrenceTrigger, RecurrencePattern
from azure.ai.ml.constants import TimeZone
schedule_name = "weekly-diabetes-training"
# Trigger every Monday at 00:00 UTC
recurrence_trigger = RecurrenceTrigger(
frequency="week",
interval=1,
schedule=RecurrencePattern(hours=0, minutes=0, week_days=["Monday"]),
time_zone=TimeZone.UTC,
)
job_schedule = JobSchedule(
name=schedule_name,
trigger=recurrence_trigger,
create_job=pipeline_job_run.name,
)
job_schedule = ml_client.schedules.begin_create_or_update(schedule=job_schedule).result()
print(f"Schedule '{job_schedule.name}' created.")
You can retrieve the schedules that are defined in the workspace like this:
for schedule in ml_client.schedules.list():
print(schedule.name)
The schedule may have triggered a run already, so let's check the most recent job for the pipeline experiment.
recent_jobs = [job for job in ml_client.jobs.list() if job.experiment_name == experiment_name]
recent_jobs.sort(key=lambda job: job.creation_context.created_at, reverse=True)
latest_job = recent_jobs[0]
print(f"Name: {latest_job.name}, Status: {latest_job.status}, Display name: {latest_job.display_name}")
More Information: You can find out more about scheduling pipeline jobs in the documentation.