Deploying Models with Azure Machine Learning

Level: Intermediate
Objective: Master ML model deployment to production on Azure

Table of Contents

1. Introduction to ML Deployment
2. Model Catalog and Fine-tuning
3. Online Endpoints – Real-time Inference
4. Scoring Script – The Core of Deployment
5. Blue/Green Deployment and Traffic Splitting
6. Batch Endpoints – Batch Inference
7. Deployment on AKS (Kubernetes)
8. Model Monitoring and Surveillance
9. CI/CD for ML Deployment
10. Complete Implementation with the SDK
11. Patterns and Best Practices
12. Summary and Key Points
13. Glossary

1. Introduction to ML Deployment

1.1 Why Is Deployment Difficult?

Training an ML model is only half the work. Production deployment is often the most complex and risky step in the ML lifecycle.

flowchart LR
    subgraph "What Data Scientists do"
        TRAIN["Train\na model"] --> EVAL["Evaluate\nmetrics"] --> SAVE["Save\nthe model"]
    end
    
    subgraph "What production requires"
        API["Stable\nREST API"] --> SCALE["Auto\nscaling"] --> HA["High\nAvailability"]
        HA --> MONITOR["24/7\nMonitoring"] --> VERSION["Version\nmanagement"]
    end
    
    subgraph "The gap (Valley of Death)"
        GAP["❓ How to move\nfrom notebook\nto production?"]
    end
    
    SAVE --> GAP --> API

Azure ML solves these problems with:

• Managed Online Endpoints: Scalable, HA REST API in a few lines of code
• Batch Endpoints: Deferred bulk data processing
• MLflow Model Registry: Model versioning and traceability
• Monitor: Performance and drift surveillance

1.2 Deployment Types

flowchart TD
    DEPLOY["Azure ML Deployment"] --> RT["Real-Time\n(Online Endpoint)\n\n• Immediate response\n• 1 request at a time\n• Always-on REST API\n• Ex: real-time fraud detection"]
    DEPLOY --> BATCH["Batch\n(Batch Endpoint)\n\n• Deferred processing\n• Millions of rows\n• Cost-effective\n• Ex: monthly scoring"]
    DEPLOY --> EDGE["Edge\n(Azure IoT Edge)\n\n• Local execution\n• No internet connection\n• Ex: factory camera"]
    DEPLOY --> EMBED["Embeddings\n(Direct application)\n\n• Model inside app\n• No network\n• Ex: mobile app"]

2. Model Catalog and Fine-tuning

2.1 Accessing the Model Catalog

The Azure ML Model Catalog is a library of 172+ pre-trained models:

mindmap
  root((Model Catalog\nAzure ML))
    OpenAI
      GPT-4 GPT-4o
      DALL-E 3
      Whisper
      Embeddings
    Microsoft
      Phi-3 mini medium
      Florence 2
      BioMedLM
    Meta
      Llama 3
      Llama 3.1
    Anthropic
      Claude 3 Opus
      Claude 3.5 Sonnet
    Mistral
      Mistral Large
      Mistral 7B
    HuggingFace
      BERT RoBERTa
      Stable Diffusion
      Falcon

2.2 LLM Fine-tuning Workflow

# Fine-tuning an LLM for sentiment classification
# Using Azure ML and a model from the library

from azure.ai.ml import MLClient
from azure.ai.ml.entities import (
    ManagedOnlineEndpoint, ManagedOnlineDeployment
)
from azure.ai.ml.finetuning import FineTuningJob
from azure.identity import DefaultAzureCredential
import os

ml_client = MLClient(
    credential=DefaultAzureCredential(),
    subscription_id=os.environ["AZURE_SUBSCRIPTION_ID"],
    resource_group_name=os.environ["AZURE_RESOURCE_GROUP"],
    workspace_name=os.environ["AZURE_ML_WORKSPACE"]
)

# 1. Prepare fine-tuning data
# Required format for text classification: JSONL
# Each line: {"text": "...", "label": "positive"}
import json

training_data = [
    {"text": "This product is excellent! I highly recommend it.", "label": "positive"},
    {"text": "Terrible experience, I will never buy again.", "label": "negative"},
    {"text": "Decent service, delivered on time.", "label": "neutral"},
    # ... Minimum 100-200 examples per class
]

with open("train_sentiment.jsonl", "w") as f:
    for item in training_data:
        f.write(json.dumps(item, ensure_ascii=False) + "\n")

# 2. Create the fine-tuning job
# (Studio UI offers a guided approach for LLM fine-tuning)
print("""
To fine-tune a model from the catalog via Studio:
1. Go to Model Catalog
2. Select the model (e.g., 'bert-base-multilingual-cased')
3. Click 'Fine-tune'
4. Upload your training data
5. Configure hyperparameters
6. Select compute (GPU recommended)
7. Submit the job (can take 30min to several hours)
8. The fine-tuned model is automatically registered
""")

# 3. Deploy the fine-tuned model (after completion)
# See Section 3 for complete deployment details

3. Online Endpoints – Real-time Inference

3.1 Online Endpoint Architecture

flowchart LR
    CLIENT["📱 Client\nApplication"] -->|"POST /score\n{data: [...]}"| APIGW["Azure API\nManagement\n(optional)"]
    APIGW --> ENDPOINT["🌐 Managed Online\nEndpoint\n(Public URL + Auth)"]
    
    ENDPOINT --> DEP_BLUE["🟦 Blue Deployment\n(v1.0 - 80% traffic)\nStandard_DS2_v2 × 2"]
    ENDPOINT --> DEP_GREEN["🟩 Green Deployment\n(v1.1 - 20% traffic)\nStandard_DS2_v2 × 1"]
    
    DEP_BLUE --> MODEL_B["Model v1.0\n+ Scoring Script"]
    DEP_GREEN --> MODEL_G["Model v1.1\n+ Scoring Script"]
    
    MODEL_B -->|"JSON Response"| ENDPOINT
    MODEL_G -->|"JSON Response"| ENDPOINT
    ENDPOINT --> CLIENT

3.2 Creating a Complete Online Endpoint

# Complete Online Endpoint deployment with Azure ML SDK v2
from azure.ai.ml import MLClient
from azure.ai.ml.entities import (
    ManagedOnlineEndpoint,
    ManagedOnlineDeployment,
    Model,
    CodeConfiguration,
    Environment,
    OnlineRequestSettings,
    ProbeSettings,
    ResourceRequirementsSettings,
    ResourceSettings
)
from azure.ai.ml.constants import AssetTypes
from azure.identity import DefaultAzureCredential
import os
import time

ml_client = MLClient(
    credential=DefaultAzureCredential(),
    subscription_id=os.environ["AZURE_SUBSCRIPTION_ID"],
    resource_group_name=os.environ["AZURE_RESOURCE_GROUP"],
    workspace_name=os.environ["AZURE_ML_WORKSPACE"]
)

# === STEP 1: Create the Endpoint ===
endpoint = ManagedOnlineEndpoint(
    name="vehicle-price-endpoint",
    description="Vehicle price prediction endpoint",
    auth_mode="key",
    tags={
        "project": "vehicle-pricing",
        "team": "ml-team",
        "version": "1.0"
    }
)

print("Creating endpoint...")
created_endpoint = ml_client.online_endpoints.begin_create_or_update(endpoint).result()
print(f"✅ Endpoint created: {created_endpoint.name}")
print(f"   URL: {created_endpoint.scoring_uri}")

