Data Observability with Monte Carlo — Anomaly Detection, Lineage, and SLAs

Why Traditional Data Monitoring Falls Short

Infrastructure monitoring — CPU, memory, query latency, job success rates — tells you whether your pipelines ran, not whether the data they produced is correct. A pipeline can complete successfully and write 40% fewer rows than yesterday, silently drop a dimension value, or calculate revenue figures an order of magnitude too high due to a fanout join. None of these anomalies trigger an infrastructure alert.

Data observability closes this gap by monitoring the data itself: its freshness, volume, statistical distributions, schema structure, and lineage. Monte Carlo operationalizes these five pillars with ML-based anomaly detection that learns normal behavior per table rather than requiring you to hand-configure thresholds for thousands of tables. The data quality observability guide covers complementary approaches including dbt tests, Great Expectations contracts, and freshness SLAs that integrate alongside Monte Carlo — the two layers catch different failure modes and are most effective together.

The Five Pillars of Data Observability

Monte Carlo's observability model covers five dimensions that together provide complete coverage of data health. Each pillar is monitored automatically after warehouse connection, with no manual schema annotation required.

Warehouse Connection and Permissions

Monte Carlo connects to your warehouse via a dedicated service account with read-only metadata permissions. For Snowflake, this means granting access to ACCOUNT_USAGE and INFORMATION_SCHEMA. For BigQuery, the service account needs bigquery.jobs.listAll, bigquery.tables.getIamPolicy, and bigquery.tables.listat the project level. The collector agent runs inside your VPC or cloud environment — Monte Carlo does not pull raw data; only metadata and query logs transit to Monte Carlo's SaaS backend.

-- Snowflake: create the Monte Carlo service role
CREATE ROLE montecarlo_role;

-- Metadata access for freshness and schema monitoring
GRANT IMPORTED PRIVILEGES ON DATABASE snowflake TO ROLE montecarlo_role;

-- Table sampling access for distribution monitoring
-- (only needed if enabling field-level health monitors)
GRANT USAGE ON WAREHOUSE analytics_wh TO ROLE montecarlo_role;
GRANT USAGE ON DATABASE analytics TO ROLE montecarlo_role;
GRANT USAGE ON ALL SCHEMAS IN DATABASE analytics TO ROLE montecarlo_role;
GRANT SELECT ON ALL TABLES IN DATABASE analytics TO ROLE montecarlo_role;
GRANT SELECT ON FUTURE TABLES IN DATABASE analytics TO ROLE montecarlo_role;

-- Assign to service account user
CREATE USER montecarlo_svc
  PASSWORD = '<rotate_via_vault>'
  DEFAULT_ROLE = montecarlo_role
  MUST_CHANGE_PASSWORD = FALSE;
GRANT ROLE montecarlo_role TO USER montecarlo_svc;

# BigQuery: Terraform service account and IAM bindings
resource "google_service_account" "montecarlo" {
  account_id   = "montecarlo-collector"
  display_name = "Monte Carlo Observability Collector"
}

# Metadata and lineage (query log) access
resource "google_project_iam_member" "montecarlo_jobs" {
  project = var.project_id
  role    = "roles/bigquery.resourceViewer"
  member  = "serviceAccount:${google_service_account.montecarlo.email}"
}

resource "google_project_iam_member" "montecarlo_jobs_list" {
  project = var.project_id
  role    = "roles/bigquery.jobUser"
  member  = "serviceAccount:${google_service_account.montecarlo.email}"
}

# Field-level health monitoring (table sampling)
resource "google_project_iam_member" "montecarlo_data_viewer" {
  project = var.project_id
  role    = "roles/bigquery.dataViewer"
  member  = "serviceAccount:${google_service_account.montecarlo.email}"
}

ML-Based Anomaly Detection and Custom Monitors

Monte Carlo's automated monitors train a time-series model per table per metric (row count, null rate, mean, etc.) using 14 days of history. The model accounts for daily and weekly seasonality, trending growth curves, and known incident periods that should be excluded from training. Sensitivity controls the width of the anomaly band: low sensitivity (1.5σ equivalent) produces fewer alerts with higher confidence; high sensitivity (0.75σ equivalent) catches subtle shifts but generates more noise for tables with inherently variable data.

Custom monitors supplement automated detection for business-logic rules the ML model cannot infer — a revenue column that must never go negative, an orders table that must have at least 1000 rows by 9 AM UTC, or a referential integrity check between two tables. Data pipeline testing covers the complementary layer of contract-based validation with Great Expectations and schema registries that run at pipeline execution time rather than after the fact — combining both gives you shift detection and rule enforcement.

