Introduction

Cloud deployment makes a trained Machine Learning model available through infrastructure hosted and managed in a cloud environment. Applications, scheduled jobs, business workflows, and authorized services can submit new data and receive model predictions.

A cloud deployment is not only an uploaded model file. A complete deployment includes the approved model artifact, preprocessing, runtime environment, scoring logic, compute resources, endpoint, authentication, network controls, logs, metrics, scaling rules, release strategy, and rollback procedure.

This lesson uses Azure Machine Learning managed online endpoints as the primary implementation example. The architectural principles also apply to managed Machine Learning services, container platforms, serverless services, and Kubernetes-based deployments on other cloud platforms.

Learning Objectives

Prerequisites

python -m pip install azure-ai-ml azure-identity pandas numpy scikit-learn joblib

What Is Cloud Model Deployment?

Cloud model deployment places an approved model and its inference dependencies in a cloud-hosted environment capable of receiving input and producing predictions.

Online vs Batch Endpoints

Cloud Deployment Options

Cloud Deployment Lifecycle

Recommended Project Structure

cloud-ml-deployment/
├── README.md
├── requirements.txt
├── .gitignore
├── artifacts/
│   ├── customer_churn_pipeline.joblib
│   └── model_metadata.json
├── cloud/
│   ├── endpoint.yml
│   ├── deployment-blue.yml
│   ├── deployment-green.yml
│   └── environment.yml
├── src/
│   ├── register_model.py
│   ├── deploy_endpoint.py
│   ├── test_endpoint.py
│   └── score.py
├── sample-data/
│   └── request.json
└── tests/
    ├── test_scoring.py
    ├── test_contract.py
    └── test_deployment.py

Prepare the Model Artifact

Connect to Azure Machine Learning

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


subscription_id = (
    "replace-with-subscription-id"
)


resource_group_name = (
    "replace-with-resource-group"
)


workspace_name = (
    "replace-with-workspace-name"
)


credential = DefaultAzureCredential()


ml_client = MLClient(
    credential=credential,
    subscription_id=subscription_id,
    resource_group_name=(
        resource_group_name
    ),
    workspace_name=workspace_name
)


workspace = ml_client.workspaces.get(
    workspace_name
)


print(
    workspace.name
)

Register the Model

from azure.ai.ml.entities import Model
from azure.ai.ml.constants import AssetTypes


model_asset = Model(
    path=(
        "artifacts/"
        "customer_churn_pipeline.joblib"
    ),
    type=AssetTypes.CUSTOM_MODEL,
    name="customer-churn-pipeline",
    description=(
        "Approved customer-churn "
        "classification pipeline"
    ),
    tags={
        "project": "customer-churn",
        "framework": "scikit-learn",
        "intended_use": (
            "retention-review-prioritization"
        )
    }
)


registered_model = (
    ml_client.models.create_or_update(
        model_asset
    )
)


print(
    registered_model.name
)


print(
    registered_model.version
)

Define the Inference Environment

name: customer-churn-inference

channels:
  - conda-forge

dependencies:
  - python=3.12
  - pip
  - pip:
      - azureml-inference-server-http
      - pandas
      - numpy
      - scikit-learn
      - joblib

Create the Scoring Script

Azure Machine Learning scoring code commonly provides an initialization function and a request-processing function. The model should load once during initialization rather than during each request.

import json
import os
from pathlib import Path

import joblib
import pandas as pd


model = None


required_features = [
    "tenure_months",
    "monthly_charges",
    "total_charges",
    "contract_type",
    "payment_method",
    "tech_support"
]


decision_threshold = 0.60


def init():
    global model

    model_directory = Path(
        os.getenv(
            "AZUREML_MODEL_DIR"
        )
    )

    model_files = list(
        model_directory.rglob(
            "*.joblib"
        )
    )

    if len(model_files) != 1:

        raise RuntimeError(
            "Exactly one Joblib model "
            "artifact is required."
        )

    model = joblib.load(
        model_files[0]
    )


def run(
    raw_data
):
    if model is None:

        raise RuntimeError(
            "The model is not initialized."
        )

    payload = json.loads(
        raw_data
    )

    records = payload.get(
        "records"
    )

    if not isinstance(
        records,
        list
    ) or not records:

        raise ValueError(
            "records must be a non-empty array."
        )

    validated_records = [
        validate_record(
            record
        )
        for record in records
    ]

    input_frame = pd.DataFrame(
        validated_records
    )[
        required_features
    ]

    probabilities = model.predict_proba(
        input_frame
    )[:, 1]

    predictions = []

    for probability in probabilities:

        probability = float(
            probability
        )

        predictions.append(
            {
                "predicted_class": (
                    "churned"
                    if probability
                    >=
                    decision_threshold
                    else "retained"
                ),
                "positive_class_probability": (
                    probability
                ),
                "retention_review_recommended": (
                    probability
                    >=
                    decision_threshold
                )
            }
        )

    return {
        "predictions": predictions,
        "prediction_count": len(
            predictions
        ),
        "decision_threshold": (
            decision_threshold
        )
    }