# === STEP 2: Deploy the model ===
deployment = ManagedOnlineDeployment(
    name="blue",
    endpoint_name="vehicle-price-endpoint",
    
    # Model from the Azure ML registry
    model=ml_client.models.get("vehicle-price-model", version="1"),
    
    # Scoring script
    code_configuration=CodeConfiguration(
        code="./scoring",
        scoring_script="score.py"
    ),
    
    # Python environment
    environment="AzureML-sklearn-1.0-ubuntu20.04-py38-cpu@latest",
    
    # Compute resources
    instance_type="Standard_DS2_v2",
    instance_count=2,  # Min 2 for high availability
    
    # Request configuration
    request_settings=OnlineRequestSettings(
        max_concurrent_requests_per_instance=10,
        request_timeout_ms=5000,
        max_queue_wait_ms=500
    ),
    
    # Health probes
    liveness_probe=ProbeSettings(
        failure_threshold=30,
        success_threshold=1,
        timeout=2,
        period=10,
        initial_delay=10
    ),
    readiness_probe=ProbeSettings(
        failure_threshold=10,
        success_threshold=1,
        timeout=2,
        period=10,
        initial_delay=10
    )
)

print("\nDeploying model...")
created_deployment = ml_client.online_deployments.begin_create_or_update(deployment).result()
print(f"✅ Deployment created: {created_deployment.name}")

# === STEP 3: Route 100% of traffic to this deployment ===
created_endpoint.traffic = {"blue": 100}
ml_client.online_endpoints.begin_create_or_update(created_endpoint).result()
print(f"✅ 100% of traffic routed to 'blue'")

# === STEP 4: Test the endpoint ===
import json
import urllib.request

def test_endpoint(endpoint_name: str, test_data: dict) -> dict:
    """
    Sends a test request to the endpoint.
    
    Args:
        endpoint_name: Endpoint name
        test_data: Test data
    
    Returns:
        API response
    """
    # Retrieve the URL and key
    endpoint = ml_client.online_endpoints.get(endpoint_name)
    keys = ml_client.online_endpoints.get_keys(endpoint_name)
    
    scoring_uri = endpoint.scoring_uri
    api_key = keys.primary_key
    
    # Prepare the request
    payload = json.dumps({"data": [list(test_data.values())]}).encode("utf-8")
    
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    
    req = urllib.request.Request(scoring_uri, payload, headers)
    
    start = time.time()
    with urllib.request.urlopen(req) as response:
        result = json.loads(response.read())
    latency_ms = (time.time() - start) * 1000
    
    return {
        "prediction": result,
        "latency_ms": round(latency_ms, 1),
        "endpoint": endpoint_name
    }

# Test
test_vehicle = {
    "symboling": 2,
    "wheel_base": 99.8,
    "length": 176.6,
    "width": 66.2,
    "height": 54.3,
    "curb_weight": 2337,
    "engine_size": 109,
    "horsepower": 102,
    "city_mpg": 24,
    "highway_mpg": 30
}

print("\n=== Endpoint Test ===")
test_result = test_endpoint("vehicle-price-endpoint", test_vehicle)
print(f"Prediction: ${test_result['prediction'][0]:,.2f}")
print(f"Latency: {test_result['latency_ms']:.1f}ms")

4. Scoring Script – The Core of Deployment

4.1 Scoring Script Structure

# score.py - Standard scoring script for Online Endpoint
"""
Scoring script template for Azure ML Online Endpoint.

Required:
- init(): Loaded ONCE at instance startup
- run(data): Called for EACH request
"""

import json
import os
import logging
import numpy as np
import pandas as pd
import joblib
from typing import Union

# Configure logging
logger = logging.getLogger("scoring_script")
logging.basicConfig(level=logging.INFO)

# Global variables for the model (loaded in init())
model = None
feature_names = None
scaler = None

def init():
    """
    Initialization: loaded ONCE at startup.
    Load the model here to avoid reloading on every request.
    """
    global model, feature_names, scaler
    
    # Model path (defined by Azure ML)
    model_dir = os.environ.get("AZUREML_MODEL_DIR", "./model")
    
    try:
        # Load the main model
        model_path = os.path.join(model_dir, "model", "model.joblib")
        model = joblib.load(model_path)
        logger.info(f"✅ Model loaded from: {model_path}")
        
        # Load the scaler if present
        scaler_path = os.path.join(model_dir, "model", "scaler.joblib")
        if os.path.exists(scaler_path):
            scaler = joblib.load(scaler_path)
            logger.info("✅ Scaler loaded")
        
        # Load feature names
        features_path = os.path.join(model_dir, "model", "feature_names.json")
        if os.path.exists(features_path):
            with open(features_path) as f:
                feature_names = json.load(f)
            logger.info(f"✅ Features loaded: {feature_names}")
    
    except Exception as e:
        logger.error(f"❌ Error during loading: {e}")
        raise

def run(raw_data: Union[str, bytes]) -> str:
    """
    Scoring: called for each incoming request.
    
    Args:
        raw_data: JSON data from the request
        
    Returns:
        JSON with predictions
    
    Expected input format:
    {
        "data": [[feature1, feature2, ...], [feature1, feature2, ...]]
    }
    or
    {
        "data": [{"feature1": value1, "feature2": value2}, ...]
    }
    """
    try:
        # Parse the input data
        if isinstance(raw_data, bytes):
            raw_data = raw_data.decode("utf-8")
        
        input_data = json.loads(raw_data)
        
        # Extract data
        if "data" not in input_data:
            return json.dumps({
                "error": "Field 'data' is required",
                "expected_format": '{"data": [[val1, val2, ...]]}'
            })
        
        data = input_data["data"]
        
        # Convert to DataFrame or numpy array
        if isinstance(data[0], dict):
            # Dictionary format: {"feature1": val1, ...}
            df = pd.DataFrame(data)
            if feature_names:
                df = df[feature_names]  # Preserve feature order
            X = df.values
        else:
            # List format: [[val1, val2, ...]]
            X = np.array(data)
        
        # Validate dimensions
        if model is not None and hasattr(model, 'n_features_in_'):
            expected_features = model.n_features_in_
            if X.shape[1] != expected_features:
                return json.dumps({
                    "error": f"Incorrect number of features: received {X.shape[1]}, expected {expected_features}"
                })
        
        # Apply scaling if available
        if scaler is not None:
            X = scaler.transform(X)
        
        # Predictions
        predictions = model.predict(X).tolist()
        
        # For classification: add probabilities
        results = {"predictions": predictions}
        
        if hasattr(model, 'predict_proba'):
            probas = model.predict_proba(X).tolist()
            results["probabilities"] = probas
        
        logger.info(f"✅ {len(predictions)} predictions generated")
        
        return json.dumps(results)
    
    except json.JSONDecodeError as e:
        logger.error(f"Invalid JSON: {e}")
        return json.dumps({"error": f"Invalid JSON: {str(e)}"})
    
    except Exception as e:
        logger.error(f"Scoring error: {e}", exc_info=True)
        return json.dumps({"error": f"Internal error: {str(e)}"})

4.2 Advanced Scoring Script with Validation

# score_advanced.py - Scoring script with validation and monitoring
import json
import os
import logging
import time
import numpy as np
import pandas as pd
import joblib
from dataclasses import dataclass
from typing import Any
import mlflow

logger = logging.getLogger("scoring_advanced")

# Global metrics
scoring_metrics = {
    "request_count": 0,
    "error_count": 0,
    "total_latency_ms": 0
}

model = None
validation_schema = None

@dataclass
class FeatureSchema:
    """Feature validation schema."""
    name: str
    expected_type: type
    min_val: float = None
    max_val: float = None
    nullable: bool = False