# Monte Carlo Python SDK — create custom monitors programmatically
# pip install pycarlo

from pycarlo.core import Client, Session

session = Session(mcd_id="your-mcd-id", mcd_token="your-mcd-token")
client = Client(session=session)

# --- Metric-based monitor: row count must be >= 1000 by 9 AM UTC ---
create_monitor_mutation = """
mutation CreateCustomMetricMonitor(
  $dwId: UUID!,
  $fullTableId: String!,
  $description: String!,
  $monitorType: String!,
  $aggTimeInterval: String,
  $aggSelect: String,
  $whereCondition: String,
  $comparisons: [MetricMonitorComparisonInput!]!
) {
  createOrUpdateMonitor(
    dwId: $dwId,
    fullTableId: $fullTableId,
    description: $description,
    monitorType: $monitorType,
    aggTimeInterval: $aggTimeInterval,
    aggSelect: $aggSelect,
    whereCondition: $whereCondition,
    comparisons: $comparisons
  ) {
    monitor { uuid }
  }
}
"""

response = client(
  create_monitor_mutation,
  variables={
    "dwId": "your-warehouse-uuid",
    "fullTableId": "analytics:production.orders",
    "description": "Orders table must have >= 1000 rows by 09:00 UTC",
    "monitorType": "CUSTOM_METRIC",
    "aggTimeInterval": "DAY",
    "aggSelect": "COUNT(*)",
    "whereCondition": "created_at >= TIMESTAMP_TRUNC(CURRENT_TIMESTAMP(), DAY)",
    "comparisons": [
      {
        "comparisonType": "THRESHOLD",
        "operator": "GTE",
        "threshold": 1000,
        "isThresholdRelative": False,
      }
    ],
  },
)
print(response["createOrUpdateMonitor"]["monitor"]["uuid"])

# --- Field health monitor: revenue must never be negative ---
field_monitor_mutation = """
mutation {
  createOrUpdateMonitor(
    dwId: "your-warehouse-uuid",
    fullTableId: "analytics:production.orders",
    description: "Revenue column must be non-negative",
    monitorType: "FIELD_HEALTH",
    fields: ["revenue_usd"],
    comparisons: [{
      comparisonType: "THRESHOLD",
      metric: "MIN",
      operator: "GTE",
      threshold: 0
    }]
  ) { monitor { uuid } }
}
"""
client(field_monitor_mutation)

# Query active incidents via the Monte Carlo API
list_incidents_query = """
query ListIncidents($first: Int, $after: String) {
  getIncidents(first: $first, after: $after) {
    edges {
      node {
        id
        status
        severity
        title
        createdTime
        tables {
          fullTableId
        }
        anomalies {
          dimension
          metric
          value
          expectedValue
        }
      }
    }
    pageInfo { endCursor hasNextPage }
  }
}
"""

page_cursor = None
all_incidents = []

while True:
    response = client(
        list_incidents_query,
        variables={"first": 50, "after": page_cursor},
    )
    data = response["getIncidents"]
    all_incidents.extend(data["edges"])
    if not data["pageInfo"]["hasNextPage"]:
        break
    page_cursor = data["pageInfo"]["endCursor"]

# Filter to open, high-severity incidents touching the revenue domain
open_critical = [
    e["node"] for e in all_incidents
    if e["node"]["status"] == "ACTIVE"
    and any("revenue" in t["fullTableId"] for t in e["node"]["tables"])
]
print(f"Open critical incidents in revenue domain: {len(open_critical)}")

Field-Level Lineage — Tracing Data from Source to Dashboard

Table-level lineage tells you that Table B depends on Table A. Field-level lineage tells you that the revenue_usd column in the fct_orders mart is computed from amount_cents / 100 in stg_orders, which originates from raw.stripe_charges.amount. When a schema change alert fires on the raw source, Monte Carlo surfaces the full downstream blast radius — exactly which dashboards, dbt models, and reverse ETL syncs depend on the changed field — within the incident card.

Monte Carlo extracts field-level lineage by parsing SQL SELECT expressions from query history. It handles CTEs, subqueries, UNION statements, and window functions. For dbt projects, Monte Carlo additionally ingests the dbt manifest to link dbt model names to their compiled SQL and warehouse table identities, giving you lineage annotations that include the dbt model name, owner, and test coverage alongside the SQL path. OpenTelemetry for Data Pipelines covers how distributed tracing complements lineage by capturing pipeline execution spans, enabling cross-system correlation between orchestrator run IDs and warehouse query IDs — linking Monte Carlo incidents to the specific Airflow DAG run that produced the anomalous data.

