dlt-doctor 0.4.0

Production Optimization Engine and Quality Accelerator for Databricks DLT, Auto Loader, and PySpark

The Problem • Video • Benchmarks • Architecture • SaaS Comparison • Quickstart • Walkthrough

The Story

Every data engineer working on Databricks knows the exact feeling:

It is 2 AM. Your streaming pipeline crashes with DeltaAnalysisException: Cannot perform Merge as multiple source rows matched. Or worse: native DLT expect_or_drop silently purged 50,000 regulatory records, and nobody noticed until audit season. You spin up an interactive cluster to debug, burning DBUs while waiting 4 minutes for workers to allocate, only to discover that someone chained 8 .withColumn() calls and wrote a row-level Python @udf that disabled Whole-Stage CodeGen and created massive JVM-to-Python socket serialization.

Meanwhile, enterprise SaaS observability tools cost $50,000 to $150,000 annually just to send a Slack alert after corrupted data has already entered downstream Gold tables.

dlt-doctor is the antidote:

• Zero cluster spend: Statically audits your PySpark AST and configurations in under 50ms locally or inside CI before code merges.
• Zero silent data loss: Replaces destructive expect_or_drop with dual-routing quarantine streams.
• Surgical healing: Calculates bounded late-data partition blast radii to restate historical rolling windows with 98.1% less I/O scan than full-table reloads.

Showcase Video

_{15-Second Showcase: Eliminate silent drops, accelerate PySpark execution, and avoid the $100k SaaS tax. Click to view full 1080p video.}

Benchmarks

Tested against a 10M-record transactional dataset on Databricks Runtime 15.4 LTS (4x worker nodes):

Architecture

Why dlt-doctor vs The $100k SaaS Stack

Commercial observability tools (Monte Carlo, Anomalo, Unravel Data, Acceldata) charge enterprise subscriptions to poll metrics post-run. dlt-doctor shifts left: it fixes the code, isolates bad records, and scripts partition-pruned restatements before compute runs.

Terminal Scorecard Preview

Quickstart

1. Installation

pip install dlt-doctor

2. Pre-Commit & GitHub Actions CI Gate

Catch PySpark anti-patterns in pull requests before they merge:

# .github/workflows/pyspark-audit.yml
name: PySpark Quality Gate
on: [pull_request]

jobs:
  audit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run dlt-doctor Quality Audit
        uses: MsMansiDhruv/dlt-doctor@main
        with:
          path: "src/pipelines"
          format: "text"

3. Essential CLI Commands

# Statically audit a pipeline script
dlt-doctor analyze src/pipelines/orders_silver.py

# Calculate late-arriving partition blast radius and emit pruned MERGE SQL
dlt-doctor restate --table main.gold.orders --late-date 2026-10-25 --window 7

# Generate streaming CDC deduplication and SCD merge templates
dlt-doctor cdc --table main.silver.customers --keys customer_id --event-time updated_at

# Audit legacy DBFS mounts and two-tier Hive Metastore references for Unity Catalog
dlt-doctor uc-migrate src/pipelines/orders.py --catalog prod_lakehouse --schema bronze

# Forecast Databricks DBU burn and cluster sizing (Serverless vs Classic DLT)
dlt-doctor cost --files-per-hour 25000 --avg-file-mb 12 --joins 2

Production Walkthrough

Act 1: Pre-Flight Code & Ingestion Health

Catch Catalyst compilation degradation, Auto Loader backpressure, and legacy storage paths before code touches an active cluster:

from dlt_doctor import PlanDoctor, AdvisorDoctor, AutoLoaderDoctor

# 1. Statically audit PySpark code for chained withColumn calls and row-level UDFs
doctor = PlanDoctor()
findings = doctor.analyze_source_code("""
df_clean = (
    df.withColumn("c1", F.col("a"))
      .withColumn("c2", F.col("b"))
      .withColumn("clean_val", my_python_udf(F.col("raw")))
      .repartition(4)
)
""")

report = AdvisorDoctor().compile_report("orders_etl", findings)
# Render scorecard directly inside a Databricks notebook cell
displayHTML(AdvisorDoctor().to_databricks_html(report))

