📊 Live MLOps Tracking Dashboard¶

This dashboard aggregates and presents the tracking data generated by running the tutorial pipeline modules. It queries MLflow for experiment history and model registrations, and inspects DVC for data version status.

Because execute: true is configured in mkdocs.yml, the metrics and logs shown here are automatically re-evaluated and refreshed whenever the tutorials are compiled.

In [1]:

import os
import mlflow
import pandas as pd
from IPython.display import display, HTML

# Display current tracking URI configurations
tracking_uri = mlflow.get_tracking_uri()
print(f"📁 MLflow Tracking URI: {tracking_uri}")
try:
    resolved_path = os.path.abspath(tracking_uri.replace('file://', ''))
    print(f"📂 Resolved Absolute Path: {resolved_path}")
except Exception:
    pass

import os import mlflow import pandas as pd from IPython.display import display, HTML # Display current tracking URI configurations tracking_uri = mlflow.get_tracking_uri() print(f"📁 MLflow Tracking URI: {tracking_uri}") try: resolved_path = os.path.abspath(tracking_uri.replace('file://', '')) print(f"📂 Resolved Absolute Path: {resolved_path}") except Exception: pass

📁 MLflow Tracking URI: sqlite:////home/t/MLOps/mlflow.db
📂 Resolved Absolute Path: /home/t/MLOps/sqlite:/home/t/MLOps/mlflow.db

🔬 MLflow Runs Summary¶

List of all active tracking experiments and their recorded metrics:

In [2]:

try:
    experiments = mlflow.search_experiments()
    if not experiments:
        print("ℹ️ No MLflow experiments found. Run some tutorial scripts first!")
    else:
        runs_df = mlflow.search_runs()
        if runs_df.empty:
            print("ℹ️ MLflow tracking directory exists, but no runs have been logged yet.")
        else:
            # Format Date/Time
            if 'start_time' in runs_df.columns:
                runs_df['start_time'] = pd.to_datetime(runs_df['start_time']).dt.strftime('%Y-%m-%d %H:%M:%S')
            
            # Map experiment names
            exp_mapping = {exp.experiment_id: exp.name for exp in experiments}
            runs_df['experiment_name'] = runs_df['experiment_id'].map(exp_mapping)
            
            # Identify parameter and metric columns
            param_cols = [col for col in runs_df.columns if col.startswith('params.')]
            metric_cols = [col for col in runs_df.columns if col.startswith('metrics.')]
            
            # Select columns to display
            all_display_cols = ['experiment_name', 'run_id', 'status', 'start_time'] + param_cols + metric_cols
            all_display_cols = [c for c in all_display_cols if c in runs_df.columns]
            
            display_df = runs_df[all_display_cols].copy()
            
            # Strip column prefixes
            rename_dict = {}
            for col in display_df.columns:
                if col.startswith('params.'):
                    rename_dict[col] = f"🔧 {col.replace('params.', '')}"
                elif col.startswith('metrics.'):
                    rename_dict[col] = f"📈 {col.replace('metrics.', '')}"
                elif col == 'experiment_name':
                    rename_dict[col] = 'Experiment'
                elif col == 'run_id':
                    rename_dict[col] = 'Run ID'
                elif col == 'status':
                    rename_dict[col] = 'Status'
                elif col == 'start_time':
                    rename_dict[col] = 'Date/Time'
            
            display_df.rename(columns=rename_dict, inplace=True)
            display(HTML(display_df.to_html(classes='dataframe', index=False)))
except Exception as e:
    print(f"⚠️ Error loading MLflow runs: {e}")