# Query field-level lineage for a specific column
lineage_query = """
query FieldLineage($fullTableId: String!, $field: String!) {
  getFieldLineage(fullTableId: $fullTableId, field: $field) {
    upstream {
      tableId
      field
      queryHash
    }
    downstream {
      tableId
      field
      assetType   # TABLE, BI_DASHBOARD, REVERSE_ETL
    }
  }
}
"""

result = client(
    lineage_query,
    variables={
        "fullTableId": "analytics:production.fct_orders",
        "field": "revenue_usd",
    },
)

lineage = result["getFieldLineage"]
print("Upstream sources:")
for node in lineage["upstream"]:
    print(f"  {node['tableId']}.{node['field']}")

print("\nDownstream consumers:")
for node in lineage["downstream"]:
    print(f"  [{node['assetType']}] {node['tableId']}.{node['field']}")

Data SLAs — Freshness, Volume, and Quality Agreements

Data SLAs in Monte Carlo define the expected state of a table at a specific time — the orders mart must be fresh by 07:00 UTC or the revenue table must have at least 500 rows per daily partition. SLA breaches generate incidents independently of anomaly detection, ensuring that pipeline delays surface as actionable alerts even when the eventual data is statistically normal.

# Monte Carlo SLA configuration via the pycarlo SDK
create_sla_mutation = """
mutation CreateSla(
  $dwId: UUID!,
  $fullTableId: String!,
  $slaType: String!,
  $slaConfig: SlaConfigInput!
) {
  createSla(
    dwId: $dwId,
    fullTableId: $fullTableId,
    slaType: $slaType,
    slaConfig: $slaConfig
  ) { sla { uuid slaType } }
}
"""

# Freshness SLA: orders mart must update by 07:00 UTC daily
client(
    create_sla_mutation,
    variables={
        "dwId": "your-warehouse-uuid",
        "fullTableId": "analytics:production.fct_orders",
        "slaType": "FRESHNESS",
        "slaConfig": {
            "breachRules": [
                {
                    "windowDuration": "PT7H",   # breach if not updated by 07:00 UTC
                    "windowOffset": "PT0H",      # relative to midnight UTC
                    "timezone": "UTC",
                }
            ],
            "notificationChannels": [
                {"channelType": "SLACK", "channelId": "C0123DATATEAM"},
                {"channelType": "PAGERDUTY", "integrationKey": "pd-orders-oncall"},
            ],
        },
    },
)

# Volume SLA: at least 10,000 rows per daily partition
client(
    create_sla_mutation,
    variables={
        "dwId": "your-warehouse-uuid",
        "fullTableId": "analytics:production.fct_orders",
        "slaType": "VOLUME",
        "slaConfig": {
            "breachRules": [
                {
                    "threshold": 10000,
                    "operator": "GTE",
                    "windowDuration": "P1D",
                }
            ],
            "notificationChannels": [
                {"channelType": "SLACK", "channelId": "C0123DATATEAM"},
            ],
        },
    },
)

dbt Integration — Model Lineage, Tests, and Freshness Bridging

Monte Carlo's dbt integration ingests the compiled manifest.json and run_results.json artifacts from your CI/CD pipeline. This enriches the lineage graph with dbt model metadata — model name, owner (from meta.owner), tags, description, and dbt test results — and links dbt model identities to their warehouse table identities for unified incident attribution.

# Upload dbt artifacts after each dbt run in CI
# Install: pip install 'pycarlo[dbt]'

from pycarlo.features.dbt import DbtHelper

helper = DbtHelper(
    mcd_id="your-mcd-id",
    mcd_token="your-mcd-token",
)

# Upload manifest and run_results after dbt run
helper.upload(
    manifest_path="target/manifest.json",
    run_results_path="target/run_results.json",
    project_name="analytics",       # matches your dbt_project.yml name
    environment="production",       # prod vs staging namespace
)

print("dbt artifacts uploaded to Monte Carlo")

# GitHub Actions: upload dbt artifacts to Monte Carlo after production run
name: dbt Production Run

on:
  schedule:
    - cron: "0 4 * * *"   # 04:00 UTC daily
  workflow_dispatch:

