dlt-doctor 0.5.0

Production Optimization Engine and Quality Accelerator for Databricks Lakeflow Spark Declarative Pipelines, DLT, and PySpark

Documentation • The Problem • Video • Benchmarks • Where It Fits • Diagnostic Rules • Quickstart • Walkthrough

The Story

Every data engineer working on Databricks Lakeflow Spark Declarative Pipelines (formerly Delta Live Tables) 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 dropped 50,000 regulatory records with no record-level trace, 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.

dlt-doctor is the antidote:

• Zero cluster spend: Statically audits your PySpark AST and configurations locally or inside CI before code merges.
• Helps prevent silent data loss: Intercepts destructive expect_or_drop policies and synthesizes complete, disjoint, three-valued-logic NULL-safe single-pass quarantine streams via dlt-doctor heal.
• Surgical healing: Calculates bounded late-data partition blast radii to restate historical rolling windows without full-table scans.

Showcase Video

_{15-Second Showcase: Eliminate silent drops, accelerate PySpark execution, and shift quality left. Click to view full 1080p video.}

Benchmarks

Static Analyzer Throughput (Locally Measured)

Measured on an AMD Ryzen 7 5800H (16 vCPUs, 3.2GHz, 16GB RAM) running Python 3.11 with zero JVM cluster overhead across 25 production-scale pipeline files:

Where dlt-doctor Fits

Diagnostic Rules

All diagnostic rules are managed in dlt_doctor.rules and queryable via dlt-doctor rules and dlt-doctor explain <RULE_ID>.

Terminal Scorecard Preview

Quickstart

Interactive Walkthrough Notebook

Import notebooks/01_dlt_doctor_walkthrough.ipynb directly into your Databricks workspace to run the interactive cell magic and AST optimizers live on your cluster.

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@v0.5.0
        with:
          path: "src/pipelines"
          format: "text"

3. Essential CLI Commands

# Statically audit a pipeline script, SQL file, notebook (.ipynb), or directory
dlt-doctor analyze src/pipelines/orders_silver.py
dlt-doctor analyze notebooks/01_ingest.ipynb
dlt-doctor analyze src/pipelines/orders.sql
dlt-doctor analyze src/pipelines --format markdown

# Auto-heal expect_or_drop anti-patterns into single-pass quarantine streams (preview diff)
dlt-doctor heal src/pipelines/orders_silver.py

# Apply healed quarantine routing directly to file or directory
dlt-doctor heal src/pipelines/orders_silver.py --write
dlt-doctor heal src/pipelines/ --write

# CI Compliance Gate: Check files or entire directory for unhealed drop rules
# Exit codes: 0 = Compliant, 1 = Needs healing, 2 = Usage/path error, 3 = Cannot heal
dlt-doctor heal src/pipelines/ --check

# List and explain diagnostic rules
dlt-doctor rules
dlt-doctor explain DD001

# 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 --source main.silver.orders_recalculated

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

# Generate idempotent replay SQL to reprocess quarantined records
dlt-doctor replay --quarantine-table main.silver.orders_quarantine --target-table main.silver.orders --keys order_id --valid-filter "amount > 0" --erasure-table main.silver.orders_tombstones

# 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

# Inspect live Delta Lake table telemetry in Databricks (compaction debt, commit velocity, vacuum safety)
dlt-doctor inspect-table main.gold.orders --warehouse-id <warehouse_id>

4. Interactive Databricks Notebook Cell Magic

Audit PySpark code interactively inside Databricks or Jupyter notebook cells before cluster execution:

# Load the extension in notebook setup cell
%load_ext dlt_doctor

%%dlt_doctor
# Audits cell Catalyst AST and flags anti-patterns before cluster execution
from pyspark.sql import functions as F
from pyspark.sql.functions import udf

@udf
def parse_status(x):
    return x.lower() if x else "unknown"

df_orders = (
    spark.readStream.table("bronze_orders")
    .withColumn("c1", F.col("id"))
    .withColumn("c2", F.col("ts"))
    .withColumn("c3", F.col("amt"))
    .withColumn("c4", F.col("code"))
    .withColumn("status_clean", parse_status(F.col("status")))
)

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("""
from pyspark.sql import functions as F
from pyspark.sql.functions import udf

@udf
def clean_it(x):
    return x.strip() if x else None

df_clean = (
    df.withColumn("c1", F.col("a"))
      .withColumn("c2", F.col("b"))
      .withColumn("c3", F.col("c"))
      .withColumn("c4", F.col("d"))
      .withColumn("clean_val", clean_it(F.col("raw")))
      .repartition(4)
)
""")

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

# 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:

from dlt_doctor import ExpectationDoctor, ExpectationRule, CdcDoctor

# 1. Single-Pass NULL-Safe Quarantine (Valid -> Clean, 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)

# Synthesizes complete single-pass pipeline code with failed rules tracking
pipeline_code = exp_doctor.generate_pipeline_code(plan, source_table="bronze_orders", flavor="dp")
print(pipeline_code)


# 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,
    source_table="main.silver.orders_recalculated",
    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 diagnostic engines and integration components across Python 3.10, 3.11, and 3.12:

# Fast unit test suite (zero external cluster dependencies)
pytest -m "not integration"
# 197 passed in 0.52s

# Integration test suite (executed on local PySpark + Delta Lake)
REQUIRE_SPARK=1 pytest -m integration -rs -v
# 12 passed (executed Delta MERGE, SCD1/SCD2 CDC microbatches, and expectation pipelines)

All source code is audited with ruff check src tests, type-checked with mypy src, and verified against security vulnerabilities with bandit -r src.

For environment setup (including Windows JAVA_HOME and Hadoop binaries) and contributor guidelines, see CONTRIBUTING.md.

Limitations & Operational Boundaries

• Static AST Analysis: Static analysis inspects source AST without cluster execution. Dynamic constructs (e.g. eval(), runtime-generated expectation dicts, or dynamic SQL queries) cannot be resolved statically (emits DD003 INFO).
• SQL Pipeline Auto-Heal: dlt-doctor heal synthesizes Python AST definitions. SQL pipelines using ON VIOLATION DROP ROW must be manually migrated to dual-table quarantine structures.
• Derived Non-Deterministic Rules: Expectation conditions referencing runtime boundary calls (such as current_date()) emit a warning (UserWarning) because records evaluated near midnight boundaries may evaluate differently upon replay.
• Configuration: Audits and CI gates can be customized using pyproject.toml or dlt-doctor.toml. For complete options, see the Configuration Guide and Limitations Guide.

Disclaimers

dlt-doctor is an independent open-source project and is not affiliated with, sponsored by, or endorsed by Databricks, Inc. Apache Spark, Delta Lake, and Databricks are trademarks of their respective owners. Use of dlt-doctor does not guarantee regulatory compliance (such as GDPR, CCPA, or HIPAA); data engineering teams remain responsible for their own retention, masking, and governance policies.

License

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