Data Observability Setup

Data Observability Setup

Complete observability framework for Databricks data pipelines — lineage tracking, anomaly detection, SLA monitoring, and alerting.

By Datanest Digital | Version 1.0.0 | $49

What You Get

• Data Lineage Tracking — Trace every record from source through transformations to target, stored in Delta
• Metric Collection — Automated capture of row counts, durations, data volumes, and error rates
• Statistical Anomaly Detection — Z-score, IQR, and moving average methods to catch pipeline drift
• Freshness Monitoring — SLA-aware data freshness checks with breach detection
• Alert Routing — Multi-channel alerting via Slack, PagerDuty, and email with severity-based routing
• Dashboard-Ready Data — Pre-aggregated health scores and metrics for observability dashboards

File Tree

data-observability-setup/
├── README.md
├── manifest.json
├── LICENSE
├── src/
│   ├── lineage_tracker.py          # Source→transform→target lineage
│   ├── metric_collector.py         # Pipeline metric collection
│   ├── anomaly_detector.py         # Statistical anomaly detection
│   ├── freshness_monitor.py        # Data freshness & SLA monitoring
│   ├── alert_manager.py            # Multi-channel alert routing
│   └── dashboard_data.py           # Dashboard aggregation & health scores
├── configs/
│   ├── observability_config.yaml   # Main configuration
│   └── alert_rules.yaml            # Alert rule definitions
├── notebooks/
│   ├── observability_dashboard.py  # Health & lineage dashboard
│   └── setup_observability.py      # Initialize observability tables
├── tests/
│   ├── conftest.py                 # Shared fixtures
│   ├── test_anomaly_detector.py    # Anomaly detection tests
│   └── test_freshness_monitor.py   # SLA breach tests
└── guides/
    └── data-observability-guide.md # Observability strategy guide

Getting Started

Follow this guide to get data observability setup up and running in your environment.

Getting Started

1. Initialize Observability Tables

Run the setup notebook in your Databricks workspace to create the required Delta tables:

# In Databricks — run notebooks/setup_observability.py
# Creates: observability.lineage, observability.metrics,
#          observability.alerts, observability.anomalies

2. Track Lineage in Your Pipelines

from src.lineage_tracker import LineageTracker

tracker = LineageTracker(catalog="main", schema="observability")
tracker.register_source("raw_orders", "s3://lake/raw/orders/")
tracker.register_transformation(
    "clean_orders",
    sql="SELECT * FROM raw_orders WHERE status != 'cancelled'",
    input_datasets=["raw_orders"],
)
tracker.register_target("curated_orders", "main.curated.orders")
tracker.commit_lineage(pipeline_name="orders_etl")

3. Collect Pipeline Metrics

from src.metric_collector import MetricCollector

collector = MetricCollector(config_path="configs/observability_config.yaml")
with collector.track_pipeline("orders_etl"):
    # ... your pipeline code ...
    collector.record_metric("orders_etl", "row_count", 125430)
# Duration is recorded automatically

4. Enable Anomaly Detection

from src.anomaly_detector import AnomalyDetector

detector = AnomalyDetector(method="zscore", threshold=3.0)
anomalies = detector.detect(
    pipeline_name="orders_etl",
    metric_name="row_count",
    lookback_days=30,
)
# Returns list of AnomalyResult with severity, score, details

5. Monitor Freshness and Route Alerts

from src.freshness_monitor import FreshnessMonitor
from src.alert_manager import AlertManager

monitor = FreshnessMonitor(config_path="configs/observability_config.yaml")
breaches = monitor.check_all_slas()

alerts = AlertManager(config_path="configs/observability_config.yaml")
for breach in breaches:
    alerts.send(
        severity="critical",
        title=f"SLA Breach: {breach.table_name}",
        message=f"Data is {breach.hours_stale:.1f}h stale (SLA: {breach.sla_hours}h)",
    )

Architecture

┌─────────────┐    ┌──────────────────┐    ┌───────────────────┐
│  Data        │───▶│  Lineage         │───▶│  Delta Tables     │
│  Pipelines   │    │  Tracker         │    │  (lineage store)  │
└──────┬───────┘    └──────────────────┘    └───────────────────┘
       │
       ▼
┌──────────────┐    ┌──────────────────┐    ┌───────────────────┐
│  Metric      │───▶│  Anomaly         │───▶│  Alert            │
│  Collector   │    │  Detector        │    │  Manager          │
└──────────────┘    └──────────────────┘    └────────┬──────────┘
                                                     │
       ┌──────────────────┐                          ▼
       │  Freshness       │              ┌───────────────────────┐
       │  Monitor         │─────────────▶│  Slack / PagerDuty /  │
       └──────────────────┘              │  Email                │
                                         └───────────────────────┘
       ┌──────────────────┐
       │  Dashboard       │◀──── Aggregated metrics & health scores
       │  Data            │
       └──────────────────┘