jobs:
  dbt-run:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: pip install dbt-bigquery "pycarlo[dbt]"

      - name: Run dbt
        env:
          DBT_PROFILES_DIR: .
          GOOGLE_APPLICATION_CREDENTIALS_JSON: ${{ secrets.GCP_SA_KEY }}
        run: |
          echo "$GOOGLE_APPLICATION_CREDENTIALS_JSON" > /tmp/sa.json
          export GOOGLE_APPLICATION_CREDENTIALS=/tmp/sa.json
          dbt run --target prod --select tag:daily
          dbt test --target prod --select tag:daily

      - name: Upload artifacts to Monte Carlo
        env:
          MCD_ID: ${{ secrets.MONTE_CARLO_MCD_ID }}
          MCD_TOKEN: ${{ secrets.MONTE_CARLO_MCD_TOKEN }}
        run: |
          python - <<'EOF'
          from pycarlo.features.dbt import DbtHelper
          import os
          helper = DbtHelper(
              mcd_id=os.environ["MCD_ID"],
              mcd_token=os.environ["MCD_TOKEN"],
          )
          helper.upload(
              manifest_path="target/manifest.json",
              run_results_path="target/run_results.json",
              project_name="analytics",
              environment="production",
          )
          EOF

Airflow Integration — Tagging Failed Runs as Incidents

Monte Carlo's Airflow provider package installs an on_failure_callback that creates a Monte Carlo incident when a DAG task fails, linking the incident to all downstream tables in the lineage graph. This bridges the pipeline execution layer and the data observability layer — when an operator queries the incident list, they see both the Airflow task failure and any resulting data anomalies as a unified incident thread.

# pip install apache-airflow-providers-monte-carlo

from datetime import datetime
from airflow.decorators import dag, task
from monte_carlo_provider.operators.monte_carlo import MonteCarloIncidentCallback

# Attach the callback at DAG level — applies to all tasks
@dag(
    schedule="0 3 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={
        "on_failure_callback": MonteCarloIncidentCallback(
            mcd_id="{{ var.value.monte_carlo_mcd_id }}",
            mcd_token="{{ var.value.monte_carlo_mcd_token }}",
            incident_severity="SEV2",
            affected_table_ids=[
                "analytics:production.fct_orders",
                "analytics:production.fct_revenue",
            ],
        ),
    },
)
def daily_revenue_pipeline():

    @task
    def extract_stripe_charges():
        # ... extract logic
        return {"rows": 12500}

    @task
    def transform_orders(extract_result: dict):
        # ... transformation logic
        pass

    @task
    def load_revenue_mart(transform_result):
        # ... load logic
        pass

    load_revenue_mart(transform_orders(extract_stripe_charges()))

daily_revenue_pipeline()

Circuit Breakers — Stopping Downstream Consumption of Bad Data

A circuit breaker is a monitor that pauses downstream pipeline execution when an anomaly is detected, preventing bad data from propagating to BI tools, reverse ETL targets, or ML feature stores. Monte Carlo implements circuit breakers as webhook callbacks that Airflow sensors can poll: the sensor blocks the downstream DAG until Monte Carlo signals that the upstream table is healthy.

# Airflow circuit breaker pattern using the Monte Carlo table health sensor
from monte_carlo_provider.sensors.monte_carlo import MonteCarloTableHealthSensor
from airflow.decorators import dag, task
from datetime import datetime

@dag(
    schedule="0 8 * * *",   # runs after the upstream revenue pipeline (03:00 UTC)
    start_date=datetime(2026, 1, 1),
    catchup=False,
)
def downstream_revenue_export():

    # Block until Monte Carlo confirms fct_orders is healthy
    wait_for_healthy_orders = MonteCarloTableHealthSensor(
        task_id="wait_for_fct_orders_health",
        mcd_id="{{ var.value.monte_carlo_mcd_id }}",
        mcd_token="{{ var.value.monte_carlo_mcd_token }}",
        full_table_id="analytics:production.fct_orders",
        poke_interval=300,    # check every 5 minutes
        timeout=7200,         # fail after 2 hours if still unhealthy
        mode="reschedule",    # release worker between checks
        # health_check_types defines which monitors must pass
        health_check_types=["FRESHNESS", "VOLUME", "FIELD_HEALTH"],
    )

    @task
    def export_to_salesforce():
        # ... Salesforce reverse ETL
        pass

    wait_for_healthy_orders >> export_to_salesforce()

downstream_revenue_export()

Production Checklist