tutorial

Monitoring dlt (dlt-hub) Data Pipelines with Vigilmon

dlt is the open-source Python library for building data pipelines — but silent extraction failures, schema evolution errors, and destination write failures leave your data warehouse with stale or missing tables. Here's how to monitor dlt pipeline execution, load job health, schema drift, and destination availability with Vigilmon.

dlt (data load tool) is an open-source Python library for building data pipelines that extract from APIs, databases, and files and load into destinations like BigQuery, Snowflake, DuckDB, and Postgres. It handles schema inference, incremental loading, and normalization automatically. When dlt pipelines run on a schedule — loading Salesforce data into your warehouse, syncing API responses to DuckDB, or streaming events into BigQuery — failures are deceptively quiet: an extraction that returns zero rows due to a changed API response structure, a schema evolution error that silently drops columns, or a destination write timeout that rolls back the load without retrying. Vigilmon gives you external monitoring for dlt pipeline execution, load job health, schema drift detection, and destination availability so your data team catches pipeline failures before analysts hit empty tables.

What You'll Set Up

  • dlt pipeline execution health via cron heartbeats
  • Load job success and row count validation
  • Schema drift and migration alerts
  • Destination availability checks
  • Incremental state consistency monitoring

Prerequisites

  • dlt 0.4+ installed (pip install dlt)
  • Pipelines run on a schedule (cron, Airflow, Prefect, or similar)
  • Access to your dlt destination (BigQuery, Snowflake, Postgres, etc.)
  • A free Vigilmon account

Step 1: Monitor Pipeline Execution with Heartbeat Pings

