Azure Storage Cost Optimization¶

from azure.storage.blob import BlobServiceClient, BlobClient, StandardBlobTier
from azure.identity import DefaultAzureCredential
from datetime import datetime, timedelta

def tier_old_blobs(storage_account, container_name, days_threshold=30):
    """Automatically tier blobs based on age"""

    account_url = f"https://{storage_account}.blob.core.windows.net"
    credential = DefaultAzureCredential()

    blob_service_client = BlobServiceClient(account_url, credential=credential)
    container_client = blob_service_client.get_container_client(container_name)

    cutoff_date = datetime.now() - timedelta(days=days_threshold)
    total_savings = 0

    for blob in container_client.list_blobs():
        if blob.last_modified < cutoff_date and blob.blob_tier == StandardBlobTier.HOT:
            blob_client = container_client.get_blob_client(blob.name)

            # Tier to Cool
            blob_client.set_standard_blob_tier(StandardBlobTier.COOL)

            # Calculate savings
            size_gb = blob.size / (1024 ** 3)
            monthly_savings = size_gb * (0.0184 - 0.0100)
            total_savings += monthly_savings

            print(f"Tiered {blob.name} ({size_gb:.2f} GB) → Cool")

    print(f"\nTotal Monthly Savings: ${total_savings:.2f}")
    print(f"Annual Savings: ${total_savings * 12:.2f}")

# Execute tiering
tier_old_blobs("storagecsa", "analytics-data", days_threshold=30)

from datetime import datetime

def get_cost_optimized_path(data_class, entity_type, date):
    """Generate storage path optimized for lifecycle policies"""

    year = date.strftime("%Y")
    month = date.strftime("%m")
    day = date.strftime("%d")

    # Organize by data class for different lifecycle policies
    paths = {
        "hot": f"hot-data/{entity_type}/year={year}/month={month}/day={day}",
        "warm": f"warm-data/{entity_type}/year={year}/month={month}",
        "cold": f"cold-data/{entity_type}/year={year}",
        "archive": f"archive-data/{entity_type}/year={year}"
    }

    return paths.get(data_class, paths["warm"])

# Example usage
hot_path = get_cost_optimized_path("hot", "transactions", datetime.now())
print(f"Hot Data Path: {hot_path}")
# Output: hot-data/transactions/year=2024/month=12/day=10

cold_path = get_cost_optimized_path("cold", "historical_sales", datetime(2022, 1, 1))
print(f"Cold Data Path: {cold_path}")
# Output: cold-data/historical_sales/year=2022

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("FileConsolidation").getOrCreate()

# ❌ BAD: Many small files (high transaction costs)
# Reading 10,000 × 1 MB files = 10,000 transactions

# ✅ GOOD: Consolidate into larger files
df = spark.read.parquet("abfss://container@storage.dfs.core.windows.net/small-files/")

df.coalesce(100) \  # Reduce to ~100 files
    .write \
    .mode("overwrite") \
    .parquet("abfss://container@storage.dfs.core.windows.net/consolidated/")

# Result: 100 × 100 MB files = 100 transactions (99% reduction)

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Read uncompressed CSV
df = spark.read \
    .option("header", "true") \
    .csv("abfss://container@storage.dfs.core.windows.net/raw/data.csv")

# Write with optimal compression
df.write \
    .format("parquet") \
    .mode("overwrite") \
    .option("compression", "snappy") \  # or "gzip", "zstd"
    .save("abfss://container@storage.dfs.core.windows.net/compressed/data")

# Measure compression
from pyspark.sql.functions import col, sum as _sum

original_size = spark.read.csv("abfss://container@storage.dfs.core.windows.net/raw/").count()
compressed_files = spark.read.parquet("abfss://container@storage.dfs.core.windows.net/compressed/")

# Compare file sizes via Azure Storage
# Original: ~10 GB
# Compressed: ~2 GB (80% savings)

from delta.tables import DeltaTable

def optimize_and_compress_delta(table_path, zorder_cols=None):
    """Optimize Delta table for cost and performance"""

    delta_table = DeltaTable.forPath(spark, table_path)

    # Get table size before optimization
    detail_before = spark.sql(f"DESCRIBE DETAIL delta.`{table_path}`").first()
    size_before = detail_before.sizeInBytes

    # Optimize with Z-Order
    if zorder_cols:
        delta_table.optimize().executeZOrderBy(zorder_cols)
    else:
        delta_table.optimize().executeCompaction()

    # Vacuum old files
    delta_table.vacuum(retentionHours=168)

    # Get table size after optimization
    detail_after = spark.sql(f"DESCRIBE DETAIL delta.`{table_path}`").first()
    size_after = detail_after.sizeInBytes

    # Calculate savings
    savings_gb = (size_before - size_after) / (1024 ** 3)
    savings_pct = ((size_before - size_after) / size_before) * 100
    monthly_savings = savings_gb * 0.0184  # Hot tier cost

    print(f"Optimization Results for {table_path}:")
    print(f"  Size Before: {size_before / (1024 ** 3):.2f} GB")
    print(f"  Size After: {size_after / (1024 ** 3):.2f} GB")
    print(f"  Storage Savings: {savings_gb:.2f} GB ({savings_pct:.1f}%)")
    print(f"  Monthly Cost Savings: ${monthly_savings:.2f}")

# Run optimization
optimize_and_compress_delta("/mnt/data/sales", zorder_cols=["date", "region"])

from azure.storage.filedatalake import DataLakeServiceClient
from azure.identity import DefaultAzureCredential

def batch_upload_files(storage_account, container, files_to_upload):
    """Batch upload files to minimize transactions"""

    account_url = f"https://{storage_account}.dfs.core.windows.net"
    credential = DefaultAzureCredential()

    service_client = DataLakeServiceClient(account_url, credential=credential)
    file_system_client = service_client.get_file_system_client(container)

    # ❌ BAD: Individual uploads (many transactions)
    # for file in files:
    #     file_client = file_system_client.get_file_client(file)
    #     file_client.upload_data(data, overwrite=True)

    # ✅ GOOD: Batch upload
    for file_path, file_data in files_to_upload.items():
        file_client = file_system_client.get_file_client(file_path)
        file_client.create_file()

        # Upload in chunks
        chunk_size = 4 * 1024 * 1024  # 4 MB
        for i in range(0, len(file_data), chunk_size):
            chunk = file_data[i:i + chunk_size]
            file_client.append_data(chunk, offset=i, length=len(chunk))

        # Flush once at the end
        file_client.flush_data(len(file_data))

# Usage
files = {
    "data/file1.parquet": file1_bytes,
    "data/file2.parquet": file2_bytes,
    "data/file3.parquet": file3_bytes
}

batch_upload_files("storagecsa", "analytics", files)

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
def cached_read_blob(storage_account, container, blob_path):
    """Cache frequently accessed blobs"""

    # Generate cache key
    cache_key = hashlib.md5(f"{storage_account}/{container}/{blob_path}".encode()).hexdigest()

    # Read from storage (only once, then cached)
    account_url = f"https://{storage_account}.blob.core.windows.net"
    blob_service_client = BlobServiceClient(account_url, credential=DefaultAzureCredential())
    blob_client = blob_service_client.get_blob_client(container, blob_path)

    blob_data = blob_client.download_blob().readall()
    return blob_data

# First call: reads from storage
data1 = cached_read_blob("storagecsa", "reference-data", "lookup.csv")

# Subsequent calls: served from cache (no transaction cost)
data2 = cached_read_blob("storagecsa", "reference-data", "lookup.csv")