DOCSEARCHbyAlgolia
Submit Docs

dbt-af

repository·main·Indexed 41 days ago

https://github.com/toloka/dbt-af

A tool for running dbt models in a distributed manner using Apache Airflow. It wraps Airflow DAGs to allow models to run independently while preserving dependencies, supporting large, domain-driven data meshes. Features include automated DAG generation from dbt manifests, custom scheduling tags, dependency wait policies, and integrations for Tableau refreshes and Kubernetes tasks.

Tokens
11K
Snippets
36
Records
44
Agent score
51%

What's inside dbt-af

  1. Understand dbt-af DAG generation patterns

    main

    When using dbt-af, DAGs are generated based on domain-oriented structures and scheduling annotations.

    Key patterns include:

    • Scheduling-based DAGs: Models with specific scheduling annotations (e.g., @daily, @hourly) are grouped into unified DAGs named after their domain and schedule (e.g., svc_jaffle_shop__daily or svc_jaffle_shop__hourly).
    • Cross-scheduling dependencies: If a model in an hourly DAG depends on a model in a daily DAG, dbt-af resolves this using Airflow waits (sensors).
    • Backfill DAGs: For every domain, dbt-af generates a backfill DAG following the pattern <domain_name>__backfill. These DAGs have @daily scheduling but are intended for manual triggering to recover data in case of issues with regular DAGs.

  2. Understand how dbt-af determines the execution target

    main

    dbt-af follows a specific hierarchy to decide which target to use for a task:

    1. Tests: Uses config.dbt_default_targets.default_for_tests_target.
    2. Models:
      • If the model has no pre-hooks AND its scheduling is @daily or @weekly, it uses the daily_sql_cluster target.
      • Otherwise, it uses the sql_cluster target.
    3. Other nodes: Uses the py_cluster target.

    Note: You can override this behavior for a specific model by setting the dbt_target parameter in the model's dbt configuration.

    # Explicitly setting a target in a dbt model config
```yaml
config:
  dbt_target: large_dev

  3. Configure small, medium, and large tests

    main

    dbt-af categorizes dbt tests into three types based on their impact on the DAG and resource consumption:

    1. Small Tests: These are all dbt tests that do not have a specific tag. They are collected into a single task group for a model and run immediately after the model finishes. If a small test fails, downstream models are blocked. Each test runs in a separate operator to ensure atomicity.

    2. Medium Tests: These are non-blocking tests designed for more resource-intensive checks. They run after all models in the DAG have finished. To configure a medium test, add the @medium tag to the test in your dbt YAML configuration. Note that medium tests can only be configured for models that have their own .yml file.

    3. Large Tests: These are the heaviest tests. They are placed in a dedicated DAG for each domain (named <domain_name>__large_tests_) and are scheduled with @daily frequency.

    # Example of a medium test configuration in dbt YAML
```yaml
tests:
  - unique: # or any other test
      tags: [ "@medium" ]

  4. Manage cross-domain dependencies in dbt-af

    main
    In a multi-domain project, you can have models in one domain (e.g., dmn_jaffle_analytics) that depend on models from another domain (e.g., dmn_jaffle_shop). dbt-af handles this by creating task groups with sensors. The dependent group will only start execution once the upstream models in the required domain have finished successfully.

  5. Configure Databricks maintenance tasks

    main

    To use maintenance tasks in your project, you must first copy the files from the dags/macros directory of the dbt-af repository into your dbt project.

    Maintenance tasks are configured within the config block of your dbt models using the maintenance key. Currently, these tasks are specifically designed for Databricks.

  6. Integrate dbt-af with Tableau

    main

    You can trigger Tableau extract refreshes automatically after dbt runs.

    Installation

    Install dbt-af with the tableau extra:

    pip install dbt-af[tableau]

    Configuration

    Use TableauIntegrationConfig to define your Tableau server credentials.

    Authentication Modes:

    1. Basic Authentication: Provide username and password.
    2. Personal Access Token (PAT): Provide token_name and pat instead of username and password.
    from dbt_af.conf import Config, TableauIntegrationConfig

# Using Basic Authentication
config = Config(
    # ...
    tableau=TableauIntegrationConfig(
        server_address='https://tableau.server.com',
        username='admin',
        password='admin',
        site_id='my_site',
    ),
)