try: experiments = mlflow.search_experiments() if not experiments: print("ℹ️ No MLflow experiments found. Run some tutorial scripts first!") else: runs_df = mlflow.search_runs() if runs_df.empty: print("ℹ️ MLflow tracking directory exists, but no runs have been logged yet.") else: # Format Date/Time if 'start_time' in runs_df.columns: runs_df['start_time'] = pd.to_datetime(runs_df['start_time']).dt.strftime('%Y-%m-%d %H:%M:%S') # Map experiment names exp_mapping = {exp.experiment_id: exp.name for exp in experiments} runs_df['experiment_name'] = runs_df['experiment_id'].map(exp_mapping) # Identify parameter and metric columns param_cols = [col for col in runs_df.columns if col.startswith('params.')] metric_cols = [col for col in runs_df.columns if col.startswith('metrics.')] # Select columns to display all_display_cols = ['experiment_name', 'run_id', 'status', 'start_time'] + param_cols + metric_cols all_display_cols = [c for c in all_display_cols if c in runs_df.columns] display_df = runs_df[all_display_cols].copy() # Strip column prefixes rename_dict = {} for col in display_df.columns: if col.startswith('params.'): rename_dict[col] = f"🔧 {col.replace('params.', '')}" elif col.startswith('metrics.'): rename_dict[col] = f"📈 {col.replace('metrics.', '')}" elif col == 'experiment_name': rename_dict[col] = 'Experiment' elif col == 'run_id': rename_dict[col] = 'Run ID' elif col == 'status': rename_dict[col] = 'Status' elif col == 'start_time': rename_dict[col] = 'Date/Time' display_df.rename(columns=rename_dict, inplace=True) display(HTML(display_df.to_html(classes='dataframe', index=False))) except Exception as e: print(f"⚠️ Error loading MLflow runs: {e}")

ℹ️ MLflow tracking directory exists, but no runs have been logged yet.

📦 Model Registry¶

The Model Registry stores versions of models ready for deployment:

In [3]:

from mlflow.tracking import MlflowClient

try:
    client = MlflowClient()
    registered_models = client.search_registered_models()
    
    if not registered_models:
        print("ℹ️ No models registered in registry yet.")
    else:
        model_data = []
        for model in registered_models:
            for version in model.latest_versions:
                model_data.append({
                    "Model Name": model.name,
                    "Version": f"v{version.version}",
                    "Stage": version.current_stage,
                    "Run ID": version.run_id,
                    "Source URI": version.source,
                    "Last Updated": pd.to_datetime(version.last_updated_timestamp, unit='ms').strftime('%Y-%m-%d %H:%M:%S')
                })
        
        if model_data:
            model_df = pd.DataFrame(model_data)
            display(HTML(model_df.to_html(classes='dataframe', index=False)))
        else:
            print("ℹ️ Models exist, but no active versions are registered.")
except Exception as e:
    print(f"⚠️ Error loading Model Registry: {e}")

from mlflow.tracking import MlflowClient try: client = MlflowClient() registered_models = client.search_registered_models() if not registered_models: print("ℹ️ No models registered in registry yet.") else: model_data = [] for model in registered_models: for version in model.latest_versions: model_data.append({ "Model Name": model.name, "Version": f"v{version.version}", "Stage": version.current_stage, "Run ID": version.run_id, "Source URI": version.source, "Last Updated": pd.to_datetime(version.last_updated_timestamp, unit='ms').strftime('%Y-%m-%d %H:%M:%S') }) if model_data: model_df = pd.DataFrame(model_data) display(HTML(model_df.to_html(classes='dataframe', index=False))) else: print("ℹ️ Models exist, but no active versions are registered.") except Exception as e: print(f"⚠️ Error loading Model Registry: {e}")

Model Name	Version	Stage	Run ID	Source URI	Last Updated
HousingPriceCTModel	v51	None	3c446dad9da34059b0ea928e58bed8d8	models:/m-26786de4111a43fcbb8a7510baa1bb65	2026-06-20 17:08:22
HousingPriceRegressionModel	v38	None	a85f84bf8c714bccafcbf8a8d42c7ae1	models:/m-7e3d625d695c45aca41c7de0edd4b967	2026-06-20 17:09:27
HousingRandomForestModel	v83	None	39597033bb2c4dca849b3e7c3174502e	models:/m-f64b9326c51a4470b65f7203ec614e0c	2026-06-20 17:09:34
TestHousingPriceCTModel	v15	None	92992c06437148bda7f3cc9a33b68019	models:/m-addfa17e4f6f43dba08a3f948bd3708f	2026-06-17 04:18:10