# Feature schema for a vehicle price model
VEHICLE_PRICE_SCHEMA = [
    FeatureSchema("symboling", int, -3, 3),
    FeatureSchema("wheel_base", float, 60, 130),
    FeatureSchema("length", float, 100, 250),
    FeatureSchema("width", float, 50, 80),
    FeatureSchema("height", float, 40, 80),
    FeatureSchema("curb_weight", int, 1000, 4000),
    FeatureSchema("engine_size", int, 50, 400),
    FeatureSchema("horsepower", float, 50, 400),
    FeatureSchema("city_mpg", int, 5, 60),
    FeatureSchema("highway_mpg", int, 10, 80)
]

def init():
    """Model initialization."""
    global model, validation_schema
    
    model_dir = os.environ.get("AZUREML_MODEL_DIR", "./model")
    model_path = os.path.join(model_dir, "model", "model.joblib")
    
    model = joblib.load(model_path)
    validation_schema = VEHICLE_PRICE_SCHEMA
    
    logger.info(f"✅ Model loaded. Expected features: {len(validation_schema)}")

def validate_features(data: list[dict]) -> tuple[list[str], list[dict]]:
    """
    Validates and cleans input features.
    
    Returns:
        (errors, validated_data)
    """
    errors = []
    validated_data = []
    
    for idx, row in enumerate(data):
        validated_row = {}
        
        for schema in validation_schema:
            value = row.get(schema.name)
            
            # Check presence
            if value is None:
                if not schema.nullable:
                    errors.append(f"Row {idx}: Feature '{schema.name}' missing")
                    continue
                value = np.nan
            
            # Check bounds
            if isinstance(value, (int, float)) and not np.isnan(value):
                if schema.min_val is not None and value < schema.min_val:
                    errors.append(f"Row {idx}: '{schema.name}'={value} < min={schema.min_val}")
                if schema.max_val is not None and value > schema.max_val:
                    errors.append(f"Row {idx}: '{schema.name}'={value} > max={schema.max_val}")
            
            validated_row[schema.name] = value
        
        validated_data.append(validated_row)
    
    return errors, validated_data

def run(raw_data: str) -> str:
    """Scoring with complete validation and metrics."""
    global scoring_metrics
    
    start = time.time()
    scoring_metrics["request_count"] += 1
    
    try:
        # Parse
        input_data = json.loads(raw_data)
        data = input_data.get("data", [])
        
        if not data:
            raise ValueError("No data provided")
        
        # Convert to list of dicts if necessary
        if isinstance(data[0], list):
            feature_names = [s.name for s in validation_schema]
            data = [dict(zip(feature_names, row)) for row in data]
        
        # Validate
        errors, valid_data = validate_features(data)
        
        if errors:
            return json.dumps({
                "error": "Feature validation failed",
                "details": errors[:10]  # Max 10 errors
            })
        
        # Prepare X
        X = pd.DataFrame(valid_data)[[s.name for s in validation_schema]].values
        
        # Predict
        predictions = model.predict(X).tolist()
        
        # Calculate metrics
        latency_ms = (time.time() - start) * 1000
        scoring_metrics["total_latency_ms"] += latency_ms
        avg_latency = scoring_metrics["total_latency_ms"] / scoring_metrics["request_count"]
        
        return json.dumps({
            "predictions": predictions,
            "prediction_count": len(predictions),
            "latency_ms": round(latency_ms, 1),
            "avg_latency_ms": round(avg_latency, 1)
        })
    
    except Exception as e:
        scoring_metrics["error_count"] += 1
        logger.error(f"Error: {e}", exc_info=True)
        return json.dumps({"error": str(e)})

5. Blue/Green Deployment and Traffic Splitting

5.1 Blue/Green Strategy

flowchart TD
    CLIENT["Client requests"] --> ENDPOINT["Online Endpoint"]
    
    ENDPOINT -->|"100% traffic\ninitial"| BLUE["🟦 Blue (v1.0)\n(Current Production)"]
    
    subgraph "Update process"
        DEPLOY_GREEN["Deploy Green (v1.1)\n0% traffic"] --> TEST_CANARY["Canary Test\n10% traffic to Green"]
        TEST_CANARY --> VALIDATE["Validate metrics\n(Latency, Errors, Accuracy)"]
        VALIDATE -->|"Metrics OK"| PROGRESSIVE["Progressively increase\n10% → 50% → 100%"]
        VALIDATE -->|"Issue detected"| ROLLBACK["Immediate rollback\n100% → Blue"]
        PROGRESSIVE --> DECOMMISSION["Decommission Blue\n(or keep on standby)"]
    end
    
    ENDPOINT -->|"0% initial\n→ 10% → 50% → 100%"| GREEN["🟩 Green (v1.1)\n(New Version)"]

# Complete Blue/Green deployment with Azure ML SDK v2
from azure.ai.ml.entities import ManagedOnlineEndpoint, ManagedOnlineDeployment
import time