# Using Personal Access Token (PAT)
# config = Config(
#     # ...
#     tableau=TableauIntegrationConfig(
#         server_address='https://tableau.server.com',
#         token_name='YOUR_TOKEN_NAME',
#         pat='YOUR_PAT_VALUE',
#         site_id='my_site',
#     ),
# )

  7. Backfill data using Backfill DAGs

    main

    To backfill data for a specific interval, use the generated backfill DAG for the relevant domain (named <domain_name>__backfill).

    1. Identify the domain-specific backfill DAG.
    2. Trigger the DAG manually in Airflow.
    3. Provide the specific date interval during the trigger.

    This will run all models and tests within that specific domain for the provided interval.

  8. Use the dbt_run_model DAG for manual execution

    main

    If include_single_model_manual_dag is enabled in your Config, a DAG named <dbt_project_name>_dbt_run_model is generated. This DAG allows you to manually execute dbt run commands for backfilling, running new models, or re-running models after code changes.

    When triggering the DAG in Airflow, you can provide the following parameters:

    1. Model Selector: The model name (e.g., jaffle_shop.ods.orders) or dbt selection syntax (e.g., jaffle_shop.ods.orders+).
    2. Interval Start Datetime: Passed to the model as {{ var("start_dttm") }}.
    3. Interval End Datetime: Passed to the model as {{ var("end_dttm") }}.
    4. Target (optional): Override the default dbt target.
    5. Full Refresh (optional): If True, runs with the --full-refresh flag.
    6. Extra Arguments (optional): Additional dbt flags/options provided in JSON format.
    {
  "--option1": "value1",
  "--option2": "value2",
  "--flag": ""
}

  9. List of dbt-af Usage Examples

    main

    The dbt-af repository provides several example configurations covering different orchestration scenarios:

    • Basic Project: Single domain, small tests, and a single target.
    • Advanced Project: Multiple domains, medium/large tests, and different targets.
    • Dependencies Management: Managing dependencies between models across different domains.
    • Manual Scheduling: Configuring domains that require manual scheduling.
    • Maintenance and Source Freshness: Managing maintenance tasks and dbt source freshness.
    • Python Venv Tasks: Running custom dbt models within Python Virtual Environments.
    • Kubernetes Tasks: Running dbt models in Kubernetes.
    • Integration with Other Tools: Integrating dbt-af with external toolsets.
    • Extras and Scripts: Available extras and utility scripts.

  10. Configure dbt targets for models and seeds

    main

    To use different compute resources (targets) for different types of models and tests, you must define four specific target keys in your dbt_project.yml for both models and seeds.

    Required keys in dbt_project.yml:

    • sql_cluster
    • daily_sql_cluster
    • py_cluster
    • bf_cluster

    Then, configure the default targets in your dbt-af Python configuration using DbtDefaultTargetsConfig.

    # dbt_project.yml
models:
  sql_cluster: "dev"
  daily_sql_cluster: "dev"
  py_cluster: "dev"
  bf_cluster: "dev"

seeds:
  sql_cluster: "dev"
  daily_sql_cluster: "dev"
  py_cluster: "dev"
  bf_cluster: "dev"
    # dbt-af Config
from dbt_af.conf import Config, DbtDefaultTargetsConfig

config = Config(
    # ...
    dbt_default_targets=DbtDefaultTargetsConfig(
        default_target='dev',
        default_for_tests_target='dev',  # if undefined, uses default_target
    ),
    # ...
)

  11. Configure wait policy for upstream dependencies

    main

    By default, dbt-af waits only for the last run of an upstream model. If a model with infrequent scheduling (e.g., @daily) depends on a model with frequent scheduling (e.g., @hourly), you may want to ensure all runs of the upstream model that occurred between the last model run and the current one are completed.

    To wait for all intermediate runs, set wait_policy: all in the dependencies section of the model's config.

    Warning: Using wait_policy: all can significantly increase the number of tasks generated in your Airflow DAG. If you use this setting, consider increasing the number of slots in the dbt_sensor_pool pool to handle the increased task load.

    config:
  dependencies:
    upstream.model.name:
      wait_policy: all

  12. Enable source freshness checks

    main

    To ensure data freshness before executing models, add a freshness section to your source's YAML configuration. When this section is present, dbt-af will automatically create sensors for all models depending on that source. If the source is not fresh, the dependent models will not be executed.

    Required configuration keys:

    • freshness: Contains the error threshold.
      • error_after.count: A positive integer representing the threshold.
      • error_after.period: The time unit (minute, hour, or day).
    • loaded_at_field: The column or expression used to check the freshness timestamp.
    sources:
  - name: my_source
    freshness:
      error_after:
        count: 15  # positive integer
        period: hour  # minute | hour | day
    loaded_at_field: my_source_loaded_at  # column or expression on which to check the freshness