MLflow
Open Command Prompt or PowerShell and run:
python --version
Create Virtual Environment
# Create virtual environment
python -m venv mlflow-env
# Activate virtual environment
mlflow-env\Scripts\activate
# You should see (mlflow-env) in your command prompt
Upgrade pip
python -m pip install --upgrade pip
Install Required Libraries
# Install MLflow
pip install mlflow
# Install scikit-learn
pip install scikit-learn
# Install additional dependencies
pip install pandas matplotlib seaborn
# Verify installations
mlflow --version
python -c "import sklearn; print(sklearn.__version__)"
Expected Output:
mlflow, version 2.21.3
1.3.0
Environment Setup
Start MLflow Tracking Server
Open a new Command Prompt/PowerShell window (keep virtual environment activated):
# Navigate to your project directory
cd C:\your-project-path
# Start MLflow UI
mlflow ui
The MLflow UI will start at: [Link]
Keep this terminal window open - it's running your MLflow server.
Create Project Structure
Create the following folder structure:
iris-mlflow-project/
│
├── data/
├── models/
├── notebooks/
└── src/
└── [Link]
Complete Iris Classification Pipeline
Create [Link] in the src/ folder:
# src/[Link]
import mlflow
import [Link]
import pandas as pd
import numpy as np
from [Link] import load_iris
from sklearn.model_selection import train_test_split
from [Link] import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
from [Link] import accuracy_score, precision_score, recall_score,
f1_score
from [Link] import StandardScaler
from [Link] import infer_signature
import [Link] as plt
import seaborn as sns
def load_and_explore_data():
"""Load and explore the Iris dataset"""
print("Loading Iris dataset...")
# Load dataset
iris = load_iris()
X = [Link]([Link], columns=iris.feature_names)
y = [Link]([Link], name='species')
# Create DataFrame for easier handling
df = [Link]()
df['species'] = y
df['species_name'] = df['species'].map({0: 'setosa', 1: 'versicolor', 2:
'virginica'})
print(f"Dataset shape: {[Link]}")
print(f"Features: {list([Link])}")
print(f"Classes: {iris.target_names}")
print(f"Class distribution:\n{df['species_name'].value_counts()}")
return X, y, df
def preprocess_data(X, y, test_size=0.2, random_state=42):
"""Split and preprocess the data"""
print("Preprocessing data...")
# Split the data
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=test_size, random_state=random_state, stratify=y
)
# Scale the features
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = [Link](X_test)
# Convert back to DataFrame for easier handling
X_train_scaled = [Link](X_train_scaled, columns=[Link])
X_test_scaled = [Link](X_test_scaled, columns=[Link])
print(f"Training set size: {X_train_scaled.shape[0]}")
print(f"Test set size: {X_test_scaled.shape[0]}")
return X_train_scaled, X_test_scaled, y_train, y_test, scaler
def calculate_metrics(y_true, y_pred):
"""Calculate comprehensive metrics"""
return {
'accuracy': accuracy_score(y_true, y_pred),
'precision': precision_score(y_true, y_pred, average='weighted'),
'recall': recall_score(y_true, y_pred, average='weighted'),
'f1_score': f1_score(y_true, y_pred, average='weighted')
}
def train_random_forest(X_train, y_train, X_test, y_test, **params):
"""Train Random Forest model with MLflow tracking"""
with mlflow.start_run(run_name="Random Forest Classifier") as run:
# Log parameters
mlflow.log_params(params)
mlflow.log_param("model_type", "RandomForestClassifier")
mlflow.log_param("train_size", len(X_train))
mlflow.log_param("test_size", len(X_test))
# Train model
model = RandomForestClassifier(**params)
[Link](X_train, y_train)
# Make predictions
y_pred_train = [Link](X_train)
y_pred_test = [Link](X_test)
# Calculate metrics
train_metrics = calculate_metrics(y_train, y_pred_train)
test_metrics = calculate_metrics(y_test, y_pred_test)
# Log training metrics
for metric, value in train_metrics.items():
mlflow.log_metric(f"train_{metric}", value)
# Log test metrics
for metric, value in test_metrics.items():
mlflow.log_metric(f"test_{metric}", value)
# Log feature importance
feature_importance = [Link]({
'feature': X_train.columns,
'importance': model.feature_importances_
}).sort_values('importance', ascending=False)
mlflow.log_text(feature_importance.to_string(),
"feature_importance.txt")
# Create and log confusion matrix plot
from [Link] import confusion_matrix
import [Link] as plt
cm = confusion_matrix(y_test, y_pred_test)
[Link](figsize=(8, 6))
[Link](cm, annot=True, fmt='d', cmap='Blues')
[Link]('Confusion Matrix - Random Forest')
[Link]('Actual')
[Link]('Predicted')
[Link]('confusion_matrix_rf.png')
mlflow.log_artifact('confusion_matrix_rf.png')
[Link]()
# Infer model signature and log model
signature = infer_signature(X_test, y_pred_test)
[Link].log_model(
sk_model=model,
artifact_path="model",
signature=signature,
input_example=X_test.head(3)
)
# Log additional tags
mlflow.set_tag("model_family", "tree_based")
mlflow.set_tag("dataset", "iris")
print(f"Random Forest - Test Accuracy: {test_metrics['accuracy']:.4f}")
return model, test_metrics['accuracy']
def train_logistic_regression(X_train, y_train, X_test, y_test, **params):
"""Train Logistic Regression model with MLflow tracking"""
with mlflow.start_run(run_name="Logistic Regression") as run:
# Log parameters
mlflow.log_params(params)
mlflow.log_param("model_type", "LogisticRegression")
mlflow.log_param("train_size", len(X_train))
mlflow.log_param("test_size", len(X_test))
# Train model
model = LogisticRegression(**params)
[Link](X_train, y_train)
# Make predictions
y_pred_train = [Link](X_train)
y_pred_test = [Link](X_test)
# Calculate metrics
train_metrics = calculate_metrics(y_train, y_pred_train)
test_metrics = calculate_metrics(y_test, y_pred_test)
# Log training metrics
for metric, value in train_metrics.items():
mlflow.log_metric(f"train_{metric}", value)
# Log test metrics
for metric, value in test_metrics.items():
mlflow.log_metric(f"test_{metric}", value)
# Create and log confusion matrix plot
from [Link] import confusion_matrix
cm = confusion_matrix(y_test, y_pred_test)
[Link](figsize=(8, 6))
[Link](cm, annot=True, fmt='d', cmap='Greens')
[Link]('Confusion Matrix - Logistic Regression')
[Link]('Actual')
[Link]('Predicted')
[Link]('confusion_matrix_lr.png')
mlflow.log_artifact('confusion_matrix_lr.png')
[Link]()
# Infer model signature and log model
signature = infer_signature(X_test, y_pred_test)
[Link].log_model(
sk_model=model,
artifact_path="model",
signature=signature,
input_example=X_test.head(3)
)
# Log additional tags
mlflow.set_tag("model_family", "linear")
mlflow.set_tag("dataset", "iris")
print(f"Logistic Regression - Test Accuracy:
{test_metrics['accuracy']:.4f}")
return model, test_metrics['accuracy']
def hyperparameter_tuning():
"""Perform hyperparameter tuning with nested runs"""
# Load and preprocess data
X, y, df = load_and_explore_data()
X_train, X_test, y_train, y_test, scaler = preprocess_data(X, y)
# Set experiment
mlflow.set_experiment("Iris Classification Hyperparameter Tuning")
with mlflow.start_run(run_name="Hyperparameter Tuning Session") as
parent_run:
best_accuracy = 0
best_model = None
best_run_id = None
# Random Forest hyperparameter combinations
rf_params = [
{'n_estimators': 50, 'max_depth': 3, 'random_state': 42},
{'n_estimators': 100, 'max_depth': 5, 'random_state': 42},
{'n_estimators': 200, 'max_depth': 7, 'random_state': 42},
{'n_estimators': 100, 'max_depth': None, 'random_state': 42}
]
# Train Random Forest variants
for params in rf_params:
with mlflow.start_run(nested=True,
run_name=f"RF_estimators_{params['n_estimators']}_depth_{params['max_depth']}")
:
model, accuracy = train_random_forest(X_train, y_train, X_test,
y_test, **params)
if accuracy > best_accuracy:
best_accuracy = accuracy
best_model = model
best_run_id = mlflow.active_run().info.run_id
# Logistic Regression hyperparameter combinations
lr_params = [
{'C': 0.1, 'random_state': 42, 'max_iter': 1000},
{'C': 1.0, 'random_state': 42, 'max_iter': 1000},
{'C': 10.0, 'random_state': 42, 'max_iter': 1000},
]
# Train Logistic Regression variants
for params in lr_params:
with mlflow.start_run(nested=True, run_name=f"LR_C_{params['C']}"):
model, accuracy = train_logistic_regression(X_train, y_train,
X_test, y_test, **params)
if accuracy > best_accuracy:
best_accuracy = accuracy
best_model = model
best_run_id = mlflow.active_run().info.run_id
# Log best model information in parent run
mlflow.log_metric("best_accuracy", best_accuracy)
mlflow.log_param("best_run_id", best_run_id)
mlflow.set_tag("status", "completed")
print(f"\nBest model accuracy: {best_accuracy:.4f}")
print(f"Best run ID: {best_run_id}")
return best_model, best_run_id
def model_registry_example(best_run_id):
"""Demonstrate model registry functionality"""
# Register the best model
model_name = "iris-classifier"
model_uri = f"runs:/{best_run_id}/model"
# Register model
model_version = mlflow.register_model(
model_uri=model_uri,
name=model_name
)
print(f"Model registered: {model_name}, Version: {model_version.version}")
# Add model version tags and description
client = [Link]()
client.update_model_version(
name=model_name,
version=model_version.version,
description="Best performing model from hyperparameter tuning session"
)
# Set model version tags
client.set_model_version_tag(
name=model_name,
version=model_version.version,
key="stage",
value="production_candidate"
)
# Transition to production (optional)
client.transition_model_version_stage(
name=model_name,
version=model_version.version,
stage="Production"
)
print(f"Model {model_name} version {model_version.version} moved to
Production stage")
return model_name, model_version.version
def load_and_predict():
"""Load model from registry and make predictions"""
# Load model from registry
model_name = "iris-classifier"
model = [Link].load_model(f"models:/{model_name}/Production")
# Create sample data for prediction
X, y, df = load_and_explore_data()
X_sample = [Link](5)
# Make predictions
predictions = [Link](X_sample)
# Display results
results = [Link]({
'sepal_length': X_sample['sepal length (cm)'],
'sepal_width': X_sample['sepal width (cm)'],
'petal_length': X_sample['petal length (cm)'],
'petal_width': X_sample['petal width (cm)'],
'predicted_class': predictions,
'predicted_species': [['setosa', 'versicolor', 'virginica'][p] for p in
predictions]
})
print("\nPrediction Results:")
print(results.to_string(index=False))
return results
def main():
"""Main function to run the complete pipeline"""
print("=== MLflow Iris Classification Pipeline ===\n")
# Set tracking URI (if using remote server)
# mlflow.set_tracking_uri("[Link]
# Run hyperparameter tuning
best_model, best_run_id = hyperparameter_tuning()
# Register the best model
model_name, version = model_registry_example(best_run_id)
# Load and make predictions
results = load_and_predict()
print(f"\n=== Pipeline Completed Successfully ===")
print(f"Best model registered as: {model_name} (Version: {version})")
print(f"MLflow UI: [Link]
if __name__ == "__main__":
main()
Run the Complete Pipeline
# Navigate to your project directory
cd iris-mlflow-project
# Run the training script
python src/[Link]
Expected Output:
=== MLflow Iris Classification Pipeline ===
Loading Iris dataset...
Dataset shape: (150, 5)
Features: ['sepal length (cm)', 'sepal width (cm)', 'petal length (cm)', 'petal width
(cm)']
Classes: ['setosa' 'versicolor' 'virginica']
Class distribution:
setosa 50
versicolor 50
virginica 50
Name: species_name, dtype: int64
Preprocessing data...
Training set size: 120
Test set size: 30
Random Forest - Test Accuracy: 1.0000
Random Forest - Test Accuracy: 1.0000
Random Forest - Test Accuracy: 1.0000
Random Forest - Test Accuracy: 1.0000
Logistic Regression - Test Accuracy: 1.0000
Logistic Regression - Test Accuracy: 1.0000
Logistic Regression - Test Accuracy: 1.0000
Best model accuracy: 1.0000
Best run ID: abc123def456
Model registered: iris-classifier, Version: 1
Model iris-classifier version 1 moved to Production stage
Prediction Results:
sepal_length sepal_width petal_length petal_width predicted_class
predicted_species
5.1 3.5 1.4 0.2 0
setosa
4.9 3.0 1.4 0.2 0
setosa
4.7 3.2 1.3 0.2 0
setosa
4.6 3.1 1.5 0.2 0
setosa
5.0 3.6 1.4 0.2 0
setosa
=== Pipeline Completed Successfully ===
Best model registered as: iris-classifier (Version: 1)
MLflow UI: [Link]
Model Serving (Optional)
To serve your model as a REST API:
# Serve the model
mlflow models serve -m "models:/iris-classifier/Production" -p 1234
Test the API:
# Test prediction
curl -X POST -H "Content-Type:application/json" --data "{"dataframe_split":
{"columns": ["sepal length (cm)", "sepal width (cm)", "petal length (cm)", "petal
width (cm)"], "data": [[5.1, 3.5, 1.4, 0.2]]}}" [Link]
Model Comparison Script
Create compare_models.py:
# src/compare_models.py
import mlflow
import pandas as pd
from [Link] import MlflowClient
def compare_experiments():
"""Compare all experiments and runs"""
client = MlflowClient()
# Get experiment by name
experiment = client.get_experiment_by_name("Iris Classification
Hyperparameter Tuning")
if experiment:
# Get all runs from the experiment
runs = client.search_runs(experiment.experiment_id)
# Create comparison DataFrame
comparison_data = []
for run in runs:
run_data = {
'run_id': [Link].run_id,
'run_name': [Link]('[Link]', 'Unknown'),
'model_type': [Link]('model_type', 'Unknown'),
'test_accuracy': [Link]('test_accuracy', 0),
'test_f1_score': [Link]('test_f1_score', 0),
'status': [Link]
}
comparison_data.append(run_data)
# Sort by test accuracy
df = [Link](comparison_data)
df = df.sort_values('test_accuracy', ascending=False)
print("Model Comparison Results:")
print("=" * 80)
print(df.to_string(index=False))
return df
else:
print("Experiment not found!")
return None
if __name__ == "__main__":
compare_experiments()
# Run as administrator or change directory permissions
# Ensure you have write access to the project directory