class BlueGreenDeployment:
    """
    Blue/Green deployment manager for Azure ML.
    
    Pattern:
    1. Deploy new version (green) with 0% traffic
    2. Test the green version
    3. Progressively switch traffic
    4. Decommission the old version (blue)
    """
    
    def __init__(self, endpoint_name: str):
        self.endpoint_name = endpoint_name
        self.ml_client = MLClient(
            credential=DefaultAzureCredential(),
            subscription_id=os.environ["AZURE_SUBSCRIPTION_ID"],
            resource_group_name=os.environ["AZURE_RESOURCE_GROUP"],
            workspace_name=os.environ["AZURE_ML_WORKSPACE"]
        )
    
    def get_active_deployments(self) -> dict:
        """Returns current deployments and their traffic."""
        endpoint = self.ml_client.online_endpoints.get(self.endpoint_name)
        return endpoint.traffic or {}
    
    def deploy_new_version(
        self, 
        deployment_name: str,
        model_version: str,
        scoring_script_dir: str,
        instance_type: str = "Standard_DS2_v2",
        instance_count: int = 2
    ) -> ManagedOnlineDeployment:
        """
        Deploys a new version with 0% traffic.
        
        Args:
            deployment_name: Name of the new deployment (e.g., "green")
            model_version: Model version in the registry
            scoring_script_dir: Folder containing score.py
        """
        print(f"📦 Deploying version '{deployment_name}' (0% traffic)...")
        
        deployment = ManagedOnlineDeployment(
            name=deployment_name,
            endpoint_name=self.endpoint_name,
            model=self.ml_client.models.get(
                "vehicle-price-model", 
                version=model_version
            ),
            code_configuration=CodeConfiguration(
                code=scoring_script_dir,
                scoring_script="score.py"
            ),
            environment="AzureML-sklearn-1.0-ubuntu20.04-py38-cpu@latest",
            instance_type=instance_type,
            instance_count=instance_count
        )
        
        result = self.ml_client.online_deployments.begin_create_or_update(deployment).result()
        
        # Ensure this deployment has no traffic
        endpoint = self.ml_client.online_endpoints.get(self.endpoint_name)
        traffic = endpoint.traffic or {}
        traffic[deployment_name] = 0
        # Redistribute so total = 100%
        endpoint.traffic = traffic
        self.ml_client.online_endpoints.begin_create_or_update(endpoint).result()
        
        print(f"✅ '{deployment_name}' deployed (0% traffic for now)")
        return result
    
    def test_deployment(
        self, 
        deployment_name: str,
        test_data: dict,
        num_requests: int = 10
    ) -> dict:
        """
        Tests a specific deployment directly (without going through main traffic).
        """
        print(f"🧪 Testing '{deployment_name}'...")
        
        endpoint = self.ml_client.online_endpoints.get(self.endpoint_name)
        keys = self.ml_client.online_endpoints.get_keys(self.endpoint_name)
        
        scoring_uri = f"{endpoint.scoring_uri}"
        
        successes = 0
        latencies = []
        
        for i in range(num_requests):
            start = time.time()
            try:
                payload = json.dumps({"data": [list(test_data.values())]}).encode("utf-8")
                headers = {
                    "Content-Type": "application/json",
                    "Authorization": f"Bearer {keys.primary_key}",
                    "azureml-model-deployment": deployment_name  # Route to this deployment
                }
                req = urllib.request.Request(scoring_uri, payload, headers)
                with urllib.request.urlopen(req, timeout=5) as resp:
                    json.loads(resp.read())
                    successes += 1
            except Exception as e:
                print(f"  Request {i+1} failed: {e}")
            
            latencies.append((time.time() - start) * 1000)
        
        return {
            "success_rate": successes / num_requests,
            "avg_latency_ms": sum(latencies) / len(latencies),
            "p99_latency_ms": sorted(latencies)[int(0.99 * len(latencies))]
        }
    
    def switch_traffic(
        self,
        current_deployment: str,
        new_deployment: str,
        new_percentage: int,
        validate_before: bool = True
    ) -> bool:
        """
        Progressively switches traffic.
        
        Args:
            current_deployment: Name of current deployment (blue)
            new_deployment: Name of new deployment (green)
            new_percentage: % to route to new deployment (0-100)
            validate_before: If True, checks metrics before switching
        
        Returns:
            True if success, False if rollback performed
        """
        current_percentage = 100 - new_percentage
        
        print(f"🔄 Switching: {current_deployment}={current_percentage}% / {new_deployment}={new_percentage}%")
        
        try:
            endpoint = self.ml_client.online_endpoints.get(self.endpoint_name)
            endpoint.traffic = {
                current_deployment: current_percentage,
                new_deployment: new_percentage
            }
            self.ml_client.online_endpoints.begin_create_or_update(endpoint).result()
            
            print(f"✅ Traffic updated: {current_deployment}={current_percentage}% | {new_deployment}={new_percentage}%")
            return True
        
        except Exception as e:
            print(f"❌ Error during switch: {e}")
            # Automatic rollback
            self.rollback(current_deployment)
            return False
    
    def rollback(self, stable_deployment: str) -> None:
        """Immediate rollback to the stable deployment."""
        print(f"⚠️ ROLLBACK to '{stable_deployment}'...")
        
        endpoint = self.ml_client.online_endpoints.get(self.endpoint_name)
        endpoint.traffic = {stable_deployment: 100}
        
        # Find and set other deployments to 0
        for dep_name in list(endpoint.traffic.keys()):
            if dep_name != stable_deployment:
                endpoint.traffic[dep_name] = 0
        
        self.ml_client.online_endpoints.begin_create_or_update(endpoint).result()
        print(f"✅ Rollback complete. 100% of traffic on '{stable_deployment}'")
    
    def complete_migration(
        self,
        old_deployment: str,
        new_deployment: str
    ) -> None:
        """
        Finalizes migration: 100% on new, deletes the old.
        """
        print(f"🎉 Final migration to '{new_deployment}'")
        
        # 100% on the new
        endpoint = self.ml_client.online_endpoints.get(self.endpoint_name)
        endpoint.traffic = {new_deployment: 100}
        self.ml_client.online_endpoints.begin_create_or_update(endpoint).result()
        
        # Delete the old
        self.ml_client.online_deployments.begin_delete(
            endpoint_name=self.endpoint_name,
            deployment_name=old_deployment
        ).result()
        
        print(f"✅ '{old_deployment}' deleted. '{new_deployment}' = 100% traffic")

# Blue/Green manager usage
manager = BlueGreenDeployment("vehicle-price-endpoint")

# 1. Check current deployments
print("Current deployments:", manager.get_active_deployments())

# 2. Deploy new version
manager.deploy_new_version(
    deployment_name="green",
    model_version="2",
    scoring_script_dir="./scoring",
    instance_count=2
)

# 3. Test the new version
test_data = {"symboling": 2, "wheel_base": 99.8, "length": 176.6}
test_results = manager.test_deployment("green", test_data, num_requests=20)
print(f"\nTest results: {test_results}")

# 4. If tests OK, progressively switch
if test_results["success_rate"] > 0.98:
    # Canary phase: 10%
    manager.switch_traffic("blue", "green", 10)
    time.sleep(60)  # Monitor for 1 minute
    
    # Phase 50%
    manager.switch_traffic("blue", "green", 50)
    time.sleep(60)
    
    # Complete migration
    manager.complete_migration("blue", "green")
else:
    manager.rollback("blue")

6. Batch Endpoints – Batch Inference

6.1 Batch Endpoint Configuration

# Batch Endpoint for mass scoring
from azure.ai.ml.entities import (
    BatchEndpoint, BatchDeployment, BatchRetrySettings
)
from azure.ai.ml.constants import BatchDeploymentOutputAction

# Batch scoring script
BATCH_SCORING_SCRIPT = '''
import os
import pandas as pd
import numpy as np
import joblib
from typing import List

# Global variable
model = None

def init():
    """Loaded once at startup of each worker."""
    global model
    model_dir = os.environ.get("AZUREML_MODEL_DIR", "./model")
    model = joblib.load(os.path.join(model_dir, "model", "model.joblib"))
    print(f"Worker initialized with model: {type(model).__name__}")

def run(mini_batch: List[str]) -> pd.DataFrame:
    """
    Processes a mini-batch of files.
    
    Args:
        mini_batch: List of file paths to score
    
    Returns:
        DataFrame with predictions
    """
    results = []
    
    for file_path in mini_batch:
        print(f"Processing: {os.path.basename(file_path)}")
        
        # Load data
        if file_path.endswith(".csv"):
            df = pd.read_csv(file_path)
        elif file_path.endswith(".parquet"):
            df = pd.read_parquet(file_path)
        else:
            print(f"Unsupported format: {file_path}")
            continue
        
        # Features
        feature_cols = [c for c in df.columns if c not in ["id", "actual_price"]]
        X = df[feature_cols].values
        
        # Predictions
        predictions = model.predict(X)
        
        # Create results DataFrame
        df_result = pd.DataFrame({
            "id": df.get("id", range(len(df))),
            "predicted_price": predictions,
            "source_file": os.path.basename(file_path)
        })
        
        results.append(df_result)
    
    if results:
        return pd.concat(results, ignore_index=True)
    else:
        return pd.DataFrame(columns=["id", "predicted_price", "source_file"])
'''

os.makedirs("./batch_scoring", exist_ok=True)
with open("./batch_scoring/score_batch.py", "w") as f:
    f.write(BATCH_SCORING_SCRIPT)

# Create the Batch Endpoint
batch_endpoint = BatchEndpoint(
    name="vehicle-price-batch",
    description="Batch endpoint for mass scoring"
)
ml_client.batch_endpoints.begin_create_or_update(batch_endpoint).result()

# Create the Batch Deployment
batch_deployment = BatchDeployment(
    name="vehicle-price-batch-v1",
    endpoint_name="vehicle-price-batch",
    
    model=ml_client.models.get("vehicle-price-model", version="1"),
    code_configuration=CodeConfiguration(
        code="./batch_scoring",
        scoring_script="score_batch.py"
    ),
    environment="AzureML-sklearn-1.0-ubuntu20.04-py38-cpu@latest",
    compute="cpu-cluster-4cores",
    
    # Batch configuration
    mini_batch_size=50,          # Files per mini-batch
    max_concurrency_per_instance=4,
    instance_count=4,            # 4 parallel workers
    
    # Output
    output_action=BatchDeploymentOutputAction.APPEND_ROW,
    output_file_name="predictions.csv",
    
    # Retry
    retry_settings=BatchRetrySettings(max_retries=3, timeout=300)
)

ml_client.batch_deployments.begin_create_or_update(batch_deployment).result()

# Launch a batch job
batch_job = ml_client.batch_endpoints.invoke(
    endpoint_name="vehicle-price-batch",
    inputs={
        "data": Input(
            path="azureml:vehicle-data-to-score:1",
            type=AssetTypes.URI_FOLDER
        )
    }
)
print(f"Batch job launched: {batch_job.name}")

7. Deployment on AKS (Kubernetes)

7.1 When to Use AKS?

flowchart TD
    Q{"What level of\nscalability?"}
    Q -->|"< 100 req/s"| MANAGED["Managed Online Endpoint\n(Recommended)"]
    Q -->|"> 100 req/s"| AKS["AKS - Azure Kubernetes Service\n• Full control\n• Massive scale\n• Cost-optimized at scale"]
    Q -->|"> 1000 req/s"| AKS_PREMIUM["AKS Premium\n• KEDA Auto-scaling\n• GPU Nodes\n• Multi-region"]

# Deployment on AKS with Azure ML
from azure.ai.ml.entities import KubernetesOnlineEndpoint, KubernetesOnlineDeployment

# Attach an existing AKS cluster
from azure.ai.ml.entities import KubernetesCompute

aks_compute = KubernetesCompute(
    name="aks-scoring-cluster",
    namespace="azure-ml",
    default_instance_type="Standard_DS3_v2"
)

ml_client.compute.begin_create_or_update(aks_compute).result()
print("✅ AKS attached to workspace")

# Create a Kubernetes endpoint
aks_endpoint = KubernetesOnlineEndpoint(
    name="vehicle-price-aks",
    compute="aks-scoring-cluster",
    auth_mode="key"
)
ml_client.online_endpoints.begin_create_or_update(aks_endpoint).result()

# Deploy on AKS
aks_deployment = KubernetesOnlineDeployment(
    name="vehicle-price-aks-v1",
    endpoint_name="vehicle-price-aks",
    model=ml_client.models.get("vehicle-price-model", version="1"),
    code_configuration=CodeConfiguration(
        code="./scoring",
        scoring_script="score.py"
    ),
    environment="AzureML-sklearn-1.0-ubuntu20.04-py38-cpu@latest",
    instance_type="Standard_DS3_v2",
    instance_count=3,
    # AKS auto-scaling configuration
    resources={
        "requests": {"cpu": "0.5", "memory": "512Mi"},
        "limits": {"cpu": "2", "memory": "2Gi"}
    }
)

ml_client.online_deployments.begin_create_or_update(aks_deployment).result()
print("✅ Model deployed on AKS")

8. Model Monitoring and Surveillance

8.1 Metrics to Monitor

flowchart TD
    MONITOR["🔍 Production\nModel Monitoring"] --> PERF["Endpoint\nPerformance\n\n• Latency (P50, P95, P99)\n• Throughput (req/s)\n• Error rate (4xx, 5xx)\n• Availability (%)"]
    MONITOR --> DRIFT["Data Drift\n\n• Feature distribution\n• Change vs baseline\n• Automatic alerts"]
    MONITOR --> MODEL_PERF["Model\nPerformance\n\n• Prod vs train accuracy\n• Business metrics\n• A/B comparison"]
    MONITOR --> COST["Costs\n\n• Tokens consumed\n• Compute hours\n• Cost per prediction"]

# Azure ML monitoring configuration
from azure.ai.ml.entities import (
    MonitorSchedule, RecurrenceTrigger,
    MonitoringTarget, AlertNotification,
    DataDriftSignal, PredictionDriftSignal
)

# Create a monitor to detect drift
def configure_model_monitoring(
    endpoint_name: str,
    deployment_name: str,
    baseline_dataset: str  # Reference dataset (training data)
) -> None:
    """
    Configures automatic monitoring for an Azure ML endpoint.
    
    Monitors:
    - Input data drift (data drift)
    - Prediction drift (prediction drift)
    """
    from azure.ai.ml.entities import (
        MonitorDefinition, MonitorFeatureFilter,
        AlertNotification, EmailNotificationAction
    )
    
    monitor_definition = MonitorDefinition(
        monitoring_target=MonitoringTarget(
            endpoint_deployment_id=f"azureml:{endpoint_name}:{deployment_name}"
        ),
        alert_notification=AlertNotification(
            emails=["ml-team@company.com"]
        ),
        signals={
            "data_drift": DataDriftSignal(
                reference_data=Input(
                    path=f"azureml:{baseline_dataset}",
                    type=AssetTypes.URI_FOLDER
                ),
                alert_enabled=True,
                threshold=0.3  # Alert if drift > 30%
            ),
            "prediction_drift": PredictionDriftSignal(
                alert_enabled=True,
                threshold=0.3
            )
        }
    )
    
    monitor = MonitorSchedule(
        name=f"monitor-{endpoint_name}",
        trigger=RecurrenceTrigger(frequency="day", interval=1),
        create_monitor=monitor_definition
    )
    
    ml_client.schedules.begin_create_or_update(monitor)
    print(f"✅ Monitor configured for {endpoint_name}")
    print("   Frequency: daily")
    print("   Alerts: ml-team@company.com")

# Simple monitoring example with Azure Monitor SDK
import requests
from datetime import datetime, timedelta

def get_endpoint_metrics(
    endpoint_name: str,
    period_hours: int = 24
) -> dict:
    """
    Retrieves performance metrics for an endpoint.
    
    Available metrics:
    - RequestsPerMinute
    - Latency
    - ErrorRate
    - SuccessRate
    """
    endpoint = ml_client.online_endpoints.get(endpoint_name)
    
    print(f"\n=== Endpoint Metrics: {endpoint_name} ===")
    print(f"   URL: {endpoint.scoring_uri}")
    print(f"   Traffic: {endpoint.traffic}")
    print(f"   Auth: {endpoint.auth_mode}")
    
    # Detailed metrics are in Azure Monitor
    # Here we return a summary
    return {
        "endpoint_name": endpoint_name,
        "status": endpoint.provisioning_state,
        "scoring_uri": endpoint.scoring_uri,
        "traffic": endpoint.traffic,
        "note": "Detailed metrics available in Azure Monitor/Application Insights"
    }

# Display metrics
metrics = get_endpoint_metrics("vehicle-price-endpoint")
print(f"\n{json.dumps(metrics, indent=2)}")

9. CI/CD for ML Deployment

9.1 Continuous Deployment Workflow

flowchart TD
    PR["Pull Request\n(New model)"] -->|"Merge to main"| CI["CI Pipeline\n(GitHub Actions)"]
    CI --> TESTS["Unit Tests\n+ Data Validation"]
    TESTS --> TRAIN["Train model\n(Azure ML Pipeline)"]
    TRAIN --> EVAL["Evaluate metrics\n(R² > 0.85?)"]
    EVAL -->|"Metrics OK"| REGISTER["Register model\n(Model Registry v+1)"]
    EVAL -->|"Metrics KO"| NOTIF["Failure\nNotification"]
    REGISTER --> STAGING["Deploy Staging\n(0 → 100%)"]
    STAGING --> TEST_INT["Integration Tests\n(Smoke tests)"]
    TEST_INT -->|"OK"| APPROVAL["⚠️ Manual Approval\n(Prod)"]
    APPROVAL --> PROD["Deploy Production\n(Blue/Green)"]
    PROD --> MONITOR["Monitor\n(Alerts if issue)"]
    MONITOR -->|"Anomaly"| ROLLBACK["Auto rollback\n(previous version)"]

# .github/workflows/deploy_model.yml
name: Deploy ML Model

on:
  push:
    branches: [main]
    paths:
      - 'ml/models/**'
      - 'ml/scoring/**'

env:
  AZURE_SUBSCRIPTION_ID: ${{ secrets.AZURE_SUBSCRIPTION_ID }}
  AZURE_RESOURCE_GROUP: ${{ secrets.AZURE_RESOURCE_GROUP }}
  AZURE_ML_WORKSPACE: ${{ secrets.AZURE_ML_WORKSPACE }}
  ENDPOINT_NAME: vehicle-price-endpoint

jobs:
  train-and-evaluate:
    name: Train and Evaluate
    runs-on: ubuntu-latest
    outputs:
      model_version: ${{ steps.register.outputs.version }}
      metrics_ok: ${{ steps.evaluate.outputs.ok }}
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Azure Login
        uses: azure/login@v2
        with:
          creds: ${{ secrets.AZURE_CREDENTIALS }}
      
      - name: Install Azure ML SDK
        run: pip install azure-ai-ml azure-identity
      
      - name: Run Training Pipeline
        id: train
        run: |
          python ml/pipeline.py \
            --experiment_name "ci-deploy-${{ github.run_number }}"
      
      - name: Evaluate Metrics
        id: evaluate
        run: |
          python ml/check_quality.py \
            --experiment "ci-deploy-${{ github.run_number }}" \
            --min_r2 0.85
          echo "ok=true" >> $GITHUB_OUTPUT
      
      - name: Register Model
        id: register
        if: steps.evaluate.outputs.ok == 'true'
        run: |
          VERSION=$(python ml/register_model.py \
            --experiment "ci-deploy-${{ github.run_number }}" \
            --model_name "vehicle-price-model")
          echo "version=$VERSION" >> $GITHUB_OUTPUT

  deploy-staging:
    name: Deploy Staging
    needs: train-and-evaluate
    if: needs.train-and-evaluate.outputs.metrics_ok == 'true'
    runs-on: ubuntu-latest
    environment: staging
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Azure Login
        uses: azure/login@v2
        with:
          creds: ${{ secrets.AZURE_CREDENTIALS }}
      
      - name: Deploy to Staging
        run: |
          python ml/deploy.py \
            --endpoint $ENDPOINT_NAME-staging \
            --model_version ${{ needs.train-and-evaluate.outputs.model_version }} \
            --environment staging
      
      - name: Run Smoke Tests
        run: |
          python ml/smoke_tests.py \
            --endpoint $ENDPOINT_NAME-staging \
            --num_tests 100

  deploy-production:
    name: Deploy Production
    needs: [train-and-evaluate, deploy-staging]
    runs-on: ubuntu-latest
    environment: production  # Requires manual approval
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Azure Login
        uses: azure/login@v2
        with:
          creds: ${{ secrets.AZURE_CREDENTIALS }}
      
      - name: Deploy Blue/Green
        run: |
          python ml/blue_green_deploy.py \
            --endpoint $ENDPOINT_NAME \
            --model_version ${{ needs.train-and-evaluate.outputs.model_version }} \
            --canary_percent 10
      
      - name: Monitor Canary
        run: |
          sleep 300  # Wait 5 minutes
          python ml/check_canary_health.py \
            --endpoint $ENDPOINT_NAME \
            --max_error_rate 0.01
      
      - name: Complete Migration
        run: |
          python ml/complete_migration.py \
            --endpoint $ENDPOINT_NAME

10. Complete Implementation with the SDK

10.1 Complete Deployment Management Class

# deployment_manager.py - Complete Azure ML deployment manager
from azure.ai.ml import MLClient
from azure.ai.ml.entities import (
    ManagedOnlineEndpoint, ManagedOnlineDeployment,
    CodeConfiguration, Model
)
from azure.identity import DefaultAzureCredential
from dataclasses import dataclass
import json
import os
import time
import urllib.request
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("deployment_manager")

@dataclass
class DeploymentConfig:
    """Configuration for a deployment."""
    endpoint_name: str
    model_name: str
    model_version: str
    scoring_script_dir: str
    instance_type: str = "Standard_DS2_v2"
    instance_count: int = 2
    env_name: str = "AzureML-sklearn-1.0-ubuntu20.04-py38-cpu@latest"
    description: str = ""

class DeploymentManager:
    """Centralized Azure ML deployment manager."""
    
    def __init__(self):
        self.ml_client = MLClient(
            credential=DefaultAzureCredential(),
            subscription_id=os.environ["AZURE_SUBSCRIPTION_ID"],
            resource_group_name=os.environ["AZURE_RESOURCE_GROUP"],
            workspace_name=os.environ["AZURE_ML_WORKSPACE"]
        )
    
    def create_or_update_endpoint(
        self, 
        name: str, 
        description: str = ""
    ) -> ManagedOnlineEndpoint:
        """Creates or updates an endpoint."""
        endpoint = ManagedOnlineEndpoint(
            name=name,
            description=description,
            auth_mode="key",
            tags={"managed_by": "deployment_manager"}
        )
        
        return self.ml_client.online_endpoints.begin_create_or_update(endpoint).result()
    
    def deploy(self, config: DeploymentConfig, deployment_name: str) -> bool:
        """
        Deploys a model according to the provided configuration.
        
        Returns:
            True if success, False otherwise
        """
        try:
            # Create/Update the endpoint
            logger.info(f"Configuring endpoint: {config.endpoint_name}")
            self.create_or_update_endpoint(
                config.endpoint_name, 
                config.description
            )
            
            # Create the deployment
            logger.info(f"Deploying '{deployment_name}'...")
            deployment = ManagedOnlineDeployment(
                name=deployment_name,
                endpoint_name=config.endpoint_name,
                model=self.ml_client.models.get(
                    config.model_name, 
                    version=config.model_version
                ),
                code_configuration=CodeConfiguration(
                    code=config.scoring_script_dir,
                    scoring_script="score.py"
                ),
                environment=config.env_name,
                instance_type=config.instance_type,
                instance_count=config.instance_count
            )
            
            self.ml_client.online_deployments.begin_create_or_update(deployment).result()
            
            logger.info(f"✅ Deployment '{deployment_name}' successful")
            return True
        
        except Exception as e:
            logger.error(f"❌ Deployment error: {e}", exc_info=True)
            return False
    
    def route_traffic(self, endpoint_name: str, routing: dict) -> bool:
        """
        Configures traffic routing.
        
        Args:
            endpoint_name: Endpoint name
            routing: Dict {deployment_name: percentage}
                     Ex: {"blue": 80, "green": 20}
        
        Returns:
            True if success
        """
        assert sum(routing.values()) == 100, "Total traffic must be 100%"
        
        try:
            endpoint = self.ml_client.online_endpoints.get(endpoint_name)
            endpoint.traffic = routing
            self.ml_client.online_endpoints.begin_create_or_update(endpoint).result()
            
            logger.info(f"✅ Traffic configured: {routing}")
            return True
        
        except Exception as e:
            logger.error(f"❌ Routing error: {e}")
            return False
    
    def health_check(
        self, 
        endpoint_name: str, 
        test_data: dict,
        deployment_name: str = None
    ) -> dict:
        """
        Checks the health of an endpoint.
        
        Returns:
            Health report {status, latency, prediction}
        """
        endpoint = self.ml_client.online_endpoints.get(endpoint_name)
        keys = self.ml_client.online_endpoints.get_keys(endpoint_name)
        
        payload = json.dumps({"data": [list(test_data.values())]}).encode("utf-8")
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {keys.primary_key}"
        }
        
        if deployment_name:
            headers["azureml-model-deployment"] = deployment_name
        
        start = time.time()
        try:
            req = urllib.request.Request(endpoint.scoring_uri, payload, headers)
            with urllib.request.urlopen(req, timeout=10) as resp:
                response = json.loads(resp.read())
                latency = (time.time() - start) * 1000
                
                return {
                    "status": "healthy",
                    "latency_ms": round(latency, 1),
                    "prediction": response
                }
        
        except Exception as e:
            return {
                "status": "unhealthy",
                "error": str(e),
                "latency_ms": (time.time() - start) * 1000
            }
    
    def list_endpoints(self) -> list[dict]:
        """Lists all endpoints with their status."""
        endpoints = []
        for ep in self.ml_client.online_endpoints.list():
            endpoints.append({
                "name": ep.name,
                "status": ep.provisioning_state,
                "url": ep.scoring_uri,
                "traffic": ep.traffic or {}
            })
        return endpoints
    
    def delete_endpoint(self, name: str, confirm: bool = False) -> bool:
        """Deletes an endpoint (requires confirmation)."""
        if not confirm:
            print(f"⚠️ To delete '{name}', call with confirm=True")
            return False
        
        self.ml_client.online_endpoints.begin_delete(name).result()
        logger.info(f"✅ Endpoint '{name}' deleted")
        return True

# Usage
manager = DeploymentManager()

# Deployment configuration
config = DeploymentConfig(
    endpoint_name="vehicle-price-endpoint",
    model_name="vehicle-price-model",
    model_version="2",
    scoring_script_dir="./scoring",
    instance_type="Standard_DS2_v2",
    instance_count=2,
    description="Vehicle price prediction endpoint v2"
)

# Deploy new version
success = manager.deploy(config, deployment_name="green")

if success:
    # Health check
    test = manager.health_check(
        "vehicle-price-endpoint",
        {"symboling": 2, "wheel_base": 99.8, "length": 176.6},
        deployment_name="green"
    )
    print(f"Health: {test['status']} ({test['latency_ms']:.1f}ms)")
    
    # Switch traffic
    if test["status"] == "healthy":
        manager.route_traffic("vehicle-price-endpoint", {"blue": 80, "green": 20})

# Display all endpoints
print("\n=== Active Endpoints ===")
for ep in manager.list_endpoints():
    print(f"  {ep['name']}: {ep['status']} - Traffic: {ep['traffic']}")

11. Patterns and Best Practices

11.1 Deployment Checklist

Pre-deployment:
✅ Model registered in Model Registry with documented metrics
✅ Scoring script tested locally
✅ Scoring script unit tests passed
✅ Test dataset ready for smoke tests
✅ Endpoint configured with HTTPS and authentication

Deployment:
✅ Initial deployment with 0% traffic
✅ Health checks validated before routing traffic
✅ Canary deployment (10%) before complete migration
✅ Metrics monitored during migration
✅ Rollback plan documented and tested

Post-deployment:
✅ Monitoring enabled (latency, errors, drift)
✅ Alerts configured (email, Teams)
✅ Documentation updated
✅ Previous version kept on standby
✅ Load tests performed

12. Summary and Key Points

12.1 Choosing the Deployment Type

flowchart TD
    Q1{"Need\nimmediate\nresponse?"}
    Q1 -->|Yes| Q2{"Request\nvolume?"}
    Q2 -->|"< 100/s"| MANAGED["Managed Online\nEndpoint\n(Recommended)"]
    Q2 -->|"> 100/s"| AKS_REC["AKS Online\nEndpoint"]
    Q1 -->|No, batch OK| Q3{"Volume?"}
    Q3 -->|"Thousands"| BATCH_EP["Batch Endpoint\n(Cost-effective)"]
    Q3 -->|"Millions"| BATCH_BIG["Batch Pipeline\n+ Parallel Step"]

12.2 Summary Table

13. Glossary

Additional Resources:

Model Catalog and Fine-tuning

Model Catalog in Azure ML Studio provides access to pre-trained models:

• LLMs: GPT, Llama, Mistral, Phi, etc.
• Classification/Regression/Clustering: scikit-learn, XGBoost
• Vision: YOLO, ResNet, EfficientNet
• NLP: BERT, RoBERTa, etc.

Fine-tuning workflow:

Model Catalog → Select a model
    → Fine-tune on custom dataset
    → Register fine-tuned model
    → Deploy as endpoint
    → Test

Steps in Studio:

Studio → Model catalog → select model
→ Fine-tune → configure dataset and parameters
→ Submit fine-tuning job
→ Once complete → Register model
→ Deploy → Create endpoint

Online Endpoints – Real-time Inference

Managed Online Endpoint

A Managed Online Endpoint exposes an ML model as a real-time REST API.

Characteristics:

• Automatic infrastructure management
• Deployment versioning (blue/green)
• Automatic scaling
• Integrated monitoring with Application Insights

Inference request flow:

Client application
    → HTTPS POST /score
        → Endpoint URL
            → Deployment (blue or green)
                → Container with loaded model
                    → Scoring script (run function)
                        → Prediction returned

Scoring Script

The scoring script is the Python code that runs in the deployment container.

Required structure:

# score.py
import json
import numpy as np
import joblib
import os

def init():
    """
    Called only once at container startup.
    Load the model here to avoid reloading on every request.
    """
    global model
    
    # Get the model path from the environment variable
    model_path = os.path.join(os.getenv('AZUREML_MODEL_DIR'), 'model.pkl')
    model = joblib.load(model_path)
    print("Model loaded successfully")

def run(raw_data):
    """
    Called for each inference request.
    Receives: raw_data (JSON string)
    Returns: JSON with predictions
    """
    try:
        # Parse the input data
        data = json.loads(raw_data)
        inputs = np.array(data['data'])
        
        # Make the prediction
        predictions = model.predict(inputs)
        
        # Return the result
        return json.dumps({'predictions': predictions.tolist()})
    
    except Exception as e:
        return json.dumps({'error': str(e)})

Blue/Green Deployment

The Blue/Green deployment allows deploying a new version without interruption.

Concept:

• Blue: current version (in production, 100% traffic)
• Green: new version (tested, then progressive migration)

# Configure traffic distribution
endpoint:
  name: fraud-detection-endpoint
  traffic:
    blue: 80    # 80% to old version
    green: 20   # 20% to new version

Complete migration:

# After validation, switch all traffic
endpoint:
  traffic:
    blue: 0
    green: 100

Via SDK:

from azure.ai.ml.entities import ManagedOnlineEndpoint, ManagedOnlineDeployment, Model

# Create the endpoint
endpoint = ManagedOnlineEndpoint(
    name="fraud-detection-endpoint",
    auth_mode="key"
)
ml_client.online_endpoints.begin_create_or_update(endpoint).result()

# Create the blue deployment
blue_deployment = ManagedOnlineDeployment(
    name="blue",
    endpoint_name="fraud-detection-endpoint",
    model="azureml:fraud-detection-model:1",
    code_configuration=CodeConfiguration(
        code="./src",
        scoring_script="score.py"
    ),
    environment="AzureML-sklearn-1.0:1",
    instance_type="Standard_DS3_v2",
    instance_count=1
)
ml_client.online_deployments.begin_create_or_update(blue_deployment).result()

# Send 100% of traffic to blue
endpoint.traffic = {"blue": 100}
ml_client.online_endpoints.begin_create_or_update(endpoint).result()

CLI Commands for Deployment

Create the endpoint:

az ml online-endpoint create \
  --name fraud-detection-endpoint \
  --auth-mode key \
  --workspace-name my-workspace \
  --resource-group my-rg

Create the deployment:

# deployment.yml file
az ml online-deployment create \
  --file deployment.yml \
  --workspace-name my-workspace \
  --resource-group my-rg

deployment.yml file:

$schema: https://azuremlschemas.azureedge.net/latest/managedOnlineDeployment.schema.json
name: blue
endpoint_name: fraud-detection-endpoint
model: azureml:fraud-detection-model:1
code_configuration:
  code: ./src
  scoring_script: score.py
environment: azureml:AzureML-sklearn-1.0-ubuntu20.04-py38-cpu:1
instance_type: Standard_DS3_v2
instance_count: 1

Update traffic:

az ml online-endpoint update \
  --name fraud-detection-endpoint \
  --traffic "blue=100"

Test the endpoint:

az ml online-endpoint invoke \
  --name fraud-detection-endpoint \
  --request-file test-data.json \
  --workspace-name my-workspace

Test file (test-data.json):

{
  "data": [[1.2, 3.4, 5.6, 7.8], [2.1, 4.3, 6.5, 8.7]]
}

Get access keys:

az ml online-endpoint get-credentials \
  --name fraud-detection-endpoint \
  --workspace-name my-workspace

Deployment on AKS

For deployment on Azure Kubernetes Service:

Prerequisites:

• Azure Container Registry (ACR) attached to the cluster
• Configured and operational AKS cluster

Inference Cluster in Azure ML:

# Attach an existing AKS cluster
az ml compute attach \
  --name production-aks \
  --type Kubernetes \
  --resource-id /subscriptions/{sub}/resourceGroups/{rg}/providers/Microsoft.ContainerService/managedClusters/{cluster}

Deploy on AKS Compute:

# deployment-aks.yml
name: aks-deployment
endpoint_name: fraud-detection-endpoint
model: azureml:fraud-detection-model:1
code_configuration:
  code: ./src
  scoring_script: score.py
environment: azureml:sklearn-env:1
compute: azureml:production-aks
resources:
  requests:
    cpu: "0.5"
    memory: "512Mi"
  limits:
    cpu: "2"
    memory: "2Gi"
instance_count: 3

Key Points Summary

Complete deployment workflow:

1. Train the model (pipeline or job)
2. Register the model (Register Model)
3. Create the scoring script (init + run)
4. Create the environment (dependencies)
5. Create the endpoint
6. Create the deployment (blue)
7. Assign traffic
8. Test (invoke)
9. Monitor (Application Insights)

Review Questions

1. What is the difference between a ManagedOnlineEndpoint and a KubernetesOnlineEndpoint in Azure ML?

A ManagedOnlineEndpoint fully delegates infrastructure management to Azure (provisioning, scaling, load balancing). A KubernetesOnlineEndpoint deploys on an AKS cluster you manage, offering more control over resources, namespaces, and custom configurations (GPU, KEDA, etc.).

2. What does the init() function do in a scoring script, and why not load the model in run()?

init() is called only once at container startup. Loading the model in run() would cause a reload on every request, increasing latency by several seconds. Loading in init() allows the model to stay in memory between requests.

3. You have an endpoint in production with the blue deployment at 100% traffic. You want to deploy a new version with minimal risk. Describe the canary process.

1. Create green deployment (0% traffic). 2. Assign 5% to green. 3. Monitor metrics (latency, error rate) for 30 min. 4. If OK, increase to 20%, then 50%. 5. Migrate 100% to green. 6. Delete blue. On issue: revert to blue=100, green=0.

4. When should you use a Batch Endpoint instead of an Online Endpoint?

The Batch Endpoint is optimal for scoring large volumes of data asynchronously (millions of rows), without real-time latency constraints. The Online Endpoint is for low-latency predictions (< 1s) triggered in real-time by an application.

5. What are the two types of autoscaling available for ManagedOnlineDeployment?

1. target_utilization: scales based on target CPU utilization % (e.g. 70%). 2. Azure Monitor rules: scales based on custom metrics like RequestsPerSecond, with configurable scale-out and scale-in rules.

6. What is mini_batch_size in a BatchDeployment and what is its impact?

mini_batch_size defines the number of rows (or files) processed per call to the worker’s run() function. A high value reduces function calls (lower overhead) but increases memory consumption. It must be calibrated based on available memory on the compute nodes.

7. Why use Managed Identity rather than an API key to call an endpoint in production?

Managed Identity requires no secrets stored in code or environment variables, eliminating the risk of key leakage. The identity is managed by Azure AD and tokens have a short lifetime with automatic rotation.

8. What is KEDA and what advantage does it provide over native Kubernetes HPA for inference?

KEDA (Kubernetes Event-Driven Autoscaling) allows scaling pods based on external metrics (Azure Monitor, Event Hub, Kafka…). Unlike native HPA limited to CPU/memory metrics, KEDA can scale based on HTTP request volume, message queue lag, or any Azure Monitor metric, which is more relevant for ML inference.

9. Explain the difference between output_action: append_row and summary_only in a BatchDeployment.

• append_row: each prediction result is added as a row in the output file (predictions.csv). Ideal for keeping individual predictions. - summary_only: only aggregated metrics (accuracy, F1, etc.) are recorded. Suitable for model evaluation on large datasets.

10. What are the advantages of packaging a model in ONNX format for edge deployment?

1. Portability: runs on any platform (Linux, Windows, ARM). 2. Performance: graph optimizations (operator fusion, quantization) by ONNX Runtime. 3. Reduced size: compressed models suitable for memory-limited devices. 4. Framework independence: no PyTorch or TensorFlow needed on the edge.

11. How does ModelDataCollector contribute to data drift detection?

ModelDataCollector records model inputs (features) and outputs (predictions) in production to Azure Blob Storage. This data is then compared to a reference dataset (training data) by the Azure ML monitoring service, which calculates drift metrics (Wasserstein distance, PSI, etc.) and triggers alerts when drift exceeds a configured threshold.

12. You observe that the P99 latency of your endpoint is 2 seconds while P50 is 80ms. What are the probable causes and solutions?

Probable causes: 1. Large requests (big batches) processed by a single thread. 2. Python GC (Garbage Collection) causing pauses. 3. Container cold starts during scaling. 4. Memory contention on instances.
Solutions: 1. Enable batching to absorb peaks. 2. Increase min_instances to avoid cold start. 3. Use request_timeout_ms and circuit-breaker on the client side. 4. Migrate to a VM with more RAM (Standard_DS4_v2). 5. Profile with Application Insights to identify slow requests.

Search Terms

deploying · models · azure · machine · ml · platforms · deployment · data · science · online · endpoint · model · scoring · script · aks · blue · catalog · endpoints · fine-tuning · green · inference · batch · points · real-time