Azure ML: Pipelines and Experiment Tracking

Course: Introduction to Azure ML Pipelines and Experiment Tracking Level: Intermediate Prerequisites: Basics of Azure ML Studio and Python

Table of Contents

1. Azure ML Pipelines – Core Concepts
2. Pipeline Components
3. Creating a Pipeline with the Designer
4. Creating a Pipeline with the Python SDK v2
5. Submitting and Monitoring Pipelines
6. Handling Failures and Retries
7. Experiment Tracking with MLflow
8. Comparing Runs and Selecting the Best Model
9. Model Registration and Versioning
10. Advanced Patterns – Hyperparameter Tuning
11. MLOps – CI/CD Integration
12. Summary and Key Takeaways
13. Glossary

1. Azure ML Pipelines – Core Concepts

1.1 Why ML Pipelines?

Without a pipeline, an ML workflow looks like this: you manually execute each step, in the right order, on the right machine. If a step fails at 3 AM, nobody knows until the next morning. If you want to re-run the same workflow next month, you have to start from scratch.

An ML Pipeline solves these problems:

flowchart LR
    subgraph "Without Pipeline ❌"
        M1["Step 1\n(Manual)"] --> M2["Step 2\n(Manual)"]
        M2 --> M3["Step 3\n(Manual)"]
        M3 --> M4["Result\n(??)"]
    end
    
    subgraph "With Pipeline ✅"
        P1["Step 1\n(Automated)"] --> P2["Step 2\n(Automated)"]
        P2 --> P3["Step 3\n(Automated)"]
        P3 --> P4["Result\n(Tracked, Versioned)"]
    end

Benefits of Azure ML Pipelines:

1.2 Two Ways to Create a Pipeline

flowchart TD
    PIPELINE["Create an\nAzure ML Pipeline"] --> DESIGNER["Azure ML Designer\n(Drag-and-drop)\n\n✅ Visual, intuitive\n✅ No code required\n✅ For beginners\n⚠️ Less flexible"]
    PIPELINE --> SDK["Azure ML SDK v2\n(Python code)\n\n✅ Very flexible\n✅ Versioned in Git\n✅ Native CI/CD\n⚠️ Requires Python"]

2. Pipeline Components

2.1 Pipeline Structure

flowchart LR
    subgraph "Azure ML Pipeline"
        direction LR
        
        INPUT["🔵 Inputs\n(Datasets, Parameters)"]
        
        STEP1["⬜ Step 1\n(Command Step)\nPython or Shell\nScript"]
        
        STEP2["⬜ Step 2\n(Parallel Step)\nParallelized\nProcessing"]
        
        STEP3["⬜ Step 3\n(Command Step)\nEvaluation"]
        
        OUTPUT["🟢 Outputs\n(Model, Metrics,\nDerived Datasets)"]
        
        INPUT --> STEP1
        STEP1 --> STEP2
        STEP2 --> STEP3
        STEP3 --> OUTPUT
    end
    
    subgraph "Infrastructure"
        COMPUTE["⚙️ Compute\n(CPU/GPU Cluster)"]
        ENV["📦 Environment\n(Python Dependencies)"]
        DATASTORE["💾 Datastore\n(Intermediate Storage)"]
    end
    
    STEP1 -.-> COMPUTE
    STEP1 -.-> ENV
    STEP1 -.-> DATASTORE

2.2 Step Types in a Pipeline

2.3 DAG (Directed Acyclic Graph)

Azure ML automatically builds a DAG from the dependencies between steps:

If Step B uses the output of Step A → A must run before B
If Steps B and C both use the output of A → B and C can run in parallel

# Visualize pipeline dependencies
def visualize_pipeline_dag(pipeline_name: str) -> dict:
    """
    Returns the DAG structure of a pipeline.
    
    Returns:
        Dict representing nodes and their dependencies
    """
    # Example DAG structure
    dag = {
        "nodes": [
            {"id": "data_prep", "type": "command", "label": "Data Preparation"},
            {"id": "feature_eng", "type": "command", "label": "Feature Engineering",
             "depends_on": ["data_prep"]},
            {"id": "train_model", "type": "command", "label": "Training",
             "depends_on": ["feature_eng"]},
            {"id": "eval_model", "type": "command", "label": "Evaluation",
             "depends_on": ["train_model"]},
            {"id": "register_model", "type": "command", "label": "Registration",
             "depends_on": ["eval_model"]}
        ]
    }
    
    # Compute execution levels (steps that can run in parallel)
    levels = {}
    for node in dag["nodes"]:
        deps = node.get("depends_on", [])
        if not deps:
            level = 0
        else:
            level = max(levels.get(dep, 0) for dep in deps) + 1
        levels[node["id"]] = level
    
    dag["execution_levels"] = levels
    return dag

3. Creating a Pipeline with the Designer

3.1 Automobile Price Prediction Pipeline

Components used in the Designer:

flowchart TD
    DS["📊 Automobile Price Dataset\n(Azure ML built-in dataset)"]
    DS --> SC["Select Columns in Dataset\n(Exclude normalized-losses)"]
    SC --> CMD["Clean Missing Data\n(Remove rows with null)"]
    CMD --> SPLIT["Split Data\n(Fraction: 0.7 / 0.3)"]
    SPLIT -->|"70% Train"| LR["Linear Regression\n(Algorithm)"]
    LR --> TM["Train Model\n(Target variable: price)"]
    SPLIT -->|"30% Test"| SM["Score Model\n(Test set predictions)"]
    TM --> SM
    SM --> EM["Evaluate Model\n(RMSE, MAE, R²)"]

3.2 Component Configuration

Component: Select Columns in Dataset
→ Mode: Exclude
→ Excluded columns: normalized-losses
→ Reason: Too many missing values

Component: Clean Missing Data
→ Cleaning Mode: Remove entire row
→ Apply To: All columns
→ Reason: Remove incomplete rows rather than imputing

Component: Split Data
→ Fraction: 0.7 (70% train)
→ Randomized split: True
→ Random seed: 42
→ Stratified split: False (regression)

Component: Linear Regression
→ Default parameters
→ Online gradient descent: False

