Instrukcja do tego ćwiczenia Pobierz notatnik (.ipynb)

Creating a Batch Inferencing Service¶

In previous labs, you deployed a model to a managed online endpoint for real-time inferencing (getting predictions from a model). Now you'll create a batch endpoint for batch inferencing. What does that mean? Well, imagine a health clinic takes patient measurements all day, saving the details for each patient in a separate file. Then overnight, the diabetes prediction model can be used to process all of the day's patient data as a batch, generating predictions that will be waiting the following morning so that the clinic can follow up with patients who are predicted to be at risk of diabetes. That's what you'll implement in this exercise.

Connect to Your Workspace¶

The first thing you need to do is to connect to your workspace using the Azure ML SDK.

Note: If the authenticated session with your Azure subscription has expired since you completed the previous exercise, you'll be prompted to reauthenticate.

In [ ]:

from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
ml_client = MLClient.from_config(credential=credential)
print(f"Ready to use Azure ML to work with {ml_client.workspace_name}")

Train and Register a Model¶

The batch scoring script used later in this lab expects the registered diabetes_model to be a plain scikit-learn model saved with joblib (a CUSTOM_MODEL asset containing a diabetes_model.pkl file) - not the MLflow-format model registered in some of the earlier labs (Lab 3B, Lab 6A). Run the cell below to train and register a model in that expected format, so the deployment steps that follow work regardless of what other labs you've already completed.

In [ ]:

import pandas as pd
import numpy as np
import joblib
from sklearn.model_selection import train_test_split
from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import roc_auc_score
from azure.ai.ml.entities import Model
from azure.ai.ml.constants import AssetTypes

# load the diabetes dataset
print("Loading Data...")
diabetes = pd.read_csv('data/diabetes.csv')

# 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)

# calculate AUC
y_scores = model.predict_proba(X_test)
auc = roc_auc_score(y_test, y_scores[:,1])
print('AUC: ' + str(auc))

# Save the trained model
model_file = 'diabetes_model.pkl'
joblib.dump(value=model, filename=model_file)

# Register the model
registered_model = Model(
    path=model_file,
    type=AssetTypes.CUSTOM_MODEL,
    name="diabetes_model",
    description="A decision tree model that classifies patients by their likelihood of being diabetic.",
    tags={"Training context": "Inline Training"},
    properties={"AUC": str(auc), "Accuracy": str(acc)},
)
ml_client.models.create_or_update(registered_model)

print('Model trained and registered.')

Generate and Upload Batch Data¶

Since we don't actually have a fully staffed clinic with patients from whom to get new data for this course, you'll generate a random sample from our second diabetes CSV file and use those to test the batch endpoint. Then you'll register that data as a uri_folder data asset in the Azure Machine Learning workspace.

In [ ]:

import pandas as pd
import os
from azure.ai.ml.entities import Data
from azure.ai.ml.constants import AssetTypes

# Load the diabetes data (new patient data to be scored)
diabetes = pd.read_csv('data/diabetes2.csv')
# Get a 100-item sample of the feature columns (not the diabetic label)
sample = diabetes[['Pregnancies','PlasmaGlucose','DiastolicBloodPressure','TricepsThickness','SerumInsulin','BMI','DiabetesPedigree','Age']].sample(n=100).values

# Create a local folder
batch_folder = './batch-data'
os.makedirs(batch_folder, exist_ok=True)
print("Folder created!")

# Save each sample as a separate file
print("Saving files...")
for i in range(100):
    fname = str(i+1) + '.csv'
    sample[i].tofile(os.path.join(batch_folder, fname), sep=",")
print("files saved!")

# Register the folder of files as a uri_folder data asset
print("Uploading files and registering data asset...")
batch_data_set = Data(
    path=batch_folder,
    type=AssetTypes.URI_FOLDER,
    description="A folder of new patient observations to be scored in batch",
    name="diabetes_batch_data",
)
ml_client.data.create_or_update(batch_data_set)

print("Done!")

Create Compute¶