dlt pipelines execute synchronously in a Python script. When the script crashes — due to a network timeout during extraction, an authentication failure against the source API, or an OOM kill during normalization of a large response — the pipeline produces no output and the destination table receives no new rows. Instrument your pipeline script to ping Vigilmon only on success:

  1. In Vigilmon, click Add Monitor → Cron Heartbeat.
  2. Set Expected ping interval to match your pipeline schedule (e.g., 30 minutes).
  3. Copy the heartbeat URL (e.g., https://vigilmon.online/heartbeat/abc123).

Wrap your dlt pipeline:

#!/usr/bin/env python3
# run_salesforce_pipeline.py

import dlt
import requests
import sys
from dlt.sources.rest_api import rest_api_source

HEARTBEAT_URL = "https://vigilmon.online/heartbeat/abc123"

@dlt.source
def salesforce_source(api_key: str = dlt.secrets.value):
    return rest_api_source({
        "client": {
            "base_url": "https://your-instance.salesforce.com/services/data/v58.0/",
            "auth": {"type": "bearer", "token": api_key},
        },
        "resources": [
            {"name": "accounts", "endpoint": "sobjects/Account"},
            {"name": "opportunities", "endpoint": "sobjects/Opportunity"},
        ],
    })

def run():
    pipeline = dlt.pipeline(
        pipeline_name="salesforce_pipeline",
        destination="bigquery",
        dataset_name="salesforce_raw",
    )

    load_info = pipeline.run(salesforce_source())

    # Check for load errors before pinging
    if load_info.has_failed_jobs:
        failed = [j for j in load_info.load_packages[0].jobs.get("failed_jobs", [])]
        raise RuntimeError(f"dlt load failed: {len(failed)} failed jobs")

    requests.get(HEARTBEAT_URL, timeout=10)
    print(f"Pipeline complete: {load_info.asstr(verbosity=1)}")

if __name__ == "__main__":
    try:
        run()
    except Exception as e:
        print(f"Pipeline failed: {e}", file=sys.stderr)
        sys.exit(1)

If the heartbeat stops, Vigilmon alerts within one missed interval — far faster than an analyst noticing a table that stopped updating.

Step 2: Validate Row Counts on Each Load

dlt can complete a load with zero rows if the source returns an empty response — a changed filter, a pagination bug, or an API endpoint that returns HTTP 200 with an empty array. A zero-row load is not an exception; dlt treats it as a successful run. Add row count validation after each load to distinguish real success from silent empty loads:

import dlt
import requests
import json
from datetime import datetime

HEARTBEAT_URL = "https://vigilmon.online/heartbeat/abc123"
MIN_ROWS_PER_LOAD = {
    "accounts": 10,         # Must load at least 10 accounts per run
    "opportunities": 5,     # Must load at least 5 opportunities per run
}
METRICS_LOG = "/var/log/dlt-metrics.jsonl"

def run_with_validation():
    pipeline = dlt.pipeline(
        pipeline_name="salesforce_pipeline",
        destination="bigquery",
        dataset_name="salesforce_raw",
    )

    load_info = pipeline.run(salesforce_source())

    if load_info.has_failed_jobs:
        raise RuntimeError("dlt reported failed load jobs")

    # Inspect row counts per table
    row_counts = {}
    for package in load_info.load_packages:
        for job in package.jobs.get("completed_jobs", []):
            table_name = job.job_file_path.split(".")[-2].split("__")[0]
            row_counts[table_name] = row_counts.get(table_name, 0) + getattr(job, "rows_count", 0)

    violations = []
    for table, min_rows in MIN_ROWS_PER_LOAD.items():
        loaded = row_counts.get(table, 0)
        if loaded < min_rows:
            violations.append(f"{table}: {loaded} rows (min {min_rows})")

    metric = {
        "timestamp": datetime.utcnow().isoformat(),
        "pipeline": "salesforce_pipeline",
        "row_counts": row_counts,
        "violations": violations,
    }
    with open(METRICS_LOG, "a") as f:
        f.write(json.dumps(metric) + "\n")

    if violations:
        raise RuntimeError(f"Row count violations: {'; '.join(violations)}")

    requests.get(HEARTBEAT_URL, timeout=10)
    print(f"Load validated: {row_counts}")

Row count monitoring is especially important for incremental pipelines — if the incremental state cursor gets corrupted or advances too far, the pipeline will stop loading new rows without raising an exception, because "no new rows since cursor" is a valid incremental run result.

Step 3: Monitor Schema Drift and Migration Failures

dlt's automatic schema evolution is powerful but can surface problems: a source API that renames a field causes dlt to add a new column and stop populating the old one, existing reports break silently. A source that starts returning a new nested object causes an unexpected table to be created. Monitor schema changes to catch drift before it reaches downstream BI tools:

#!/bin/bash
# /usr/local/bin/dlt-schema-check.sh

HEARTBEAT_URL="https://vigilmon.online/heartbeat/def456"
SCHEMA_SNAPSHOT="/var/lib/dlt/schema-snapshots/salesforce.json"
DLT_PIPELINE_DIR="$HOME/.dlt/pipelines/salesforce_pipeline"
TIMEOUT=15

# Export current schema from dlt pipeline state
CURRENT_SCHEMA=$(python3 -c "
import dlt, json
pipeline = dlt.pipeline('salesforce_pipeline', destination='bigquery', dataset_name='salesforce_raw')
schema = pipeline.default_schema
# Get column names per table
tables = {}
for table_name, table in schema.tables.items():
    if not table_name.startswith('_dlt'):
        tables[table_name] = sorted(table.get('columns', {}).keys())
print(json.dumps(tables, sort_keys=True))
" 2>/dev/null)

if [ -z "$CURRENT_SCHEMA" ]; then
  echo "Failed to read dlt schema"
  exit 1
fi

if [ ! -f "$SCHEMA_SNAPSHOT" ]; then
  # First run — save snapshot and exit cleanly
  echo "$CURRENT_SCHEMA" > "$SCHEMA_SNAPSHOT"
  curl -s "$HEARTBEAT_URL"
  echo "Schema snapshot created"
  exit 0
fi

PREV_SCHEMA=$(cat "$SCHEMA_SNAPSHOT")

CHANGES=$(python3 -c "
import json, sys

prev = json.loads('''$PREV_SCHEMA''')
curr = json.loads('''$CURRENT_SCHEMA''')

changes = []
for table in set(list(prev.keys()) + list(curr.keys())):
    prev_cols = set(prev.get(table, []))
    curr_cols = set(curr.get(table, []))
    added = curr_cols - prev_cols
    removed = prev_cols - curr_cols
    if added:
        changes.append(f'{table}: +{sorted(added)}')
    if removed:
        changes.append(f'{table}: -{sorted(removed)}')

print('; '.join(changes) if changes else '')
")

if [ -z "$CHANGES" ]; then
  echo "Schema unchanged"
  echo "$CURRENT_SCHEMA" > "$SCHEMA_SNAPSHOT"
  curl -s "$HEARTBEAT_URL"
else
  echo "Schema drift detected: $CHANGES"
  # Update snapshot so next run catches NEW changes, not the same ones
  echo "$CURRENT_SCHEMA" > "$SCHEMA_SNAPSHOT"
  # Don't ping — alert on the missed heartbeat
  exit 1
fi

Set the Vigilmon heartbeat to 35 minutes (just over your pipeline interval). Schema drift alerts aren't always actionable immediately — removed columns need investigation, but added columns from a source API change are often expected. Use the drift log to triage.

Step 4: Monitor Destination Availability

dlt cannot load data if the destination is unavailable. BigQuery token refresh failures, Snowflake warehouse suspension (auto-suspend is on by default), Postgres connection limit exhaustion, or S3 bucket permission changes all cause dlt loads to fail at the write phase — after extraction and normalization have already succeeded, wasting the upstream API call budget. Monitor destination availability independently of pipeline execution:

#!/bin/bash
# /usr/local/bin/dlt-destination-check.sh

HEARTBEAT_URL="https://vigilmon.online/heartbeat/ghi789"
TIMEOUT=20

# Postgres destination check
check_postgres() {
  PGPASSWORD="$DB_PASSWORD" psql \
    -h "$DB_HOST" -p "${DB_PORT:-5432}" -U "$DB_USER" -d "$DB_NAME" \
    -c "SELECT 1" \
    --no-password \
    -t -q 2>/dev/null
  return $?
}

# BigQuery destination check (uses gcloud)
check_bigquery() {
  bq query --nouse_legacy_sql "SELECT 1" > /dev/null 2>&1
  return $?
}

# DuckDB destination check (local file)
check_duckdb() {
  DUCKDB_FILE="${DUCKDB_PATH:-/data/warehouse.duckdb}"
  python3 -c "
import duckdb
con = duckdb.connect('$DUCKDB_FILE', read_only=True)
con.execute('SELECT 1')
con.close()
print('ok')
" > /dev/null 2>&1
  return $?
}

DESTINATION="${DLT_DESTINATION:-postgres}"

case "$DESTINATION" in
  postgres)   check_postgres ;;
  bigquery)   check_bigquery ;;
  duckdb)     check_duckdb ;;
  *)          echo "Unknown destination: $DESTINATION"; exit 1 ;;