Component: Train Model
→ Label column: price
→ This field specifies the target variable

Component: Evaluate Model
→ Computed metrics: RMSE, MAE, R², Relative Squared Error

4. Creating a Pipeline with the Python SDK v2

4.1 Pipeline Project Structure

pipeline_project/
├── pipeline.py              # Main pipeline script
├── components/
│   ├── data_prep/
│   │   ├── data_prep.py     # Python data preparation script
│   │   └── component.yml    # Component definition
│   ├── training/
│   │   ├── train.py
│   │   └── component.yml
│   └── evaluation/
│       ├── evaluate.py
│       └── component.yml
├── environments/
│   └── ml_env.yml           # Conda dependencies
└── config/
    └── workspace_config.json

4.2 YAML Component Definitions

# components/data_prep/component.yml
$schema: https://azuremlschemas.azureedge.net/latest/commandComponent.schema.json

name: data_preparation
version: "1.0.0"
display_name: Data Preparation
description: Cleans and prepares data for ML training

inputs:
  raw_data:
    type: uri_file
    description: Raw CSV data file
  test_ratio:
    type: number
    default: 0.2
    description: Proportion of data for testing (0-1)

outputs:
  train_data:
    type: uri_folder
    description: Cleaned training data
  test_data:
    type: uri_folder
    description: Cleaned test data
  preprocessing_stats:
    type: uri_file
    description: Preprocessing statistics (JSON)

command: >-
  python data_prep.py
  --raw_data ${{inputs.raw_data}}
  --test_ratio ${{inputs.test_ratio}}
  --train_data ${{outputs.train_data}}
  --test_data ${{outputs.test_data}}
  --stats ${{outputs.preprocessing_stats}}

environment: azureml:AzureML-sklearn-1.0-ubuntu20.04-py38-cpu@latest

# components/training/component.yml
$schema: https://azuremlschemas.azureedge.net/latest/commandComponent.schema.json

name: model_training
version: "1.0.0"
display_name: Model Training
description: Trains a regression/classification model

inputs:
  train_data:
    type: uri_folder
  algorithm:
    type: string
    default: "gradient_boosting"
    enum: ["linear_regression", "gradient_boosting", "random_forest"]
  learning_rate:
    type: number
    default: 0.1
  n_estimators:
    type: integer
    default: 100
  max_depth:
    type: integer
    default: 3

outputs:
  model:
    type: uri_folder
  training_metrics:
    type: uri_file

command: >-
  python train.py
  --train_data ${{inputs.train_data}}
  --algorithm ${{inputs.algorithm}}
  --learning_rate ${{inputs.learning_rate}}
  --n_estimators ${{inputs.n_estimators}}
  --max_depth ${{inputs.max_depth}}
  --model ${{outputs.model}}
  --metrics ${{outputs.training_metrics}}

environment: azureml:AzureML-sklearn-1.0-ubuntu20.04-py38-cpu@latest

4.3 Python Component Scripts

# components/data_prep/data_prep.py
import argparse
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
import json
import os

def parse_arguments():
    parser = argparse.ArgumentParser(description="ML Data Preparation")
    parser.add_argument("--raw_data", type=str, help="Path to raw data")
    parser.add_argument("--test_ratio", type=float, default=0.2)
    parser.add_argument("--train_data", type=str)
    parser.add_argument("--test_data", type=str)
    parser.add_argument("--stats", type=str)
    return parser.parse_args()

def clean_data(df: pd.DataFrame) -> tuple[pd.DataFrame, dict]:
    """Cleans the dataframe and returns statistics."""
    stats = {
        "initial_rows": len(df),
        "initial_columns": len(df.columns),
        "initial_missing_values": df.isnull().sum().sum()
    }
    
    # Remove duplicates
    df = df.drop_duplicates()
    
    # Remove rows with too many missing values (> 50%)
    nan_threshold = len(df.columns) * 0.5
    df = df.dropna(thresh=nan_threshold)
    
    # Encode categorical variables
    encoders = {}
    for col in df.select_dtypes(include=['object']).columns:
        if col != df.columns[-1]:  # Not the target variable
            le = LabelEncoder()
            df[col] = le.fit_transform(df[col].astype(str))
            encoders[col] = list(le.classes_)
    
    stats.update({
        "rows_after_cleaning": len(df),
        "missing_values_after": df.isnull().sum().sum(),
        "encoded_columns": list(encoders.keys()),
        "rows_removed": stats["initial_rows"] - len(df)
    })
    
    return df, stats

def main():
    args = parse_arguments()
    
    print(f"Loading data: {args.raw_data}")
    df = pd.read_csv(args.raw_data)
    
    print(f"Cleaning ({len(df)} rows)...")
    df_clean, stats = clean_data(df)
    
    print(f"Splitting train/test (test ratio: {args.test_ratio})...")
    # Assume the last column is the target variable
    X = df_clean.iloc[:, :-1]
    y = df_clean.iloc[:, -1]
    
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, 
        test_size=args.test_ratio, 
        random_state=42
    )
    
    # Save outputs
    os.makedirs(args.train_data, exist_ok=True)
    os.makedirs(args.test_data, exist_ok=True)
    
    train_df = pd.concat([X_train, y_train], axis=1)
    test_df = pd.concat([X_test, y_test], axis=1)
    
    train_df.to_csv(os.path.join(args.train_data, "train.csv"), index=False)
    test_df.to_csv(os.path.join(args.test_data, "test.csv"), index=False)
    
    # Statistics
    stats.update({
        "train_rows": len(train_df),
        "test_rows": len(test_df),
        "test_ratio": args.test_ratio
    })
    
    with open(args.stats, "w") as f:
        json.dump(stats, f, indent=2)
    
    print(f"✅ Preparation complete:")
    print(f"   Train: {len(train_df)} rows")
    print(f"   Test: {len(test_df)} rows")
    print(json.dumps(stats, indent=2))

if __name__ == "__main__":
    main()

# components/training/train.py
import argparse
import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor, RandomForestRegressor
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error, r2_score, mean_absolute_error
import mlflow
import mlflow.sklearn
import joblib
import json
import os

