Fast.BI DBT Runner

A comprehensive Python library for managing DBT (Data Build Tool) DAGs within the Fast.BI data development platform. This package provides multiple execution operators optimized for different cost-performance trade-offs, from low-cost slow execution to high-cost fast execution.

Overview

Fast.BI DBT Runner is part of the Fast.BI Data Development Platform, designed to provide flexible and scalable DBT workload execution across various infrastructure options. The package offers four distinct operator types, each optimized for specific use cases and requirements.

Key Features

• Multiple Execution Operators: Choose from K8S, Bash, API, or GKE operators
• Cost-Performance Optimization: Scale from low-cost to high-performance execution
• Airflow Integration: Seamless integration with Apache Airflow workflows
• Manifest Parsing: Intelligent DBT manifest parsing for dynamic DAG generation
• Tag-based Filtering: Filter which nodes run using DBT_TAGS
• Sharding Control: Run models/seeds/snapshots/sources as individual tasks (lineage) or as a single batch task (--select)
• Manifest Caching: Hash-based caching reduces DAG parse time by 99%+ for unchanged manifests
• Airbyte Integration: Built-in support for Airbyte task group building

Installation

pip install fast-bi-dbt-runner

# With Airflow
pip install fast-bi-dbt-runner[airflow]

# With development tools
pip install fast-bi-dbt-runner[dev]

Operator Types

Airflow Variable Reference

All variables are read from Airflow Variables at DAG load time. Defaults shown in parentheses.

Infrastructure & Identity

Model Execution Control

These three flags follow the same pattern: is_in_manifest → DBT_X → DBT_X_SHARDING.

Seed Execution Control

Snapshot Execution Control

Source Freshness Control

Pipeline Steps

Filtering & Selection

Full Refresh

E2E / Testing Modes

Monitoring & Logging

Manifest Cache (Environment Variables)

These are set as environment variables on the Airflow workers, not as Airflow Variables:

Sharding Explained

Sharding = True (default) — Full Lineage

Each model/seed/snapshot/source becomes its own Airflow task. Airflow builds the full dependency graph from the dbt manifest, so tasks execute in dependency order and you can retry individual failed nodes.

seed_customers → model_stg_customers → model_orders → model_revenue
                                     ↗
                model_stg_orders ───

Use when: You need visibility into individual model failures, want to retry a single model, or have long-running models that benefit from parallelism.

Sharding = False — Batch Mode

All tag-filtered models are collected from the manifest and passed in a single dbt run --select "model1 model2 ..." command, running as one Airflow task. Tag filtering (DBT_TAGS) is fully respected — only manifested, tag-matching nodes are included.

single_task: dbt run --select "stg_customers stg_orders dim_revenue"

Use when: You have many small models, want simpler DAGs with fewer tasks, or Airflow overhead per-task is significant.

Configuration Examples

Standard Daily Pipeline (K8S)

# Airflow Variables
{
    "PROJECT_ID": "my-gcp-project",
    "DBT_PROJECT_NAME": "analytics",
    "DAG_SCHEDULE_INTERVAL": "@daily",
    "DBT_DEPS": "True",
    "DBT_SOURCE": "True",
    "DBT_SOURCE_SHARDING": "True",
    "DBT_SEED": "False",
    "DBT_MODEL": "True",
    "DBT_MODEL_SHARDING": "True",
    "DBT_SNAPSHOT": "False",
    "DATA_QUALITY": "True"
}

Batch Mode (Many Small Models, Low Overhead)

# All resource types run as single batch tasks — fewer Airflow tasks, simpler DAG
{
    "DBT_MODEL": "True",
    "DBT_MODEL_SHARDING": "False",   # dbt run --select "model1 model2 ..."
    "DBT_SEED": "True",
    "DBT_SEED_SHARDING": "False",    # dbt seed --select "seed1 seed2 ..."
    "DBT_SOURCE": "True",
    "DBT_SOURCE_SHARDING": "False",  # dbt source freshness --select "src1 src2 ..."
    "DBT_SNAPSHOT": "True",
    "DBT_SNAPSHOT_SHARDING": "False" # dbt snapshot --select "snap1 snap2 ..."
}

Tag-Filtered Pipeline

# Only run nodes tagged with both "marketing" and "daily"
{
    "DBT_TAGS": "marketing,daily",
    "DBT_MODEL": "True",
    "DBT_MODEL_SHARDING": "True"
}

Full Refresh Specific Models

# Full refresh only two models; others run normally
{
    "FULL_REFRESH": "False",
    "FULL_REFRESH_MODEL_NAME": "dim_customers,fct_orders"
}

Full Refresh All Models

{
    "FULL_REFRESH": "True"
}

E2E Schema Validation

# Creates empty tables (no data) to validate schema changes end-to-end
{
    "E2E_MODE_EMPTY": "True",
    "DBT_MODEL": "True",
    "DBT_MODEL_SHARDING": "False"
}

Skip Package Installation (Vendored Packages)

# Packages are committed to the repo — skip dbt deps for faster, more reliable runs
{
    "DBT_DEPS": "False"
}

High-Performance Real-Time Pipeline (API Operator)

{
    "PROJECT_ID": "my-gcp-project",
    "DBT_PROJECT_NAME": "realtime_analytics",
    "DAG_SCHEDULE_INTERVAL": "*/15 * * * *",
    "DBT_DEPS": "False",
    "DBT_MODEL": "True",
    "DBT_MODEL_SHARDING": "False",  # batch mode for speed
    "DBT_TAGS": "realtime",
    "MODEL_DEBUG_LOG": "True"
}

Architecture

Execution Flow per DAG

[Airbyte sync] (optional)
     ↓
[dbt deps]     (if DBT_DEPS=True)
     ↓
[dbt debug]    (if DEBUG=True)
     ↓
[show_input_data]
     ↓
[dbt source freshness]  (if DBT_SOURCE=True)
     ↓
[dbt seed]              (if DBT_SEED=True)
     ↓
[dbt run]               (if DBT_MODEL=True)
     ↓
[dbt snapshot]          (if DBT_SNAPSHOT=True)
     ↓
[re_data / quality]     (if DATA_QUALITY=True)

Manifest Caching

The manifest caching system reduces DAG import time by 99%+ for unchanged manifests:

• Before caching: ~2–4 seconds per manifest parse, ~480 parses/hour with 2 schedulers
• After caching: <10ms for cache hits, only 5–10 cache misses/hour (on actual manifest changes)
• Cache keys include: file MD5 hash + DBT_TAGS + ancestor/descendant flags
• Cache is process-local (not shared across pod restarts); first parse after restart is always a cache miss

CI/CD

Tests run automatically on every push:

# Run tests locally
pytest tests/

# With coverage
pytest tests/ --cov=fast_bi_dbt_runner --cov-report=term-missing

Release Process

1. Bump version in pyproject.toml
2. Add entry to CHANGELOG.md
3. Create and push a version tag: git tag 2026.1.0.6 && git push origin 2026.1.0.6
4. GitHub Actions tests, builds, and publishes to PyPI automatically

Support

• Documentation: Fast.BI Platform Wiki
• Email: support@fast.bi
• Issues: GitHub Issues
• Changelog: CHANGELOG.md