esac

if [ $? -eq 0 ]; then
  echo "Destination $DESTINATION is reachable"
  curl -s "$HEARTBEAT_URL"
else
  echo "Destination $DESTINATION unreachable"
  exit 1
fi

Set this check to run every 5 minutes with a Vigilmon heartbeat at the same interval. Catching destination unavailability before the pipeline runs prevents wasted extraction calls and helps distinguish between source problems (the API is down) and destination problems (the warehouse is down).

Step 5: Monitor Incremental State Consistency

dlt's incremental loading relies on a state cursor stored in the destination's _dlt_pipeline_state table. If this table is deleted, the cursor is reset, or the state becomes inconsistent (e.g., after a manual table truncation), dlt will reload all historical data on the next run — potentially causing duplicates and overloading the source API. Monitor state consistency to catch resets before they cause data quality issues:

#!/usr/bin/env python3
# /usr/local/bin/dlt-state-check.py

import dlt
import requests
import json
from datetime import datetime, timezone, timedelta
import sys

HEARTBEAT_URL = "https://vigilmon.online/heartbeat/jkl012"
PIPELINE_NAME = "salesforce_pipeline"
MAX_CURSOR_AGE_HOURS = 25   # Alert if cursor hasn't advanced in 25 hours

pipeline = dlt.pipeline(
    pipeline_name=PIPELINE_NAME,
    destination="bigquery",
    dataset_name="salesforce_raw",
)