def parse_arguments():
    parser = argparse.ArgumentParser()
    parser.add_argument("--train_data", type=str)
    parser.add_argument("--algorithm", type=str, default="gradient_boosting")
    parser.add_argument("--learning_rate", type=float, default=0.1)
    parser.add_argument("--n_estimators", type=int, default=100)
    parser.add_argument("--max_depth", type=int, default=3)
    parser.add_argument("--model", type=str)
    parser.add_argument("--metrics", type=str)
    return parser.parse_args()

def create_model(algorithm: str, params: dict):
    """Instantiates the model based on the chosen algorithm."""
    algorithms = {
        "linear_regression": LinearRegression(),
        "gradient_boosting": GradientBoostingRegressor(
            learning_rate=params.get("learning_rate", 0.1),
            n_estimators=params.get("n_estimators", 100),
            max_depth=params.get("max_depth", 3),
            random_state=42
        ),
        "random_forest": RandomForestRegressor(
            n_estimators=params.get("n_estimators", 100),
            max_depth=params.get("max_depth", None),
            random_state=42
        )
    }
    
    if algorithm not in algorithms:
        raise ValueError(f"Algorithm '{algorithm}' not supported")
    
    return algorithms[algorithm]

def main():
    args = parse_arguments()
    
    # Load data
    print(f"Loading training data...")
    df = pd.read_csv(os.path.join(args.train_data, "train.csv"))
    X = df.iloc[:, :-1]
    y = df.iloc[:, -1]
    
    print(f"Data: {X.shape[0]} rows, {X.shape[1]} features")
    
    # Configure MLflow
    mlflow.set_experiment("ml-pipeline-training")
    
    params = {
        "learning_rate": args.learning_rate,
        "n_estimators": args.n_estimators,
        "max_depth": args.max_depth
    }
    
    with mlflow.start_run():
        # Train the model
        print(f"Training with {args.algorithm}...")
        model = create_model(args.algorithm, params)
        model.fit(X, y)
        
        # Training metrics
        y_pred = model.predict(X)
        train_rmse = mean_squared_error(y, y_pred, squared=False)
        train_mae = mean_absolute_error(y, y_pred)
        train_r2 = r2_score(y, y_pred)
        
        # Log with MLflow
        mlflow.log_param("algorithm", args.algorithm)
        mlflow.log_params(params)
        mlflow.log_metric("train_rmse", train_rmse)
        mlflow.log_metric("train_mae", train_mae)
        mlflow.log_metric("train_r2", train_r2)
        
        # Log the model with MLflow
        mlflow.sklearn.log_model(model, "model")
        
        print(f"\n=== Training Metrics ===")
        print(f"RMSE: {train_rmse:.4f}")
        print(f"MAE: {train_mae:.4f}")
        print(f"R²: {train_r2:.4f}")
    
    # Save model
    os.makedirs(args.model, exist_ok=True)
    joblib.dump(model, os.path.join(args.model, "model.joblib"))
    
    # Save metrics
    metrics = {
        "algorithm": args.algorithm,
        "params": params,
        "train_rmse": train_rmse,
        "train_mae": train_mae,
        "train_r2": train_r2,
        "n_features": X.shape[1],
        "n_samples_train": X.shape[0]
    }
    
    with open(args.metrics, "w") as f:
        json.dump(metrics, f, indent=2)
    
    print(f"\n✅ Training complete!")

if __name__ == "__main__":
    main()

4.4 Assembling and Submitting the Pipeline

# pipeline.py - Main pipeline script
from azure.ai.ml import MLClient, dsl, Input, Output, load_component
from azure.ai.ml.constants import AssetTypes
from azure.identity import DefaultAzureCredential
import os

# Connect to the workspace
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"]
)

# Load components from YAML files
data_prep_component = load_component(source="components/data_prep/component.yml")
training_component = load_component(source="components/training/component.yml")
evaluation_component = load_component(source="components/evaluation/component.yml")

# Define the pipeline
@dsl.pipeline(
    name="complete-ml-pipeline",
    display_name="Complete ML Pipeline (Prep → Train → Eval)",
    description="End-to-end pipeline for ML training",
    compute="cpu-cluster-4cores",
    tags={"project": "automobile-price", "version": "1.0"}
)
def complete_ml_pipeline(
    raw_data: Input(type=AssetTypes.URI_FILE),
    test_ratio: float = 0.2,
    algorithm: str = "gradient_boosting",
    learning_rate: float = 0.1,
    n_estimators: int = 100
):
    """
    Complete ML pipeline with 3 steps:
    1. Data preparation
    2. Model training
    3. Evaluation and report
    """
    
    # Step 1: Preparation
    prep_step = data_prep_component(
        raw_data=raw_data,
        test_ratio=test_ratio
    )
    prep_step.display_name = "Data Preparation"
    prep_step.compute = "cpu-cluster-2cores"  # Light step → small cluster
    
    # Step 2: Training (depends on preparation)
    train_step = training_component(
        train_data=prep_step.outputs.train_data,
        algorithm=algorithm,
        learning_rate=learning_rate,
        n_estimators=n_estimators
    )
    train_step.display_name = "Model Training"
    train_step.compute = "gpu-cluster-v100"  # Heavy step → GPU if available
    
    # Step 3: Evaluation (depends on training)
    eval_step = evaluation_component(
        test_data=prep_step.outputs.test_data,
        model=train_step.outputs.model
    )
    eval_step.display_name = "Model Evaluation"
    eval_step.compute = "cpu-cluster-2cores"
    
    # Final pipeline outputs
    return {
        "final_model": train_step.outputs.model,
        "eval_report": eval_step.outputs.evaluation_report
    }

# Create and submit the pipeline
pipeline_job = complete_ml_pipeline(
    raw_data=Input(
        path="azureml:automobile-price:1",
        type=AssetTypes.URI_FILE
    ),
    test_ratio=0.2,
    algorithm="gradient_boosting",
    learning_rate=0.05,
    n_estimators=200
)

# Global settings
pipeline_job.settings.default_compute = "cpu-cluster-4cores"
pipeline_job.settings.default_datastore = "workspaceblobstore"

# Submit
print("Submitting pipeline...")
returned_job = ml_client.jobs.create_or_update(pipeline_job)

print(f"✅ Pipeline submitted!")
print(f"   Name: {returned_job.name}")
print(f"   Status: {returned_job.status}")
print(f"   Studio URL: {returned_job.studio_url}")

# Wait for completion (optional)
ml_client.jobs.stream(returned_job.name)

5. Submitting and Monitoring Pipelines

5.1 Azure ML Job Lifecycle

stateDiagram-v2
    [*] --> Queued: Submission
    Queued --> Preparing: Resources available
    Preparing --> Running: Environment ready
    Running --> Completed: Success
    Running --> Failed: Error
    Running --> Canceled: Manual cancellation
    
    Completed --> [*]
    Failed --> [*]
    Canceled --> [*]

5.2 Monitoring and Logs

# Comprehensive monitoring of an Azure ML job
import time
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential

def monitor_job(job_name: str, interval_sec: int = 15) -> dict:
    """
    Monitors an Azure ML job until completion.
    
    Args:
        job_name: Name of the job to monitor
        interval_sec: Polling interval in seconds
    
    Returns:
        Final job information
    """
    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"]
    )
    
    terminal_states = ["Completed", "Failed", "Canceled", "NotStarted"]
    last_status = None
    start_time = time.time()
    
    print(f"Monitoring job: {job_name}")
    print("-" * 50)
    
    while True:
        job = ml_client.jobs.get(job_name)
        status = job.status
        
        if status != last_status:
            elapsed = (time.time() - start_time) / 60
            icon = {
                "Queued": "⏳",
                "Preparing": "🔧",
                "Running": "▶️",
                "Completed": "✅",
                "Failed": "❌",
                "Canceled": "⏹️"
            }.get(status, "❓")
            
            print(f"{icon} [{elapsed:.1f} min] Status → {status}")
            last_status = status
        
        if status in terminal_states:
            break
        
        time.sleep(interval_sec)
    
    # Retrieve final metrics
    final_metrics = {}
    try:
        for run in ml_client.jobs.list(parent_job_name=job_name):
            if run.display_name and "train" in run.display_name.lower():
                final_metrics = {
                    k: v for k, v in run.properties.items()
                    if k.startswith("train_")
                }
    except Exception:
        pass
    
    total_elapsed = (time.time() - start_time) / 60
    
    return {
        "job_name": job_name,
        "final_status": status,
        "duration_minutes": round(total_elapsed, 1),
        "studio_url": job.studio_url,
        "metrics": final_metrics
    }

def analyze_job_failure(job_name: str) -> dict:
    """
    Analyzes the causes of a failed job.
    
    Returns:
        Failure diagnostic
    """
    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"]
    )
    
    job = ml_client.jobs.get(job_name)
    
    if job.status != "Failed":
        return {"message": f"Job did not fail (status: {job.status})"}
    
    # Retrieve failed children
    failed_steps = []
    for child in ml_client.jobs.list(parent_job_name=job_name):
        if child.status == "Failed":
            failed_steps.append({
                "step": child.display_name,
                "error_message": child.error.message if child.error else "Unknown",
                "error_code": child.error.code if child.error else "N/A"
            })
    
    return {
        "job_name": job_name,
        "status": "Failed",
        "failed_steps": failed_steps,
        "recommended_actions": [
            "1. Check logs for the failed step",
            "2. Validate that input files exist",
            "3. Check Python dependencies in the environment",
            "4. Increase compute resources if memory error",
            "5. Fix the script and rerun only the failed step"
        ]
    }

# Usage
import json

# Monitor a job
monitoring_result = monitor_job("complete-ml-pipeline-xxx")
print(json.dumps(monitoring_result, indent=2))

# If failed, analyze
if monitoring_result["final_status"] == "Failed":
    diagnostic = analyze_job_failure("complete-ml-pipeline-xxx")
    print("\n=== Failure Diagnostic ===")
    print(json.dumps(diagnostic, indent=2))

6. Handling Failures and Retries

6.1 Common Errors and Solutions

6.2 Retry and Error Handling in the SDK

# Robust submission with retry
import time
import functools
from azure.core.exceptions import ServiceRequestError, HttpResponseError

def with_retry(max_attempts: int = 3, base_delay: float = 30.0):
    """Decorator for automatic retry on Azure ML calls."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                
                except HttpResponseError as e:
                    if e.status_code == 429:  # Rate limit
                        delay = base_delay * (2 ** (attempt - 1))
                        print(f"  ⏳ Azure ML rate limit. Waiting {delay}s...")
                        time.sleep(delay)
                    elif e.status_code >= 500:  # Server error
                        if attempt < max_attempts:
                            print(f"  ⚠️ Server error {e.status_code}. Retry {attempt}/{max_attempts}...")
                            time.sleep(base_delay)
                        else:
                            raise
                    else:
                        raise
                
                except ServiceRequestError as e:
                    if attempt < max_attempts:
                        print(f"  ⚠️ Network error: {e}. Retry {attempt}/{max_attempts}...")
                        time.sleep(base_delay)
                    else:
                        raise
            
            raise Exception(f"Failed after {max_attempts} attempts")
        
        return wrapper
    return decorator

@with_retry(max_attempts=3)
def submit_pipeline_safely(ml_client, pipeline_job):
    """Submits a pipeline with error handling and retry."""
    returned = ml_client.jobs.create_or_update(pipeline_job)
    print(f"✅ Pipeline submitted: {returned.name}")
    return returned

# Rerun only failed steps
def rerun_failed_steps(ml_client, job_name: str) -> str:
    """
    Reruns only the failed steps of a pipeline.
    
    Note: Azure ML does not directly support partial retry via the SDK —
    this function illustrates the conceptual approach.
    In practice, use the 'Resume' button in Studio.
    """
    print(f"Analyzing failed job: {job_name}")
    
    job = ml_client.jobs.get(job_name)
    
    # Clone the job with the same parameters
    # (Azure ML Studio allows direct 'Resubmit')
    print("Recommendation: Use 'Resubmit' in Azure ML Studio")
    print("This reruns only the incomplete steps")
    
    return job.id

7. Experiment Tracking with MLflow

7.1 MLflow Core Concepts

flowchart TD
    EXPERIMENT["🗂️ Experiment\n(Top-level container)\n\nEx: 'automobile-price-prediction'"] --> RUN1["▶️ Run 1\nLR=0.01, n=100\nRMSE=1500, R²=0.82"]
    EXPERIMENT --> RUN2["▶️ Run 2\nLR=0.05, n=200\nRMSE=1200, R²=0.89"]
    EXPERIMENT --> RUN3["▶️ Run 3\nLR=0.1, n=300\nRMSE=1100, R²=0.91"]
    
    RUN2 --> PARAMS["📊 Parameters\n(Hyperparameters)"]
    RUN2 --> METRICS["📈 Metrics\n(RMSE, R², MAE...)"]
    RUN2 --> ARTIFACTS["📁 Artifacts\n(Model, Charts,\nDatasets)"]
    RUN2 --> TAGS["🏷️ Tags\n(Metadata)"]

Terminology:

• Experiment = Container for all runs of a project
• Run = A single training execution
• Parameters = Hyperparameters (learning rate, n_estimators…)
• Metrics = Results (RMSE, accuracy, F1…)
• Artifacts = Saved files (model.pkl, charts…)

7.2 Complete MLflow Logging

# Complete MLflow logging in a training script
import mlflow
import mlflow.sklearn
import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.metrics import mean_squared_error, r2_score, mean_absolute_error
from sklearn.model_selection import cross_val_score
import matplotlib.pyplot as plt
import joblib
import os

def train_with_mlflow_tracking(
    X_train: pd.DataFrame,
    y_train: pd.Series,
    X_test: pd.DataFrame,
    y_test: pd.Series,
    params: dict,
    experiment_name: str = "automobile-price-prediction"
) -> dict:
    """
    Trains a model with complete MLflow tracking.
    
    Args:
        X_train, y_train: Training data
        X_test, y_test: Test data
        params: Model hyperparameters
        experiment_name: MLflow experiment name
    
    Returns:
        Dict with metrics and run ID
    """
    # Configure the experiment
    mlflow.set_experiment(experiment_name)
    
    with mlflow.start_run(run_name=f"GB-lr{params['learning_rate']}-n{params['n_estimators']}") as run:
        
        # === 1. PARAMETERS ===
        mlflow.log_params({
            "algorithm": "GradientBoostingRegressor",
            "learning_rate": params["learning_rate"],
            "n_estimators": params["n_estimators"],
            "max_depth": params.get("max_depth", 3),
            "n_features": X_train.shape[1],
            "n_samples_train": X_train.shape[0],
            "n_samples_test": X_test.shape[0]
        })
        
        # === 2. TRAINING ===
        model = GradientBoostingRegressor(
            learning_rate=params["learning_rate"],
            n_estimators=params["n_estimators"],
            max_depth=params.get("max_depth", 3),
            random_state=42
        )
        model.fit(X_train, y_train)
        
        # === 3. METRICS - Training ===
        y_train_pred = model.predict(X_train)
        train_rmse = mean_squared_error(y_train, y_train_pred, squared=False)
        train_mae = mean_absolute_error(y_train, y_train_pred)
        train_r2 = r2_score(y_train, y_train_pred)
        
        mlflow.log_metrics({
            "train_rmse": train_rmse,
            "train_mae": train_mae,
            "train_r2": train_r2
        })
        
        # === 4. METRICS - Test ===
        y_test_pred = model.predict(X_test)
        test_rmse = mean_squared_error(y_test, y_test_pred, squared=False)
        test_mae = mean_absolute_error(y_test, y_test_pred)
        test_r2 = r2_score(y_test, y_test_pred)
        
        mlflow.log_metrics({
            "test_rmse": test_rmse,
            "test_mae": test_mae,
            "test_r2": test_r2
        })
        
        # === 5. CROSS-VALIDATION ===
        cv_scores = cross_val_score(
            model, X_train, y_train,
            cv=5, scoring="r2"
        )
        mlflow.log_metric("cv_r2_mean", cv_scores.mean())
        mlflow.log_metric("cv_r2_std", cv_scores.std())
        
        # === 6. CHARTS ===
        fig, axes = plt.subplots(1, 2, figsize=(12, 5))
        
        # Actual vs Predicted
        axes[0].scatter(y_test, y_test_pred, alpha=0.5)
        axes[0].plot([y_test.min(), y_test.max()], 
                     [y_test.min(), y_test.max()], 'r--', lw=2)
        axes[0].set_xlabel("Actual values")
        axes[0].set_ylabel("Predicted values")
        axes[0].set_title(f"Actual vs Predicted (R²={test_r2:.3f})")
        
        # Feature Importance
        if hasattr(model, 'feature_importances_'):
            importances = pd.Series(
                model.feature_importances_,
                index=X_train.columns
            ).sort_values(ascending=True).tail(10)
            
            importances.plot(kind='barh', ax=axes[1])
            axes[1].set_title("Top 10 Important Features")
        
        plt.tight_layout()
        
        # Save and log the chart
        fig.savefig("model_performance.png", dpi=150, bbox_inches='tight')
        mlflow.log_artifact("model_performance.png")
        plt.close()
        
        # === 7. TAGS ===
        mlflow.set_tags({
            "dataset": "automobile-price",
            "developer": os.environ.get("USER", "unknown"),
            "objective": "regression",
            "prod_ready": str(test_r2 > 0.85)  # Quality criterion
        })
        
        # === 8. MODEL ===
        # Log the model with MLflow (with signature)
        from mlflow.models import infer_signature
        signature = infer_signature(X_train, y_train_pred)
        
        mlflow.sklearn.log_model(
            model,
            artifact_path="model",
            signature=signature,
            registered_model_name="automobile-price-model"
        )
        
        # Also save as joblib for direct use
        joblib.dump(model, "model_backup.joblib")
        mlflow.log_artifact("model_backup.joblib", artifact_path="model_backup")
        
        print(f"\n=== Run {run.info.run_id[:8]}... ===")
        print(f"Train R²: {train_r2:.4f}")
        print(f"Test R²: {test_r2:.4f}")
        print(f"Test RMSE: {test_rmse:.2f}")
        print(f"CV R² mean: {cv_scores.mean():.4f} (±{cv_scores.std():.4f})")
        
        return {
            "run_id": run.info.run_id,
            "test_rmse": test_rmse,
            "test_mae": test_mae,
            "test_r2": test_r2,
            "cv_r2_mean": cv_scores.mean()
        }

7.3 MLflow Autologging

# MLflow Autologging - automatic logging without explicit code
import mlflow

# Enable autologging for sklearn
mlflow.sklearn.autolog(
    log_input_examples=True,    # Log input examples
    log_model_signatures=True,  # Log model signature
    log_models=True,            # Log the model itself
    disable=False
)

with mlflow.start_run(run_name="autolog-run"):
    # Everything will be logged automatically!
    model = GradientBoostingRegressor(learning_rate=0.05, n_estimators=200)
    model.fit(X_train, y_train)
    
    # Only additional metrics require explicit logging
    y_pred = model.predict(X_test)
    mlflow.log_metric("test_r2_custom", r2_score(y_test, y_pred))
    
    print("✅ Autologging: parameters and metrics logged automatically")

8. Comparing Runs and Selecting the Best Model

8.1 Programmatic Comparison with MLflow

# Compare MLflow runs and select the best
import mlflow
from mlflow.tracking import MlflowClient
import pandas as pd

def compare_experiment_runs(experiment_name: str, 
                              key_metric: str = "test_r2",
                              top_n: int = 5) -> pd.DataFrame:
    """
    Retrieves and compares runs from an MLflow experiment.
    
    Args:
        experiment_name: Experiment name
        key_metric: Metric for sorting runs
        top_n: Number of runs to return
    
    Returns:
        DataFrame with the top N runs
    """
    client = MlflowClient()
    
    # Retrieve the experiment
    experiment = client.get_experiment_by_name(experiment_name)
    if not experiment:
        raise ValueError(f"Experiment '{experiment_name}' not found")
    
    # Search all completed runs
    runs = client.search_runs(
        experiment_ids=[experiment.experiment_id],
        filter_string="status = 'FINISHED'",
        order_by=[f"metrics.{key_metric} DESC"],
        max_results=top_n
    )
    
    # Build the comparison table
    comparison = []
    for run in runs:
        row = {
            "run_id": run.info.run_id[:8] + "...",
            "run_name": run.data.tags.get("mlflow.runName", "N/A"),
            "date": run.info.start_time,
            "duration_min": round((run.info.end_time - run.info.start_time) / 60000, 1)
        }
        
        # Add key parameters
        key_params = ["algorithm", "learning_rate", "n_estimators", "max_depth"]
        for param in key_params:
            row[f"param_{param}"] = run.data.params.get(param, "N/A")
        
        # Add metrics
        key_metrics = ["train_rmse", "test_rmse", "train_r2", "test_r2", "cv_r2_mean"]
        for metric in key_metrics:
            val = run.data.metrics.get(metric)
            row[metric] = round(val, 4) if val else "N/A"
        
        comparison.append(row)
    
    df = pd.DataFrame(comparison)
    
    print(f"=== Top {len(df)} runs (sorted by {key_metric}) ===")
    print(df.to_string(index=False))
    
    return df

def select_best_run(experiment_name: str,
                    criterion: str = "test_r2",
                    min_threshold: float = 0.85) -> dict | None:
    """
    Selects the best run based on a criterion and minimum threshold.
    
    Args:
        experiment_name: Experiment name
        criterion: Metric to maximize
        min_threshold: Minimum acceptable value
    
    Returns:
        Best run information or None if no run meets the threshold
    """
    client = MlflowClient()
    experiment = client.get_experiment_by_name(experiment_name)
    
    runs = client.search_runs(
        experiment_ids=[experiment.experiment_id],
        filter_string=f"status = 'FINISHED' AND metrics.{criterion} >= {min_threshold}",
        order_by=[f"metrics.{criterion} DESC"],
        max_results=1
    )
    
    if not runs:
        print(f"❌ No run with {criterion} >= {min_threshold}")
        return None
    
    best = runs[0]
    
    result = {
        "run_id": best.info.run_id,
        "score": best.data.metrics.get(criterion),
        "parameters": best.data.params,
        "metrics": best.data.metrics,
        "artifact_uri": best.info.artifact_uri,
        "model_uri": f"runs:/{best.info.run_id}/model"
    }
    
    print(f"✅ Best run selected:")
    print(f"   Run ID: {result['run_id'][:8]}...")
    print(f"   {criterion}: {result['score']:.4f}")
    
    return result

# Usage
comparison = compare_experiment_runs(
    experiment_name="automobile-price-prediction",
    key_metric="test_r2",
    top_n=5
)

best = select_best_run(
    experiment_name="automobile-price-prediction",
    criterion="test_r2",
    min_threshold=0.85
)

9. Model Registration and Versioning

9.1 Azure ML Model Registry

# Model registration and version management
from azure.ai.ml.entities import Model
from azure.ai.ml.constants import AssetTypes
import mlflow

def register_best_model_from_mlflow(
    run_id: str,
    model_name: str,
    tags: dict = None
) -> Model:
    """
    Registers the best MLflow model in the Azure ML registry.
    
    Args:
        run_id: MLflow run ID containing the model
        model_name: Name for the Azure ML registry
        tags: Optional tags for the model
    
    Returns:
        Registered model
    """
    # Path to the model in MLflow
    model_uri = f"runs:/{run_id}/model"
    
    # Register in MLflow Model Registry
    mlflow.register_model(
        model_uri=model_uri,
        name=model_name
    )
    
    # Register in Azure ML Model Registry
    model = Model(
        path=f"azureml://jobs/{run_id}/outputs/artifacts/paths/model",
        name=model_name,
        type="mlflow_model",
        description=f"Best automobile regression model (Run: {run_id[:8]})",
        tags=tags or {
            "type": "regression",
            "dataset": "automobile-price",
            "framework": "sklearn"
        }
    )
    
    registered_model = ml_client.models.create_or_update(model)
    
    print(f"✅ Model registered in Azure ML:")
    print(f"   Name: {registered_model.name}")
    print(f"   Version: {registered_model.version}")
    print(f"   URI: {registered_model.id}")
    
    return registered_model

def list_model_versions(model_name: str) -> pd.DataFrame:
    """Lists all versions of a registered model."""
    versions = list(ml_client.models.list(name=model_name))
    
    data = [{
        "version": v.version,
        "creation_date": v.creation_context.created_at if v.creation_context else "N/A",
        "tags": v.tags,
        "description": v.description[:50] + "..." if v.description and len(v.description) > 50 else v.description
    } for v in versions]
    
    return pd.DataFrame(data)

# Complete lifecycle
if best:
    # Register the best model
    prod_model = register_best_model_from_mlflow(
        run_id=best["run_id"],
        model_name="automobile-price-model",
        tags={
            "test_r2": str(round(best["score"], 4)),
            "prod_ready": "true",
            "author": os.environ.get("USER", "unknown")
        }
    )
    
    print("\n=== Available Versions ===")
    versions = list_model_versions("automobile-price-model")
    print(versions.to_string(index=False))

10. Advanced Patterns – Hyperparameter Tuning

10.1 Sweep Job – Hyperparameter Search

# Hyperparameter tuning with Azure ML Sweep
from azure.ai.ml.sweep import (
    Choice, Uniform, LogUniform, 
    BanditPolicy, TruncationSelectionPolicy
)
from azure.ai.ml.entities import SweepJob

# Configure the base job
command_job = command(
    code="./training/",
    command="python train.py --train_data ${{inputs.train_data}} --learning_rate ${{inputs.learning_rate}} --n_estimators ${{inputs.n_estimators}} --max_depth ${{inputs.max_depth}}",
    environment="AzureML-sklearn-1.0-ubuntu20.04-py38-cpu@latest",
    inputs={
        "train_data": Input(type="uri_folder"),
        "learning_rate": 0.1,
        "n_estimators": 100,
        "max_depth": 3
    },
    compute="cpu-cluster-4cores"
)

# Configure the Sweep
sweep_job = command_job.sweep(
    # Hyperparameter search space
    sampling_algorithm="bayesian",  # bayesian, grid, random
    
    primary_metric="test_r2",
    goal="maximize",
    
    # Early termination criterion
    early_termination=BanditPolicy(
        slack_factor=0.15,     # Stop if R² < best * (1 - 0.15)
        evaluation_interval=1  # Check at each round
    ),
    
    # Limits
    limits={
        "max_total_trials": 30,    # Max 30 combinations
        "max_concurrent_trials": 4, # 4 in parallel
        "timeout": 7200            # 2h max
    }
)

# Define the hyperparameter space
sweep_job.inputs.learning_rate = LogUniform(min_value=-3, max_value=0)  # 10^-3 to 1
sweep_job.inputs.n_estimators = Choice(values=[50, 100, 200, 300])
sweep_job.inputs.max_depth = Choice(values=[2, 3, 4, 5, 6])

sweep_job.inputs.train_data = Input(
    path="azureml:automobile-train:1",
    type="uri_folder"
)

# Submit
sweep_returned = ml_client.jobs.create_or_update(sweep_job)
print(f"✅ Sweep job submitted: {sweep_returned.name}")
print(f"   URL: {sweep_returned.studio_url}")

# Wait and retrieve the best run
ml_client.jobs.stream(sweep_returned.name)

# Find the best run from the sweep
best_sweep_run = ml_client.jobs.get(sweep_returned.name)
print(f"\nBest Sweep Run:")
print(f"  Run ID: {best_sweep_run.properties.get('best_child_run_id', 'N/A')}")

11. MLOps – CI/CD Integration

11.1 Complete MLOps Workflow

flowchart TD
    CODE["👨‍💻 Data Scientist\n(Code + Data)"] -->|git push| REPO["📁 Git Repository\n(Azure DevOps / GitHub)"]
    
    REPO -->|Trigger CI| CI["🔧 CI Pipeline\n(Azure DevOps)"]
    CI --> LINT["Code Linting\n(flake8, black)"]
    CI --> UNIT_TEST["Unit Tests\n(pytest)"]
    CI --> VALIDATE["Validate YAML\n(Component schemas)"]
    
    LINT --> TRAIN_PIPELINE["🚀 ML Training Pipeline\n(Azure ML)"]
    UNIT_TEST --> TRAIN_PIPELINE
    VALIDATE --> TRAIN_PIPELINE
    
    TRAIN_PIPELINE --> EVAL_GATE{"Metrics\nsatisfactory?\n(R² > 0.85)"}
    EVAL_GATE -->|Yes| REGISTER["📦 Register Model\n(Azure ML Model Registry)"]
    EVAL_GATE -->|No| NOTIF_FAIL["❌ Notification\n(Email, Teams)"]
    
    REGISTER --> CD["🚀 CD Pipeline\n(Azure DevOps)"]
    CD --> STAGING["Staging Deployment\n(Integration Tests)"]
    STAGING --> PROD_GATE{"Manual\nApproval?"}
    PROD_GATE -->|Approved| PROD["🌍 Production\n(Managed Online Endpoint)"]
    PROD_GATE -->|Rejected| ROLLBACK["↩️ Rollback\n(Previous Version)"]

11.2 GitHub Actions Pipeline for MLOps

# .github/workflows/ml_pipeline.yml
name: ML Training Pipeline

on:
  push:
    branches: [main, develop]
    paths:
      - 'ml/**'
      - 'data/**'

env:
  AZURE_SUBSCRIPTION_ID: ${{ secrets.AZURE_SUBSCRIPTION_ID }}
  AZURE_RESOURCE_GROUP: ${{ secrets.AZURE_RESOURCE_GROUP }}
  AZURE_ML_WORKSPACE: ${{ secrets.AZURE_ML_WORKSPACE }}

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.10"
      
      - name: Install dependencies
        run: |
          pip install flake8 black pytest azure-ai-ml
      
      - name: Lint with flake8
        run: |
          flake8 ml/ --max-line-length=88
      
      - name: Format check with black
        run: |
          black --check ml/
      
      - name: Run unit tests
        run: |
          pytest tests/ -v --tb=short

  train-model:
    needs: lint-and-test
    runs-on: ubuntu-latest
    
    steps:
      - uses: actions/checkout@v3
      
      - name: Azure Login
        uses: azure/login@v1
        with:
          creds: ${{ secrets.AZURE_CREDENTIALS }}
      
      - name: Install Azure ML SDK
        run: pip install azure-ai-ml azure-identity
      
      - name: Run ML Pipeline
        run: |
          python ml/pipeline.py \
            --subscription_id $AZURE_SUBSCRIPTION_ID \
            --resource_group $AZURE_RESOURCE_GROUP \
            --workspace $AZURE_ML_WORKSPACE \
            --experiment_name "ci-cd-pipeline-${{ github.run_number }}"
      
      - name: Check model quality
        run: |
          python ml/check_model_quality.py \
            --min_r2 0.85 \
            --experiment "ci-cd-pipeline-${{ github.run_number }}"
      
      - name: Deploy to staging if quality passes
        if: success()
        run: |
          python ml/deploy_staging.py \
            --model_name "automobile-price-model"

12. Summary and Key Takeaways

12.1 Reference MLOps Architecture

flowchart TB
    subgraph "Development"
        DEV["Jupyter / VSCode\n(Exploration)"] --> GIT["Git\n(Version Control)"]
    end
    
    subgraph "Azure ML Workspace"
        DATA["Datasets\n(Versioned)"]
        PIPE["Pipelines\n(Reproducible)"]
        EXP["Experiments\n(MLflow Tracking)"]
        REG["Model Registry\n(Versions)"]
    end
    
    subgraph "Production"
        ONLINE["Managed Online Endpoint\n(Real-time)"]
        BATCH_EP["Batch Endpoint\n(Batch)"]
        MONITOR["Monitor\n(Metrics, Drift)"]
    end
    
    GIT --> PIPE
    DATA --> PIPE
    PIPE --> EXP
    EXP --> REG
    REG --> ONLINE
    REG --> BATCH_EP
    ONLINE --> MONITOR
    BATCH_EP --> MONITOR

12.2 Summary Table

13. Glossary

Additional Resources:

Appendix: Quick Reference

A.1 What is an ML Pipeline?

An ML Pipeline is an ordered, modular sequence of steps for building, training, and deploying ML models.

Benefits:

• Modularity: each step is independent and reusable
• Scalability: parallel execution of independent steps
• Traceability: every run is logged with its parameters, metrics, and artifacts
• Automation: automatic scheduling and triggering

Typical pipeline structure:

Data Preparation Step
        ↓
Training Step
        ↓
Evaluation Step
        ↓
Save / Register Model Step

A.2 Creating a Pipeline – Two Methods

Via SDK v2:

from azure.ai.ml import MLClient, dsl, Output
from azure.ai.ml.dsl import pipeline
from azure.ai.ml import command, Input

@pipeline(compute="my-cluster", experiment_name="ml-pipeline-exp")
def my_pipeline(data_input):
    prep_step = command(
        command="python prep.py --data ${{inputs.data}}",
        inputs={"data": data_input},
        outputs={"processed": Output(type="uri_folder")},
        code="./src/prep",
        environment="AzureML-sklearn-1.0-ubuntu20.04-py38-cpu:1"
    )
    train_step = command(
        command="python train.py --data ${{inputs.data}}",
        inputs={"data": prep_step.outputs.processed},
        code="./src/train",
        environment="AzureML-sklearn-1.0-ubuntu20.04-py38-cpu:1"
    )
    return {"trained_model": train_step.outputs.model}

A.3 MLflow Tracking – Essential Commands

Logging in a training script:

import mlflow
import mlflow.sklearn

with mlflow.start_run():
    # Log parameters
    lr = 0.01
    mlflow.log_param("learning_rate", lr)
    mlflow.log_param("max_iter", 200)

    # Train the model
    model = LogisticRegression(max_iter=200)
    model.fit(X_train, y_train)

    # Log metrics
    accuracy = accuracy_score(y_test, model.predict(X_test))
    mlflow.log_metric("accuracy", accuracy)
    mlflow.log_metric("n_features", X_train.shape[1])

    # Log the model
    mlflow.sklearn.log_model(model, "model")
    
    print(f"Accuracy: {accuracy}")

Auto-logging (without manual code):

import mlflow.sklearn

# Automatically enable for sklearn
mlflow.sklearn.autolog()

# Then train normally
model = LogisticRegression()
model.fit(X_train, y_train)
# Parameters and metrics are logged automatically!

Supported frameworks for autolog:

• mlflow.sklearn.autolog()
• mlflow.pytorch.autolog()
• mlflow.tensorflow.autolog()
• mlflow.xgboost.autolog()
• mlflow.lightgbm.autolog()

A.4 Registering a Model

Via SDK:

import mlflow

# After an MLflow run
with mlflow.start_run() as run:
    mlflow.sklearn.log_model(model, "model")
    run_id = run.info.run_id

# Register the model in the registry
model_uri = f"runs:/{run_id}/model"
mlflow.register_model(model_uri, "fraud-detection-model")

Model versions:

• Each registration creates a new version
• Versions can have tags (Staging, Production)
• Deployment endpoints point to a specific version

A.5 Essential SDK Commands

ml_client.jobs.create_or_update(pipeline_job)   # Submit
ml_client.jobs.stream(job_name)                  # Wait
ml_client.jobs.get(job_name)                     # Check status
mlflow.log_param("key", value)                   # Log parameter
mlflow.log_metric("key", value)                  # Log metric
mlflow.log_artifact("path/to/file")              # Log artifact
mlflow.register_model(model_uri, "model-name")  # Register model

Search Terms

azure · ml · pipelines · experiment · tracking · platforms · deployment · machine · data · science · pipeline · mlflow · mlops · model · component · sdk · commands · concepts · core · essential · handling · hyperparameter · job · monitoring