The ML Cost Visibility Problem
ML infrastructure costs are growing faster than any other cloud spend category—typically 30-50% year-over-year. A mid-size company might spend $200K/month on GPU training, $50K/month on inference, and $30K/month on data storage and processing. But when the CFO asks "which team is spending what?", the answer is usually "we don't know." Cloud bills show aggregate costs by service (EC2, S3, SageMaker) but not by team, project, model, or business outcome.
Without cost attribution, there's no accountability. Teams have no incentive to optimize because they don't see their bill. Finance can't forecast because they can't map costs to business units. And leadership can't make informed decisions about ML investment because they can't connect spend to value. This article provides a complete FinOps showback model for MLOps—attributing every dollar of ML infrastructure cost to the team, project, and model that generated it.
Phase 1: Cost Tagging Strategy
Mandatory Tag Schema
# finops/tag_schema.yaml — Mandatory tags for all ML resources
tag_schema:
required_tags:
team:
description: "Team that owns this resource"
values: ["data-science", "ml-engineering", "product-ml", "research"]
example: "data-science"
cost_center:
description: "Finance cost center code"
pattern: "^CC-\\d{4}$"
example: "CC-4521"
project:
description: "ML project identifier"
example: "churn-prediction"
environment:
description: "Deployment environment"
values: ["development", "staging", "production"]
example: "production"
model_name:
description: "Name of the ML model"
example: "customer-churn-v3"
resource_purpose:
description: "What this resource is used for"
values: ["training", "inference", "experimentation", "data-processing", "feature-store", "monitoring"]
example: "training"
optional_tags:
experiment_id:
description: "MLflow experiment ID"
example: "exp-42"
pipeline_run_id:
description: "CI/CD pipeline run that created this resource"
example: "run-20250115-001"
owner_email:
description: "Email of the resource owner"
example: "alice@company.com"
# Enforcement: deny resource creation without required tags
# Applied via AWS Service Control Policies or Kubernetes OPA/GatekeeperAutomated Tag Enforcement
# finops/tag_enforcement.py — Enforce tagging on all ML resources
import boto3
import json
from typing import Dict, List
REQUIRED_TAGS = ["team", "cost_center", "project", "environment", "model_name", "resource_purpose"]
class TagEnforcer:
"""Enforce mandatory tags on all ML infrastructure resources."""
def __init__(self):
self.ec2 = boto3.client("ec2")
self.sagemaker = boto3.client("sagemaker")
self.s3 = boto3.client("s3")
def validate_tags(self, tags: Dict[str, str]) -> List[str]:
"""Validate that all required tags are present."""
missing = [tag for tag in REQUIRED_TAGS if tag not in tags]
return missing
def tag_sagemaker_training_job(self, job_name: str, tags: Dict[str, str]):
"""Ensure SageMaker training jobs are properly tagged."""
missing = self.validate_tags(tags)
if missing:
raise ValueError(f"Missing required tags: {missing}")
tag_list = [{"Key": k, "Value": v} for k, v in tags.items()]
self.sagemaker.add_tags(
ResourceArn=self._get_training_job_arn(job_name),
Tags=tag_list,
)
def auto_tag_from_mlflow(self, resource_id: str, mlflow_run_id: str):
"""
Auto-tag resources based on MLflow run metadata.
This eliminates manual tagging for experiment resources.
"""
import mlflow
run = mlflow.get_run(mlflow_run_id)
auto_tags = {
"team": run.data.tags.get("mlflow.user", "unknown"),
"project": run.data.tags.get("mlflow.project", "unknown"),
"experiment_id": run.info.experiment_id,
"pipeline_run_id": mlflow_run_id,
"model_name": run.data.tags.get("model_name", "unknown"),
"environment": run.data.tags.get("environment", "development"),
}
return auto_tags
def scan_untagged_resources(self) -> Dict[str, List]:
"""Find all ML resources missing required tags."""
untagged = {"ec2": [], "sagemaker": [], "s3": []}
# Scan EC2 instances (GPU training servers)
instances = self.ec2.describe_instances(
Filters=[{"Name": "instance-type", "Values": ["p*", "g*"]}]
)
for reservation in instances["Reservations"]:
for instance in reservation["Instances"]:
tags = {t["Key"]: t["Value"] for t in instance.get("Tags", [])}
missing = self.validate_tags(tags)
if missing:
untagged["ec2"].append({
"instance_id": instance["InstanceId"],
"missing_tags": missing,
})
# Scan SageMaker endpoints
endpoints = self.sagemaker.list_endpoints()["Endpoints"]
for ep in endpoints:
tags_response = self.sagemaker.list_tags(ResourceArn=ep["EndpointArn"])
tags = {t["Key"]: t["Value"] for t in tags_response.get("Tags", [])}
missing = self.validate_tags(tags)
if missing:
untagged["sagemaker"].append({
"endpoint_name": ep["EndpointName"],
"missing_tags": missing,
})
return untaggedPhase 2: Cost Attribution Engine
AWS Cost Explorer Integration
# finops/cost_attribution.py — Attribute costs to teams and models
import boto3
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List
class MLOpsCostAttribution:
"""Attribute ML infrastructure costs to teams, projects, and models."""
def __init__(self):
self.ce = boto3.client("ce")
def get_costs_by_team(self, start_date: str, end_date: str) -> pd.DataFrame:
"""Get ML costs grouped by team tag."""
response = self.ce.get_cost_and_usage(
TimePeriod={"Start": start_date, "End": end_date},
Granularity="DAILY",
Metrics=["UnblendedCost", "UsageQuantity"],
GroupBy=[
{"Type": "TAG", "Key": "team"},
{"Type": "TAG", "Key": "resource_purpose"},
],
Filter={
"Or": [
{"Dimensions": {"Key": "SERVICE", "Values": ["Amazon SageMaker"]}},
{"Dimensions": {"Key": "SERVICE", "Values": ["Amazon EC2"]}},
{"Tags": {"Key": "resource_purpose", "Values": ["training", "inference", "experimentation", "data-processing"]}},
]
},
)
rows = []
for period in response["ResultsByTime"]:
date = period["TimePeriod"]["Start"]
for group in period["Groups"]:
team = group["Keys"][0].split("$")[1] if "$" in group["Keys"][0] else "untagged"
purpose = group["Keys"][1].split("$")[1] if "$" in group["Keys"][1] else "untagged"
cost = float(group["Metrics"]["UnblendedCost"]["Amount"])
rows.append({
"date": date,
"team": team,
"resource_purpose": purpose,
"cost_usd": cost,
})
return pd.DataFrame(rows)
def get_costs_by_model(self, start_date: str, end_date: str) -> pd.DataFrame:
"""Get costs attributed to specific ML models."""
response = self.ce.get_cost_and_usage(
TimePeriod={"Start": start_date, "End": end_date},
Granularity="MONTHLY",
Metrics=["UnblendedCost"],
GroupBy=[
{"Type": "TAG", "Key": "model_name"},
{"Type": "TAG", "Key": "resource_purpose"},
],
)
rows = []
for period in response["ResultsByTime"]:
for group in period["Groups"]:
model = group["Keys"][0].split("$")[1] if "$" in group["Keys"][0] else "untagged"
purpose = group["Keys"][1].split("$")[1] if "$" in group["Keys"][1] else "untagged"
cost = float(group["Metrics"]["UnblendedCost"]["Amount"])
rows.append({
"model": model,
"purpose": purpose,
"cost_usd": cost,
})
return pd.DataFrame(rows)
def generate_showback_report(self, month: str) -> Dict:
"""Generate a comprehensive showback report for a given month."""
start = f"{month}-01"
end = (datetime.strptime(start, "%Y-%m-%d") + timedelta(days=32)).strftime("%Y-%m-01")
team_costs = self.get_costs_by_team(start, end)
model_costs = self.get_costs_by_model(start, end)
# Summary by team
team_summary = team_costs.groupby("team")["cost_usd"].sum().sort_values(ascending=False)
# Summary by purpose
purpose_summary = team_costs.groupby("resource_purpose")["cost_usd"].sum()
# Top models by cost
model_summary = model_costs.groupby("model")["cost_usd"].sum().sort_values(ascending=False).head(20)
# Untagged resources (cost that can't be attributed)
untagged_cost = team_costs[team_costs["team"] == "untagged"]["cost_usd"].sum()
total_cost = team_costs["cost_usd"].sum()
return {
"month": month,
"total_ml_cost": total_cost,
"team_breakdown": team_summary.to_dict(),
"purpose_breakdown": purpose_summary.to_dict(),
"top_models": model_summary.to_dict(),
"untagged_cost": untagged_cost,
"tagging_coverage": 1 - (untagged_cost / total_cost) if total_cost > 0 else 0,
}Phase 3: Showback Dashboard
BigQuery Cost Analytics
-- BigQuery: ML cost attribution queries
-- Monthly cost by team with trend
SELECT
team,
FORMAT_DATE('%Y-%m', date) as month,
ROUND(SUM(cost_usd), 2) as total_cost,
ROUND(SUM(cost_usd) / LAG(SUM(cost_usd)) OVER (PARTITION BY team ORDER BY date) - 1, 4) as mom_growth
FROM `project.finops.ml_daily_costs`
WHERE date >= DATE_SUB(CURRENT_DATE(), INTERVAL 6 MONTH)
GROUP BY team, month
ORDER BY team, month;
-- Cost per model including full lifecycle (training + inference + storage)
SELECT
model_name,
SUM(CASE WHEN resource_purpose = 'training' THEN cost_usd ELSE 0 END) as training_cost,
SUM(CASE WHEN resource_purpose = 'inference' THEN cost_usd ELSE 0 END) as inference_cost,
SUM(CASE WHEN resource_purpose = 'data-processing' THEN cost_usd ELSE 0 END) as data_cost,
SUM(CASE WHEN resource_purpose = 'experimentation' THEN cost_usd ELSE 0 END) as experimentation_cost,
SUM(cost_usd) as total_cost,
COUNT(DISTINCT date) as active_days
FROM `project.finops.ml_daily_costs`
WHERE date >= DATE_SUB(CURRENT_DATE(), INTERVAL 3 MONTH)
GROUP BY model_name
ORDER BY total_cost DESC
LIMIT 30;
-- Cost per prediction (inference efficiency)
SELECT
model_name,
SUM(cost_usd) as monthly_inference_cost,
SUM(prediction_count) as monthly_predictions,
ROUND(SUM(cost_usd) / SUM(prediction_count) * 1000000, 4) as cost_per_million_predictions
FROM `project.finops.ml_daily_costs` c
JOIN `project.analytics.model_predictions` p
ON c.model_name = p.model_name AND c.date = p.date
WHERE c.resource_purpose = 'inference'
AND c.date >= DATE_SUB(CURRENT_DATE(), INTERVAL 1 MONTH)
GROUP BY model_name
ORDER BY cost_per_million_predictions DESC;Phase 4: Budget Alerts and Governance
Team Budget Enforcement
# finops/budget_alerts.py — Budget alerts and enforcement
from dataclasses import dataclass
from typing import Dict
@dataclass
class TeamBudget:
team: str
monthly_budget_usd: float
alert_thresholds: list # [50%, 75%, 90%, 100%]
class BudgetManager:
"""Monitor and enforce ML budgets per team."""
def __init__(self, budgets: Dict[str, TeamBudget]):
self.budgets = budgets
def check_budgets(self, current_spend: Dict[str, float]) -> list:
"""Check all team budgets and generate alerts."""
alerts = []
for team, budget in self.budgets.items():
spend = current_spend.get(team, 0)
utilization = spend / budget.monthly_budget_usd
for threshold in budget.alert_thresholds:
if utilization >= threshold / 100:
severity = "warning" if threshold < 100 else "critical"
alerts.append({
"team": team,
"severity": severity,
"spend_usd": spend,
"budget_usd": budget.monthly_budget_usd,
"utilization_pct": utilization * 100,
"threshold": threshold,
"message": f"Team {team} has used {utilization*100:.0f}% of monthly ML budget (${spend:,.0f} / ${budget.monthly_budget_usd:,.0f})",
})
break # Only alert on highest breached threshold
return alerts
def enforce_hard_limit(self, team: str, current_spend: float) -> bool:
"""
Check if a team has exceeded their hard budget limit.
Returns True if spending should be blocked.
"""
budget = self.budgets.get(team)
if not budget:
return False
# Hard limit: 110% of budget (10% buffer)
return current_spend > budget.monthly_budget_usd * 1.1Conclusion
FinOps showback for MLOps requires three layers: mandatory resource tagging (so every GPU, endpoint, and storage bucket is attributed to a team and model), automated cost attribution (mapping cloud billing data to teams, projects, and models via tags), and governance (budgets, alerts, and enforcement). The tagging strategy must be enforced at resource creation time—retroactive tagging is unreliable. Cost attribution should flow into a centralized data warehouse (BigQuery, Snowflake) where finance and engineering can query costs by any dimension. Showback dashboards make costs visible to the teams generating them, creating accountability without requiring chargeback. Teams that implement MLOps FinOps typically reduce ML cloud spend by 20-35% within six months—not through draconian cuts, but through informed optimization by teams that can finally see their own bill.