Validate Inference Records

import math


allowed_categories = {
    "contract_type": {
        "Monthly",
        "One year",
        "Two year"
    },
    "payment_method": {
        "Electronic",
        "Bank transfer",
        "Credit card",
        "Mailed check"
    },
    "tech_support": {
        "Yes",
        "No"
    }
}


def validate_record(
    record
):
    if not isinstance(
        record,
        dict
    ):

        raise ValueError(
            "Each record must be an object."
        )

    missing_features = (
        set(required_features)
        - set(record)
    )

    unexpected_features = (
        set(record)
        - set(required_features)
    )

    if missing_features:

        raise ValueError(
            "Missing features: "
            f"{sorted(missing_features)}"
        )

    if unexpected_features:

        raise ValueError(
            "Unexpected features: "
            f"{sorted(unexpected_features)}"
        )

    validated_record = {}

    for feature_name in [
        "tenure_months",
        "monthly_charges",
        "total_charges"
    ]:
        value = record[
            feature_name
        ]

        if isinstance(
            value,
            bool
        ) or not isinstance(
            value,
            (
                int,
                float
            )
        ):
            raise ValueError(
                f"{feature_name} must be numeric."
            )

        value = float(
            value
        )

        if not math.isfinite(
            value
        ):

            raise ValueError(
                f"{feature_name} must be finite."
            )

        if value < 0:

            raise ValueError(
                f"{feature_name} cannot be negative."
            )

        validated_record[
            feature_name
        ] = value

    for feature_name, allowed_values in (
        allowed_categories.items()
    ):
        value = record[
            feature_name
        ]

        if value not in allowed_values:

            raise ValueError(
                f"{feature_name} contains "
                "an unsupported value."
            )

        validated_record[
            feature_name
        ] = value

    return validated_record

Create the Managed Online Endpoint

import uuid

from azure.ai.ml.entities import (
    ManagedOnlineEndpoint
)


endpoint_name = (
    "churn-endpoint-"
    + uuid.uuid4().hex[:8]
)


endpoint = ManagedOnlineEndpoint(
    name=endpoint_name,
    description=(
        "Managed endpoint for approved "
        "customer-churn inference"
    ),
    auth_mode="key",
    tags={
        "project": "customer-churn",
        "environment": "development"
    }
)


endpoint_operation = (
    ml_client.online_endpoints
    .begin_create_or_update(
        endpoint
    )
)


created_endpoint = (
    endpoint_operation.result()
)


print(
    created_endpoint.name
)

Create the First Deployment

from azure.ai.ml.entities import (
    CodeConfiguration
)
from azure.ai.ml.entities import Environment
from azure.ai.ml.entities import (
    ManagedOnlineDeployment
)


inference_environment = Environment(
    name=(
        "customer-churn-inference"
    ),
    description=(
        "Environment for customer-churn "
        "online inference"
    ),
    conda_file=(
        "cloud/environment.yml"
    ),
    image=(
        "mcr.microsoft.com/"
        "azureml/openmpi4.1.0-ubuntu22.04"
    )
)


blue_deployment = (
    ManagedOnlineDeployment(
        name="blue",
        endpoint_name=endpoint_name,
        model=registered_model,
        environment=(
            inference_environment
        ),
        code_configuration=(
            CodeConfiguration(
                code="src",
                scoring_script="score.py"
            )
        ),
        instance_type=(
            "Standard_DS3_v2"
        ),
        instance_count=1
    )
)


deployment_operation = (
    ml_client.online_deployments
    .begin_create_or_update(
        blue_deployment
    )
)


created_deployment = (
    deployment_operation.result()
)


print(
    created_deployment.name
)

Configure Deployment Probes

from azure.ai.ml.entities import (
    ProbeSettings
)


blue_deployment.liveness_probe = (
    ProbeSettings(
        initial_delay=10,
        period=10,
        timeout=2,
        failure_threshold=30
    )
)


blue_deployment.readiness_probe = (
    ProbeSettings(
        initial_delay=10,
        period=10,
        timeout=2,
        failure_threshold=30
    )
)

Allocate Endpoint Traffic

endpoint.traffic = {
    "blue": 100
}


traffic_operation = (
    ml_client.online_endpoints
    .begin_create_or_update(
        endpoint
    )
)


traffic_operation.result()

Create a Sample Request

{
    "records": [
        {
            "tenure_months": 18,
            "monthly_charges": 64.5,
            "total_charges": 1161.0,
            "contract_type": "Monthly",
            "payment_method": "Electronic",
            "tech_support": "No"
        }
    ]
}

Invoke the Deployment

response = (
    ml_client.online_endpoints.invoke(
        endpoint_name=endpoint_name,
        deployment_name="blue",
        request_file=(
            "sample-data/request.json"
        )
    )
)


print(
    response
)

Review Deployment Logs

deployment_logs = (
    ml_client.online_deployments
    .get_logs(
        name="blue",
        endpoint_name=endpoint_name,
        lines=100
    )
)


print(
    deployment_logs
)

Invoke the Endpoint Through HTTPS

import requests


endpoint_details = (
    ml_client.online_endpoints.get(
        endpoint_name
    )
)


endpoint_credentials = (
    ml_client.online_endpoints
    .get_keys(
        endpoint_name
    )
)


with open(
    "sample-data/request.json",
    "r",
    encoding="utf-8"
) as request_file:

    request_payload = request_file.read()


response = requests.post(
    endpoint_details.scoring_uri,
    headers={
        "Authorization": (
            "Bearer "
            + endpoint_credentials.primary_key
        ),
        "Content-Type": "application/json"
    },
    data=request_payload,
    timeout=30
)


response.raise_for_status()


print(
    response.json()
)

Cloud Identity and Authorization

Network Security

Create a Candidate Deployment

green_deployment = (
    ManagedOnlineDeployment(
        name="green",
        endpoint_name=endpoint_name,
        model=(
            candidate_registered_model
        ),
        environment=(
            inference_environment
        ),
        code_configuration=(
            CodeConfiguration(
                code="src",
                scoring_script="score.py"
            )
        ),
        instance_type=(
            "Standard_DS3_v2"
        ),
        instance_count=1
    )
)


green_operation = (
    ml_client.online_deployments
    .begin_create_or_update(
        green_deployment
    )
)


green_operation.result()

Canary Traffic Allocation

endpoint.traffic = {
    "blue": 90,
    "green": 10
}


canary_operation = (
    ml_client.online_endpoints
    .begin_create_or_update(
        endpoint
    )
)


canary_operation.result()

Cloud Rollout Strategies

Roll Back Traffic

endpoint.traffic = {
    "blue": 100,
    "green": 0
}


rollback_operation = (
    ml_client.online_endpoints
    .begin_create_or_update(
        endpoint
    )
)


rollback_operation.result()

Scaling the Deployment

Cloud Deployment Testing

Test Prediction Consistency

import json
import joblib
import numpy as np
import pandas as pd


local_model = joblib.load(
    "artifacts/customer_churn_pipeline.joblib"
)


with open(
    "sample-data/request.json",
    "r",
    encoding="utf-8"
) as request_file:

    request_payload = json.load(
        request_file
    )


records = request_payload[
    "records"
]


local_probabilities = (
    local_model.predict_proba(
        pd.DataFrame(
            records
        )[
            required_features
        ]
    )[:, 1]
)


cloud_response = json.loads(
    response
)


cloud_probabilities = np.array(
    [
        item[
            "positive_class_probability"
        ]
        for item in cloud_response[
            "predictions"
        ]
    ]
)


np.testing.assert_allclose(
    local_probabilities,
    cloud_probabilities,
    rtol=1e-8,
    atol=1e-10
)


print(
    "Cloud predictions match "
    "the approved local pipeline."
)

Cloud Monitoring

Alerts and Operational Ownership

Cloud Cost Management

Delete Learning Resources

Remove temporary learning endpoints and deployments when they are no longer required. Cloud resources can continue consuming quota or generating charges until they are deleted.

delete_operation = (
    ml_client.online_endpoints
    .begin_delete(
        name=endpoint_name
    )
)


delete_operation.result()


print(
    "Endpoint deletion completed."
)

CI/CD Deployment Pipeline

Responsible Cloud Deployment

Common Cloud Deployment Mistakes

Recommended Cloud Deployment Workflow

Hands-On Practice

Knowledge Check

Cloud Deployment Readiness Checklist

Lesson Summary