We'll need a compute target for the batch deployment, so we'll use the Azure ML compute cluster you used in the previous exercises (it will be created if it doesn't already exist).

In [ ]:

from azure.ai.ml.entities import AmlCompute

cluster_name = "aml-cluster"

# Check for an existing cluster
try:
    inference_cluster = ml_client.compute.get(cluster_name)
    print('Found existing cluster, use it.')
except Exception:
    # Create an AzureML compute cluster
    compute_config = AmlCompute(
        name=cluster_name,
        type="amlcompute",
        size="Standard_D2as_v4",
        min_instances=0,
        max_instances=3,
        idle_time_before_scale_down=300,
    )
    inference_cluster = ml_client.compute.begin_create_or_update(compute_config).result()

print(inference_cluster.name, "is available.")

Create a Batch Scoring Script¶

Now we're ready to define the batch deployment. Our deployment will need Python code to perform the batch scoring, so let's create a folder where we can keep the files used by the deployment:

In [ ]:

import os
# Create a folder for the deployment files
experiment_folder = 'batch_deploy'
os.makedirs(experiment_folder, exist_ok=True)

print(experiment_folder)

Now we'll create a Python script to do the actual work, and save it in the deployment folder:

In [ ]:

%%writefile $experiment_folder/batch_diabetes.py
import os
import numpy as np
import joblib


def init():
    # Runs once when the deployment is initialized
    global model

    # AZUREML_MODEL_DIR is set by the batch deployment and points to the
    # folder containing the registered model files
    model_path = os.path.join(os.getenv("AZUREML_MODEL_DIR"), "diabetes_model.pkl")
    model = joblib.load(model_path)


def run(mini_batch):
    # This runs for each mini-batch of files
    resultList = []

    # process each file in the batch
    for f in mini_batch:
        # Read the comma-delimited data into an array
        data = np.genfromtxt(f, delimiter=',')
        # Reshape into a 2-dimensional array for prediction (model expects multiple items)
        prediction = model.predict(data.reshape(1, -1))
        # Append prediction to results
        resultList.append("{}: {}".format(os.path.basename(f), prediction[0]))
    return resultList

Next we'll define the environment that includes the dependencies required by the script

In [ ]:

%%writefile $experiment_folder/batch_env.yml
name: batch-environment
dependencies:
  - python=3.8
  - numpy
  - pip
  - pip:
      - scikit-learn
      - joblib

In [ ]:

from azure.ai.ml.entities import Environment

batch_env = Environment(
    name="batch-environment",
    conda_file=f"{experiment_folder}/batch_env.yml",
    image="mcr.microsoft.com/azureml/openmpi5.0-ubuntu24.04",
)
print('Environment configuration ready.')

You're going to deploy the batch prediction script as a batch deployment behind a batch endpoint. The batch deployment handles distributing the input files across compute nodes in mini-batches, running the scoring script against each mini-batch in parallel, and collating the results into a single output file.

So you'll need to import the classes used to define batch endpoints and deployments.

OK, now you're ready to create the batch endpoint and deployment. We'll create an endpoint named diabetes-batch-endpoint with a deployment named diabetes-batch-dpl.

In [ ]:

from azure.ai.ml.entities import BatchEndpoint, ModelBatchDeployment, ModelBatchDeploymentSettings, CodeConfiguration
from azure.ai.ml.constants import BatchDeploymentOutputAction

endpoint_name = "diabetes-batch-endpoint"

# Create the batch endpoint
endpoint = BatchEndpoint(
    name=endpoint_name,
    description="A batch endpoint for scoring diabetes patient data",
)
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()

# Get the registered model to deploy
model = ml_client.models.get(name="diabetes_model", label="latest")

# Create the batch deployment
batch_deployment = ModelBatchDeployment(
    name="diabetes-batch-dpl",
    endpoint_name=endpoint_name,
    model=model,
    environment=batch_env,
    code_configuration=CodeConfiguration(code=experiment_folder, scoring_script="batch_diabetes.py"),
    compute=cluster_name,
    settings=ModelBatchDeploymentSettings(
        instance_count=2,
        max_concurrency_per_instance=2,
        mini_batch_size=5,
        output_action=BatchDeploymentOutputAction.APPEND_ROW,
        output_file_name="predictions.csv",
    ),
)
ml_client.batch_deployments.begin_create_or_update(batch_deployment).result()

# Make this the default deployment for the endpoint
endpoint = ml_client.batch_endpoints.get(endpoint_name)
endpoint.defaults.deployment_name = batch_deployment.name
ml_client.batch_endpoints.begin_create_or_update(endpoint).result()

print('Batch endpoint and deployment created.')

Now it's time to invoke the batch endpoint, passing the batch data asset as input, and wait for the scoring job to finish.

Note: This may take some time!

In [ ]:

from azure.ai.ml import Input
from azure.ai.ml.constants import AssetTypes

# Get the latest version of the registered batch data asset
batch_data_asset = ml_client.data.get(name="diabetes_batch_data", label="latest")

job = ml_client.batch_endpoints.invoke(
    endpoint_name=endpoint_name,
    inputs={
        "input_data": Input(
            type=AssetTypes.URI_FOLDER,
            path=f"azureml:diabetes_batch_data:{batch_data_asset.version}",
        )
    },
)

ml_client.jobs.stream(job.name)

When the job has finished running, the resulting predictions will have been saved by the child scoring job. You can retrieve them as follows:

In [ ]:

import glob
import shutil
import pandas as pd

shutil.rmtree('diabetes-results', ignore_errors=True)

# The deployment runs the scoring script in a child job - get a reference to it
scoring_job = list(ml_client.jobs.list(parent_job_name=job.name))[0]

# Download its output (named "score" by default for a model batch deployment)
ml_client.jobs.download(name=scoring_job.name, download_path='diabetes-results', output_name='score')

# Find the predictions file
result_file = glob.glob('diabetes-results/**/predictions.csv', recursive=True)[0]

# cleanup output format
df = pd.read_csv(result_file, delimiter=":", header=None)
df.columns = ["File", "Prediction"]

# Display the first 20 results
df.head(20)

Use the Batch Endpoint from an Application¶

A batch endpoint is a durable, callable resource as soon as you create it. Client applications can trigger a scoring job either by using the Azure ML SDK's invoke method (as you did above), the Azure CLI (az ml batch-endpoint invoke), or by calling the endpoint's REST API directly.

Batch endpoints authenticate REST calls using Microsoft Entra ID (Azure AD) tokens rather than a simple key, so a real application would typically authenticate as a service principal and present a bearer token obtained from Azure AD. For details and examples, see the Authorization on batch endpoints documentation.

Now you have a batch endpoint that can be used to batch process daily patient data.

More Information: For more details about using batch endpoints for batch inferencing, see the Batch endpoints documentation.

Clean Up¶

Batch deployments only consume compute resources while a scoring job is running, so there's no ongoing cost from leaving the endpoint deployed. If you want to remove it anyway, run the following cell to delete it:

In [ ]:

# ml_client.batch_endpoints.begin_delete(name=endpoint_name)