# 2. Audit Auto Loader ingestion stream progress for backpressure
al_doctor = AutoLoaderDoctor()
al_health = al_doctor.diagnose_stream_progress({
    "numInputRows": 50000,
    "inputRowsPerSecond": 1200.0,
    "processedRowsPerSecond": 450.0,  # Falling behind ingestion rate
})
if al_health.is_backpressured:
    print(f"Stream backpressured: Backlog ~{al_health.estimated_backlog_rows:,} rows.")

Act 2: Streaming Execution & Zero-Loss Quality

Replace destructive expect_or_drop and prevent streaming CDC multiple-match merge failures:

import dlt
from dlt_doctor import ExpectationDoctor, ExpectationRule, CdcDoctor

# 1. Dual-Routing Quarantine (Valid -> Gold, Invalid -> Quarantine Table)
exp_doctor = ExpectationDoctor()
rules = [
    ExpectationRule("valid_id", "order_id IS NOT NULL"),
    ExpectationRule("positive_amount", "amount > 0"),
]
plan = exp_doctor.build_quarantine_routing("silver_orders", rules)

@dlt.table(name=plan.target_clean_table)
def silver_orders():
    return dlt.read_stream("bronze_orders").filter(plan.valid_filter_sql)

@dlt.table(name=plan.target_quarantine_table)
def silver_orders_quarantine():
    return (
        dlt.read_stream("bronze_orders")
        .filter(plan.invalid_filter_sql)
        .withColumn("_quarantined_at", F.current_timestamp())
    )

# 2. Streaming CDC Micro-Batch Window Deduplication
cdc = CdcDoctor().generate_cdc_templates(
    target_table="main.silver.customers",
    primary_keys=["customer_id"],
    event_time_col="updated_at",
)
# Inside foreachBatch handler: dedupes micro-batch to eliminate Delta multiple-match exception
print(cdc.microbatch_dedup_code)
print(cdc.scd1_merge_sql)

Act 3: Lakehouse Storage & Surgical Healing

Restate historical rolling metrics with surgical partition pruning and optimize Delta table storage:

from dlt_doctor import RestatementDoctor, ClusterDoctor, ColumnProfile, VacuumDoctor

# 1. Surgical Late Data Restatement (Bounded Blast Radius)
reconciler = RestatementDoctor()
scope = reconciler.calculate_blast_radius(
    target_table="main.gold.fact_orders",
    late_date_iso="2026-10-25",
    window_size_days=7,
    partition_col="order_date",
    match_key="order_id",
)
# Restates only the 7 affected partition days, skipping 358 unaffected days
spark.sql(scope.delta_merge_sql)

# 2. Select Optimal Liquid Clustering (CLUSTER BY) Keys
cluster_rec = ClusterDoctor().advise_clustering(
    table_name="main.gold.fact_orders",
    columns=[
        ColumnProfile("customer_id", "string", distinct_count=25000, filter_frequency_score=0.9),
        ColumnProfile("order_date", "date", distinct_count=1800, filter_frequency_score=0.95),
        ColumnProfile("order_uuid", "string", distinct_count=9900000),  # Rejected: near-unique
    ],
    total_rows=10000000,
)
spark.sql(cluster_rec.alter_cluster_ddl)

# 3. Safe Delta VACUUM Retention Guard
vac_report = VacuumDoctor().evaluate_vacuum_safety(
    table_name="main.silver.orders",
    requested_retention_hours=24,
    active_streaming_readers=2,
    cdf_enabled=True,
)
if vac_report.safety_level.value != "DANGEROUS":
    spark.sql(vac_report.recommended_vacuum_sql)

Verification & Test Suite

The test suite validates all 11 diagnostic engines across Python 3.10, 3.11, and 3.12 with zero external cluster dependencies:

pytest -v
# 51 passed in 0.15s

All source code is audited with ruff and verified against security vulnerabilities with bandit -r src.

License

Distributed under the Apache 2.0 License. See LICENSE for details.