Airflow dbt Integration (GoCardless)

Warnings

• gotcha This package relies on wrapping the dbt CLI. This means the dbt executable must be installed and available on the Airflow worker's PATH or explicitly set via the `dbt_bin` argument. This can be a common point of failure, especially in managed Airflow environments like AWS MWAA or GCP Cloud Composer, where managing CLI tools requires specific configurations.
  Fix: Ensure `dbt-core` and its relevant adapter are installed in your Airflow environment and that the `dbt` command is accessible from the Airflow worker processes. For managed services, this often means including it in `requirements.txt` and potentially configuring environment variables or plugins.
• gotcha The operators may not expose the full range of arguments available in the dbt CLI commands. For example, `DbtRunOperator` might not have an explicit `fail_fast` attribute, limiting granular control over dbt execution parameters directly from Airflow.
  Fix: Review the operator's source code for supported arguments. If a required argument is missing, you might need to use Airflow's `BashOperator` to execute dbt commands directly, or consider alternatives like `airflow-dbt-python` for more comprehensive dbt integration without CLI dependency.
• gotcha This package does not offer direct access to dbt artifacts (e.g., `manifest.json`, `run_results.json`) generated during execution. This limitation prevents more advanced use cases such as dynamic DAG generation based on dbt's lineage or pushing artifacts to Airflow XComs for downstream processing.
  Fix: For workflows requiring dbt artifact access, consider custom solutions using the `BashOperator` to read files from the dbt project's `target/` directory, or explore alternative integration packages that provide artifact parsing and XCom capabilities.
• deprecated The package has not received updates since September 2021 (v0.4.0), making it potentially incompatible with newer versions of Apache Airflow (e.g., Airflow 2.10+) or dbt-core (e.g., dbt-core 1.8+), which may introduce breaking changes or new features not supported by this older integration. For instance, `dbt-common`'s `isodate` constraint can conflict with Airflow 2.10.3+.
  Fix: Thoroughly test compatibility with your specific Airflow and dbt-core versions before deployment. Consider migrating to more actively maintained alternatives like `airflow-dbt-python` (github.com/tomasfarias/airflow-dbt-python) or `dbt-airflow` (github.com/gmyrianthous/dbt-airflow) which offer more recent updates and potentially address these compatibility concerns.

Install

• pip install airflow-dbt Install from PyPI

Imports

• DbtSeedOperator

  from airflow_dbt.operators.dbt_operator import DbtSeedOperator

• DbtSnapshotOperator

  from airflow_dbt.operators.dbt_operator import DbtSnapshotOperator

• DbtRunOperator

  from airflow_dbt.operators.dbt_operator import DbtRunOperator

• DbtTestOperator

  from airflow_dbt.operators.dbt_operator import DbtTestOperator

• DbtDocsGenerateOperator

  from airflow_dbt.operators.dbt_operator import DbtDocsGenerateOperator

  All operators are within the dbt_operator module.

Quickstart

This quickstart demonstrates how to define a basic Airflow DAG using `airflow-dbt` operators to run a sequence of dbt commands: `seed`, `snapshot`, `run`, and `test`. Ensure your dbt project directory and profiles directory (if not default) are correctly specified, typically via environment variables or directly in the `dir` and `profiles_dir` arguments. The dbt CLI must be installed and accessible on the Airflow worker's PATH.

from airflow import DAG
from airflow_dbt.operators.dbt_operator import (
    DbtSeedOperator,
    DbtSnapshotOperator,
    DbtRunOperator,
    DbtTestOperator
)
from airflow.utils.dates import days_ago
import os

default_args = {
    'dir': os.environ.get('DBT_PROJECT_DIR', '/path/to/your/dbt/project'),
    'start_date': days_ago(0)
}

with DAG(
    dag_id='dbt_example_dag',
    default_args=default_args,
    schedule_interval='@daily',
    tags=['dbt', 'example']
) as dag:
    dbt_seed = DbtSeedOperator(
        task_id='dbt_seed',
        profiles_dir=os.environ.get('DBT_PROFILES_DIR', '/path/to/your/.dbt') # Optional
    )

    dbt_snapshot = DbtSnapshotOperator(
        task_id='dbt_snapshot'
    )

    dbt_run = DbtRunOperator(
        task_id='dbt_run'
    )

    dbt_test = DbtTestOperator(
        task_id='dbt_test',
        retries=0 # Failing tests should fail the task, not retry
    )

    dbt_seed >> dbt_snapshot >> dbt_run >> dbt_test