Event Hubs Capture Cost Optimization¶

from azure.eventhub import EventHubConsumerClient
from azure.identity import DefaultAzureCredential
from datetime import datetime

class ConditionalCaptureManager:
    def __init__(self, resource_group, namespace, event_hub):
        self.resource_group = resource_group
        self.namespace = namespace
        self.event_hub = event_hub

    def should_capture(self):
        """Determine if capture should be enabled based on business hours"""
        now = datetime.now()
        # Capture only during business hours (9 AM - 6 PM weekdays)
        if now.weekday() >= 5:  # Weekend
            return False
        if now.hour < 9 or now.hour >= 18:  # Outside business hours
            return False
        return True

    def toggle_capture(self, enable):
        """Enable or disable capture based on schedule"""
        import subprocess

        cmd = f"""
        az eventhubs eventhub update \
            --resource-group {self.resource_group} \
            --namespace-name {self.namespace} \
            --name {self.event_hub} \
            --enable-capture {str(enable).lower()}
        """
        subprocess.run(cmd, shell=True, check=True)

# Azure Function or Logic App to run hourly
def main():
    manager = ConditionalCaptureManager(
        resource_group="rg-streaming",
        namespace="eh-namespace",
        event_hub="event-hub-name"
    )

    enable_capture = manager.should_capture()
    manager.toggle_capture(enable_capture)

import avro.datafile
import avro.io
import os

def analyze_capture_files(storage_path):
    """Analyze captured Avro files for compression efficiency"""

    total_original_size = 0
    total_compressed_size = 0

    for root, dirs, files in os.walk(storage_path):
        for file in files:
            if file.endswith('.avro'):
                file_path = os.path.join(root, file)
                compressed_size = os.path.getsize(file_path)

                # Read Avro file to estimate original size
                with avro.datafile.DataFileReader(
                    open(file_path, 'rb'),
                    avro.io.DatumReader()
                ) as reader:
                    record_count = sum(1 for _ in reader)
                    avg_record_size = compressed_size / record_count if record_count > 0 else 0
                    estimated_original = record_count * avg_record_size * 2  # Rough estimate

                    total_original_size += estimated_original
                    total_compressed_size += compressed_size

    compression_ratio = (1 - total_compressed_size / total_original_size) * 100
    print(f"Compression Ratio: {compression_ratio:.2f}%")
    print(f"Total Original Size: {total_original_size / (1024**3):.2f} GB")
    print(f"Total Compressed Size: {total_compressed_size / (1024**3):.2f} GB")
    print(f"Storage Savings: ${(total_original_size - total_compressed_size) / (1024**3) * 0.0184:.2f}/month")

# Run analysis
analyze_capture_files("/mnt/capture-analysis")

from pyspark.sql import SparkSession

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

# Read Avro captures
avro_df = spark.read \
    .format("avro") \
    .load("abfss://eventhub-capture@stcapture.dfs.core.windows.net/*/2024/12/*/*.avro")

# Convert to Parquet for long-term storage
avro_df.write \
    .format("parquet") \
    .mode("overwrite") \
    .option("compression", "snappy") \
    .partitionBy("year", "month", "day") \
    .save("abfss://analytics@stcapture.dfs.core.windows.net/captured-events/")

# Cost savings: 20-30% additional compression

from azure.monitor.query import LogsQueryClient, MetricsQueryClient
from azure.identity import DefaultAzureCredential
from datetime import timedelta

def analyze_capture_efficiency(resource_id):
    """Analyze Event Hub capture efficiency and costs"""

    credential = DefaultAzureCredential()
    metrics_client = MetricsQueryClient(credential)

    # Query metrics for last 30 days
    metrics_data = metrics_client.query_resource(
        resource_id,
        metric_names=["CapturedBytes", "IncomingBytes"],
        timespan=timedelta(days=30)
    )

    captured_bytes = sum([m.value for m in metrics_data.metrics[0].timeseries[0].data])
    incoming_bytes = sum([m.value for m in metrics_data.metrics[1].timeseries[0].data])

    capture_ratio = captured_bytes / incoming_bytes * 100
    monthly_capture_cost = (captured_bytes / (1024**3)) * 0.10
    monthly_storage_cost = (captured_bytes / (1024**3)) * 0.0184

    print(f"Capture Efficiency Analysis:")
    print(f"  Incoming Data: {incoming_bytes / (1024**3):.2f} GB")
    print(f"  Captured Data: {captured_bytes / (1024**3):.2f} GB")
    print(f"  Capture Ratio: {capture_ratio:.2f}%")
    print(f"  Monthly Capture Cost: ${monthly_capture_cost:.2f}")
    print(f"  Monthly Storage Cost: ${monthly_storage_cost:.2f}")
    print(f"  Total Monthly Cost: ${monthly_capture_cost + monthly_storage_cost:.2f}")

    return {
        "capture_ratio": capture_ratio,
        "monthly_cost": monthly_capture_cost + monthly_storage_cost
    }

# Run analysis
resource_id = "/subscriptions/{sub}/resourceGroups/rg-streaming/providers/Microsoft.EventHub/namespaces/eh-namespace"
result = analyze_capture_efficiency(resource_id)