Cost Optimization: Managing Data Platform Expenses

# Storage Tiering Implementation
import boto3
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List

@dataclass
class StorageTier:
    name: str
    storage_class: str
    cost_per_gb_month: float
    min_access_frequency_days: int

class StorageTierManager:
    """Manage storage tiering across S3 buckets."""

    TIERS = {
        "hot": StorageTier("hot", "STANDARD", 0.023, 0),
        "warm": StorageTier("warm", "STANDARD_IA", 0.0125, 30),
        "cold": StorageTier("cold", "GLACIER", 0.004, 90),
        "deep_archive": StorageTier("deep_archive", "DEEP_ARCHIVE", 0.00099, 365)
    }

    def __init__(self, bucket_name: str, region: str = "us-east-1"):
        self.bucket_name = bucket_name
        self.s3 = boto3.client('s3', region_name=region)

    def transition_objects(self, prefix: str, rules: List[dict]):
        """Apply lifecycle rules for automatic tiering."""
        lifecycle_rules = []
        for rule in rules:
            lifecycle_rules.append({
                "ID": rule["name"],
                "Filter": {"Prefix": prefix},
                "Status": "Enabled",
                "Transitions": [
                    {
                        "Days": tier["days"],
                        "StorageClass": self.TIERS[tier["tier"]].storage_class
                    }
                    for tier in rule["transitions"]
                ],
                "Expiration": rule.get("expiration", {})
            })

        self.s3.put_bucket_lifecycle_configuration(
            Bucket=self.bucket_name,
            LifecycleConfiguration={"Rules": lifecycle_rules}
        )

    def analyze_storage_costs(self, prefix: str = "") -> dict:
        """Analyze current storage distribution and costs."""
        paginator = self.s3.get_paginator('list_objects_v2')
        total_size = 0
        tier_sizes = {tier: 0 for tier in self.TIERS}

        for page in paginator.paginate(Bucket=self.bucket_name, Prefix=prefix):
            for obj in page.get('Contents', []):
                size = obj['Size']
                total_size += size
                storage_class = obj.get('StorageClass', 'STANDARD')

                for tier_name, tier in self.TIERS.items():
                    if tier.storage_class == storage_class:
                        tier_sizes[tier_name] += size
                        break

        costs = {}
        for tier_name, size_bytes in tier_sizes.items():
            size_gb = size_bytes / (1024**3)
            costs[tier_name] = {
                "size_gb": round(size_gb, 2),
                "cost_per_month": round(size_gb * self.TIERS[tier_name].cost_per_gb_month, 2)
            }

        return {
            "total_size_gb": round(total_size / (1024**3), 2),
            "total_cost_per_month": sum(c["cost_per_month"] for c in costs.values()),
            "breakdown": costs
        }

# Usage
manager = StorageTierManager("my-data-lake")
manager.transition_objects("data/events/", [
    {
        "name": "events-tiering",
        "transitions": [
            {"tier": "warm", "days": 30},
            {"tier": "cold", "days": 90},
            {"tier": "deep_archive", "days": 365}
        ],
        "expiration": {"Days": 2555}  # 7 years
    }
])

costs = manager.analyze_storage_costs("data/")
print(f"Total storage: {costs['total_size_gb']} GB")
print(f"Monthly cost: ${costs['total_cost_per_month']}")

# Cost Monitoring Dashboard
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, List
import json

@dataclass
class CostAlert:
    service: str
    threshold_usd: float
    period_days: int
    current_cost: float
    alert_level: str  # info, warning, critical

class CostMonitor:
    """Monitor and alert on data platform costs."""

    def __init__(self, budget_monthly_usd: float = 10000):
        self.budget = budget_monthly_usd
        self.alerts: List[CostAlert] = []

    def check_snowflake_costs(self, warehouse_costs: Dict[str, float]):
        """Check Snowflake warehouse costs against thresholds."""
        total = sum(warehouse_costs.values())

        # Budget alert
        if total > self.budget * 0.9:
            self.alerts.append(CostAlert(
                service="snowflake",
                threshold_usd=self.budget * 0.9,
                period_days=30,
                current_cost=total,
                alert_level="critical" if total > self.budget else "warning"
            ))

        # Individual warehouse alerts
        for warehouse, cost in warehouse_costs.items():
            if cost > self.budget * 0.3:  # Single warehouse > 30% of budget
                self.alerts.append(CostAlert(
                    service=f"snowflake.{warehouse}",
                    threshold_usd=self.budget * 0.3,
                    period_days=30,
                    current_cost=cost,
                    alert_level="warning"
                ))

        return self.alerts

    def generate_cost_report(self, costs: Dict[str, Dict]) -> str:
        """Generate a formatted cost report."""
        report = []
        report.append("=" * 60)
        report.append("DATA PLATFORM COST REPORT")
        report.append(f"Period: {datetime.now().strftime('%Y-%m')}")
        report.append("=" * 60)

        total = 0
        for service, breakdown in costs.items():
            service_total = sum(breakdown.values())
            total += service_total
            report.append(f"\n{service.upper()}")
            for component, cost in breakdown.items():
                report.append(f"  {component}: ${cost:,.2f}")

        report.append(f"\n{'=' * 60}")
        report.append(f"TOTAL: ${total:,.2f}")
        report.append(f"BUDGET: ${self.budget:,.2f}")
        report.append(f"REMAINING: ${self.budget - total:,.2f}")
        report.append(f"UTILIZATION: {total/self.budget*100:.1f}%")

        if self.alerts:
            report.append(f"\nALERTS ({len(self.alerts)}):")
            for alert in self.alerts:
                report.append(f"  [{alert.alert_level.upper()}] {alert.service}: "
                            f"${alert.current_cost:,.2f} > ${alert.threshold_usd:,.2f}")

        return "\n".join(report)

# Usage
monitor = CostMonitor(budget_monthly_usd=10000)
alerts = monitor.check_snowflake_costs({
    "analytics_wh": 2500,
    "etl_wh": 1800,
    "adhoc_wh": 400,
    "ml_wh": 3200
})

report = monitor.generate_cost_report({
    "Snowflake": {"analytics": 2500, "etl": 1800, "adhoc": 400, "ml": 3200},
    "S3": {"storage": 200, "transfer": 50},
    "Databricks": {"clusters": 1500, "storage": 100}
})
print(report)