How to use DBT with AWS Managed Airflow?

Question

hope you are doing well. I wanted to check if anyone has get up and running with dbt in aws mwaa airflow.

I have tried without success this one and this python packages but fails for some reason or another (can't find the dbt path, etc).

Did anyone has managed to use MWAA (Airflow 2) and DBT without having to build a docker image and placing it somewhere?

Thank you!

Answer 1

I've managed to solve this by doing the following steps:

1. Add dbt-core==0.19.1 to your requirements.txt
2. Add DBT cli executable into plugins.zip

#!/usr/bin/env python3
# EASY-INSTALL-ENTRY-SCRIPT: 'dbt-core==0.19.1','console_scripts','dbt'
__requires__ = 'dbt-core==0.19.1'
import re
import sys
from pkg_resources import load_entry_point

if __name__ == '__main__':
    sys.argv[0] = re.sub(r'(-script\.pyw?|\.exe)?$', '', sys.argv[0])
    sys.exit(
        load_entry_point('dbt-core==0.19.1', 'console_scripts', 'dbt')()
    )

And from here you have two options:

1. Setting dbt_bin operator argument to /usr/local/airflow/plugins/dbt
2. Add /usr/local/airflow/plugins/ to the $PATH by following the docs

Environment variable setter example:

from airflow.plugins_manager import AirflowPlugin
import os

os.environ["PATH"] = os.getenv(
    "PATH") + ":/usr/local/airflow/.local/lib/python3.7/site-packages:/usr/local/airflow/plugins/"


class EnvVarPlugin(AirflowPlugin):
    name = 'env_var_plugin'

The plugins zip content:

plugins.zip
├── dbt (DBT cli executable)
└── env_var_plugin.py (environment variable setter)

Answer 2

Using the pypi airflow-dbt-python package has simplified the setup of dbt_ to MWAA for us, as it avoids needing to amend PATH environment variables in the plugins file. However, I've yet to have a successful dbt_ run via either airflow-dbt or airflow-dbt-python packages, as MWAA worker seems to be a read only filesystem, so as soon as dbt_ tries to compile to the target directory, the following error occurs:

File "/usr/lib64/python3.7/os.py", line 223, in makedirs
    mkdir(name, mode)
OSError: [Errno 30] Read-only file system: '/usr/local/airflow/dags/dbt/target'

Answer 3

This is how I managed to do it:

    @dag(**default_args)
    def dbt_dag():
        @task()
        def run_dbt():
            from dbt.main import handle_and_check
            
            os.environ["DBT_TARGET_DIR"] = "/usr/local/airflow/tmp/target"
            os.environ["DBT_LOG_DIR"] = "/usr/local/airflow/tmp/logs"
            os.environ["DBT_PACKAGE_DIR"] = "/usr/local/airflow/tmp/packages"
            succeeded = True
            try:
                args = ['run', '--whatever', 'bla']
                results, succeeded = handle_and_check(args)
                print(results, succeeded)
            except SystemExit as e:
                if e.code != 0:
                raise e   
            if not succeeded:
                raise Exception("DBT failed")         

note that my dbt_project.yml has the following paths, this is to avoid os exception when trying to write to read only paths:

    target-path: "{{ env_var('DBT_TARGET_DIR', 'target') }}"  # directory which will store compiled SQL files
    log-path: "{{ env_var('DBT_LOG_DIR', 'logs') }}"  # directory which will store dbt logs
    packages-install-path: "{{ env_var('DBT_PACKAGE_DIR', 'packages') }}"  # directory which will store dbt packages

Answer 4

Combining the answer from @Yonatan Kiron & @Ofer Helman works for me. I just need to fix these 3 files:

• requiremnt.txt
• plugins.zip
• dbt_project.yml

My requirements.txt I specify the version I want, and looks like this:

airflow-dbt==0.4.0
dbt-core==1.0.1
dbt-redshift==1.0.0

Note that, as of v1.0.0, pip install dbt is no longer supported and will raise an explicit error. Since v0.13, the PyPi package named dbt was a simple "pass-through" of dbt-core. (refer https://docs.getdbt.com/dbt-cli/install/pip#install-dbt-core-only)

For my plugins.zip I add a file env_var_plugin.py that looks like this

from airflow.plugins_manager import AirflowPlugin
import os

os.environ["DBT_LOG_DIR"] = "/usr/local/airflow/tmp/logs"
os.environ["DBT_PACKAGE_DIR"] = "/usr/local/airflow/tmp/dbt_packages"
os.environ["DBT_TARGET_DIR"] = "/usr/local/airflow/tmp/target"

class EnvVarPlugin(AirflowPlugin):                
     name = 'env_var_plugin'

And finally I add this in my dbt_project.yml

log-path: "{{ env_var('DBT_LOG_DIR', 'logs') }}"  # directory which will store dbt logs
packages-install-path: "{{ env_var('DBT_PACKAGE_DIR', 'dbt_packages') }}"  # directory which will store dbt packages
target-path: "{{ env_var('DBT_TARGET_DIR', 'target') }}"  # directory which will store compiled SQL files

And as stated in the airflow-dbt github, (https://github.com/gocardless/airflow-dbt#amazon-managed-workflows-for-apache-airflow-mwaa) configure the dbt task like below:

  dbt_bin='/usr/local/airflow/.local/bin/dbt',
  profiles_dir='/usr/local/airflow/dags/{DBT_FOLDER}/',
  dir='/usr/local/airflow/dags/{DBT_FOLDER}/'