🗄️ DVC Data Tracking Status¶

Checking workspace status and listing tracked assets.

Equivalent Shell Command

dvc status

In [4]:

from dvc.repo import Repo
import glob

# Query DVC status using DVC Python API
try:
    repo = Repo(".")
    status = repo.status()
    if not status:
        print("✅ DVC workspace is clean. All tracked data artifacts are up to date.")
    else:
        print("⚠️ DVC workspace has modified or untracked changes:")
        for stage, changes in status.items():
            print(f"Stage '{stage}': {changes}")
except Exception as e:
    print(f"⚠️ Could not execute dvc status: {e}")

from dvc.repo import Repo import glob # Query DVC status using DVC Python API try: repo = Repo(".") status = repo.status() if not status: print("✅ DVC workspace is clean. All tracked data artifacts are up to date.") else: print("⚠️ DVC workspace has modified or untracked changes:") for stage, changes in status.items(): print(f"Stage '{stage}': {changes}") except Exception as e: print(f"⚠️ Could not execute dvc status: {e}")

⚠️ DVC workspace has modified or untracked changes:

Stage 'prepare': [{'changed deps': {'src/data_experimentation/module_06_integrated_pipeline/run_pipeline.py': 'modified'}}]

Stage 'train': [{'changed deps': {'src/data_experimentation/module_06_integrated_pipeline/run_pipeline.py': 'modified'}}]

Stage 'evaluate': [{'changed deps': {'src/data_experimentation/module_06_integrated_pipeline/run_pipeline.py': 'modified'}}]

In [5]:

# Locate and parse all .dvc files to show what assets are tracked
dvc_files = glob.glob("**/*.dvc", recursive=True)
if not dvc_files:
    print("ℹ️ No `.dvc` tracking metadata files found in this workspace.")
else:
    dvc_data = []
    for f in dvc_files:
        try:
            with open(f, 'r') as file:
                lines = file.readlines()
                outs_section = False
                md5 = ""
                path = ""
                for line in lines:
                    if "outs:" in line:
                        outs_section = True
                    if outs_section:
                        if "md5:" in line:
                            md5 = line.split("md5:")[-1].strip()
                        if "path:" in line:
                            path = line.split("path:")[-1].strip()
                dvc_data.append({
                    "DVC File": f,
                    "Tracked Asset": path if path else "N/A",
                    "Content Hash (MD5)": md5 if md5 else "N/A"
                })
        except Exception:
            pass
            
    if dvc_data:
        dvc_df = pd.DataFrame(dvc_data)
        display(HTML(dvc_df.to_html(classes='dataframe', index=False)))

# Locate and parse all .dvc files to show what assets are tracked dvc_files = glob.glob("**/*.dvc", recursive=True) if not dvc_files: print("ℹ️ No `.dvc` tracking metadata files found in this workspace.") else: dvc_data = [] for f in dvc_files: try: with open(f, 'r') as file: lines = file.readlines() outs_section = False md5 = "" path = "" for line in lines: if "outs:" in line: outs_section = True if outs_section: if "md5:" in line: md5 = line.split("md5:")[-1].strip() if "path:" in line: path = line.split("path:")[-1].strip() dvc_data.append({ "DVC File": f, "Tracked Asset": path if path else "N/A", "Content Hash (MD5)": md5 if md5 else "N/A" }) except Exception: pass if dvc_data: dvc_df = pd.DataFrame(dvc_data) display(HTML(dvc_df.to_html(classes='dataframe', index=False)))

DVC File	Tracked Asset	Content Hash (MD5)
data/housing_raw.csv.dvc	housing_raw.csv	d7636a34e39fd3c06fc3ac8a94de52b3
data/extracted/FM_3-05-2025.md.dvc	FM_3-05-2025.md	ca94dfd13c03b64a3a7fd91196a40394