state = pipeline.state
sources_state = state.get("sources", {})

if not sources_state:
    print("WARNING: No incremental state found — pipeline may full-reload on next run")
    sys.exit(1)

for source_name, source_state in sources_state.items():
    resources = source_state.get("resources", {})
    for resource_name, resource_state in resources.items():
        cursor = resource_state.get("incremental", {})
        if cursor:
            last_value = cursor.get("last_value")
            print(f"  {source_name}.{resource_name}: cursor at {last_value}")

# Check that state was updated recently by querying the destination state table
try:
    import duckdb  # or replace with psycopg2 / bigquery client
    con = duckdb.connect("/data/warehouse.duckdb", read_only=True)
    result = con.execute(
        "SELECT MAX(created_at) FROM salesforce_raw._dlt_pipeline_state WHERE pipeline_name = ?",
        [PIPELINE_NAME]
    ).fetchone()
    con.close()

    last_state_update = result[0] if result else None
    if last_state_update:
        age_hours = (datetime.now(timezone.utc) - last_state_update).total_seconds() / 3600
        if age_hours <= MAX_CURSOR_AGE_HOURS:
            requests.get(HEARTBEAT_URL, timeout=10)
            print(f"Incremental state OK: last updated {age_hours:.1f}h ago")
        else:
            print(f"Incremental state stale: {age_hours:.1f}h ago (max: {MAX_CURSOR_AGE_HOURS}h)")
            sys.exit(1)
except Exception as e:
    print(f"Could not verify state in destination: {e}")
    sys.exit(1)

Step 6: Set Up Alert Channels

Configure Vigilmon to route dlt pipeline alerts to the right teams:

  1. Go to Alert Channels in Vigilmon.
  2. Add a Slack webhook to #data-platform-alerts — empty tables affect analysts and business stakeholders immediately.
  3. Add PagerDuty for destination availability and pipeline execution monitors — a down destination stops all loads across multiple pipelines.
  4. Add email notifications for schema drift and incremental state monitors — these require investigation, not immediate incident response.
  5. Set Consecutive failures before alert to 1 for row count and execution monitors — zero-row loads and pipeline crashes are always actionable.

Add maintenance windows during dlt version upgrades or schema migrations:

curl -X POST https://vigilmon.online/api/maintenance \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "monitor_id": "DLT_EXECUTION_MONITOR_ID",
    "duration_minutes": 30,
    "reason": "dlt 0.5.x upgrade and schema migration"
  }'

Summary

| Monitor | Target | What It Catches | |---|---|---| | Cron heartbeat (execution) | Pipeline script completion | API auth failure, extraction crash, OOM kill | | Row count validation | Load package row counts | Empty loads from API changes, pagination bugs | | Cron heartbeat (schema drift) | Column diff against snapshot | Source API field renames, unexpected new tables | | Cron heartbeat (destination) | Destination connectivity check | Warehouse suspension, credential rotation, networking | | Cron heartbeat (state) | _dlt_pipeline_state age | Cursor reset, table truncation, full-reload risk |

dlt's automatic schema handling and transparent incremental loading are its biggest strengths — and the source of its most subtle failure modes. A schema evolution that silently drops a field, an incremental cursor that stops advancing, or a zero-row load that the library treats as success are all issues that require external observability to catch. Vigilmon's external heartbeat monitoring gives your data team that visibility: you know about dlt pipeline failures in minutes, not when an analyst asks why a table stopped updating.

Start monitoring for free at vigilmon.online.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles