Monitoring Best Practices¶

from applicationinsights import TelemetryClient
from opencensus.ext.azure.log_exporter import AzureLogHandler
import logging

class PipelineMetrics:
    """Custom metrics for data pipelines."""

    def __init__(self, instrumentation_key: str):
        self.client = TelemetryClient(instrumentation_key)
        self.logger = logging.getLogger(__name__)
        self.logger.addHandler(AzureLogHandler(
            connection_string=f"InstrumentationKey={instrumentation_key}"
        ))

    def track_pipeline_run(self, pipeline_name: str, duration_ms: float, status: str):
        """Track pipeline execution metrics."""
        self.client.track_metric(
            name="PipelineDuration",
            value=duration_ms,
            properties={
                "pipeline": pipeline_name,
                "status": status
            }
        )

    def track_data_quality(self, table_name: str, null_rate: float, duplicate_rate: float):
        """Track data quality metrics."""
        self.client.track_metric(name="NullRate", value=null_rate, properties={"table": table_name})
        self.client.track_metric(name="DuplicateRate", value=duplicate_rate, properties={"table": table_name})

    def flush(self):
        """Ensure all metrics are sent."""
        self.client.flush()

# In Databricks notebook
from pyspark.sql.functions import *

def publish_job_metrics(job_name: str, metrics: dict):
    """Publish custom metrics to Azure Monitor."""
    from azure.monitor.ingestion import LogsIngestionClient
    from azure.identity import DefaultAzureCredential

    credential = DefaultAzureCredential()
    client = LogsIngestionClient(
        endpoint="https://dce-xxx.eastus-1.ingest.monitor.azure.com",
        credential=credential
    )

    log_entry = [{
        "TimeGenerated": datetime.utcnow().isoformat(),
        "JobName": job_name,
        **metrics
    }]

    client.upload(
        rule_id="dcr-xxx",
        stream_name="Custom-DataPlatformMetrics_CL",
        logs=log_entry
    )