Como usar DBT com o Amazon MWAA

Este tópico demonstra como é possível usar o DBT e o Postgres com o Amazon MWAA. Nas etapas a seguir, você adicionará as dependências necessárias ao seu requirements.txt e fará o upload de um exemplo de projeto de DBT para o bucket Amazon S3 do seu ambiente. Em seguida, você usará um exemplo de DAG para verificar se o Amazon MWAA instalou as dependências e, por fim, usará BashOperator para executar o projeto de DBT.

Versão

• É possível usar o exemplo de código nesta página com o Apache Airflow v2 no Python 3.10.

Pré-requisitos

Antes de concluir as etapas a seguir, você precisará do seguinte:

• Um ambiente do Amazon MWAA usando o Apache Airflow v2.2.2. Esse exemplo foi gravado e testado com a versão 2.2.2. Talvez seja necessário modificar o exemplo para usá-lo com outras versões do Apache Airflow.

• Um exemplo de projeto de DBT. Para começar a usar o dbt com o Amazon MWAA, você pode criar uma bifurcação e clonar o projeto dbt starter a partir do repositório dbt-labs. GitHub

Dependências

Para usar o Amazon MWAA com DBT, adicione o script de inicialização a seguir ao ambiente. Para saber mais, consulte Using a startup script with Amazon MWAA.

#!/bin/bash

if [[ "${MWAA_AIRFLOW_COMPONENT}" != "worker" ]]
then
    exit 0
fi

echo "------------------------------"
echo "Installing virtual Python env"
echo "------------------------------"

pip3 install --upgrade pip

echo "Current Python version:"
python3 --version 
echo "..."

sudo pip3 install --user virtualenv
sudo mkdir python3-virtualenv
cd python3-virtualenv
sudo python3 -m venv dbt-env
sudo chmod -R 777 *

echo "------------------------------"
echo "Activating venv in"
$DBT_ENV_PATH
echo "------------------------------"

source dbt-env/bin/activate
pip3 list

echo "------------------------------"
echo "Installing libraries..."
echo "------------------------------"

# do not use sudo, as it will install outside the venv
pip3 install dbt-redshift==1.6.1 dbt-postgres==1.6.1

echo "------------------------------"
echo "Venv libraries..."
echo "------------------------------"

pip3 list
dbt --version

echo "------------------------------"
echo "Deactivating venv..."
echo "------------------------------"

deactivate

Nas seções a seguir, você fará o upload do diretório do seu projeto de DBT para o Amazon S3 e executará um DAG que valida se o Amazon MWAA instalou com sucesso as dependências DBT necessárias.

Faça o upload de um projeto de DBT para o Amazon S3

Para poder usar um projeto de DBT com seu ambiente Amazon MWAA, é possível fazer o upload todo o diretório do projeto na pasta dags do seu ambiente. Quando o ambiente é atualizado, o Amazon MWAA baixa o diretório DBT para a pasta local usr/local/airflow/dags/.

Para fazer upload de um projeto de DBT no Amazon S3

1. Navegue até o diretório em que você clonou o projeto inicial de DBT.

2. Execute o seguinte AWS CLI comando do Amazon S3 para copiar recursivamente o conteúdo do projeto para a dags pasta do seu ambiente usando o parâmetro. --recursive O comando cria um subdiretório chamado dbt que é possível usar para todos os seus projetos de DBT. Se o subdiretório já existir, os arquivos do projeto serão copiados para o diretório existente e um novo diretório não será criado. O comando também cria um subdiretório dentro do diretório dbt para esse projeto inicial específico.

   $ aws s3 cp dbt-starter-project s3://mwaa-bucket/dags/dbt/dbt-starter-project --recursive

   É possível usar nomes diferentes para os subdiretórios do projeto para organizar vários projetos de DBT dentro do diretório principal dbt.

Use um DAG para verificar a instalação da dependência de DBT

O DAG a seguir usa um código BashOperator e um comando bash para verificar se o Amazon MWAA instalou com sucesso as dependências DBT especificadas em requirements.txt.

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago

with DAG(dag_id="dbt-installation-test", schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
    cli_command = BashOperator(
        task_id="bash_command",
        bash_command=""/usr/local/airflow/python3-virtualenv/dbt-env/bin/dbt --version""
    )

Faça o seguinte para visualizar os logs de tarefas e verificar se o DBT e as dependências dele foram instalados.

1. Navegue até o console do Amazon MWAA e escolha Abrir IU do Airflow na lista de ambientes disponíveis.

2. Na IU do Apache Airflow, encontre o DAG dbt-installation-test na lista e escolha a data na coluna Last Run para abrir a última tarefa bem-sucedida.

3. Usando a Exibição de gráfico, escolha a bash_command tarefa para abrir os detalhes da instância da tarefa.

4. Escolha Log para abrir os logs de tarefas e, em seguida, verifique se os logs listam com êxito a versão do DBT que especificamos em requirements.txt.

Use um DAG para executar um projeto de DBT

O DAG a seguir usa um código BashOperator para copiar os projetos de DBT que você fez o upload para o Amazon S3 do diretório local usr/local/airflow/dags/ para o diretório acessível para gravação /tmp e, em seguida, executa o projeto de DBT. Os comandos bash pressupõem um projeto inicial de DBT intitulado dbt-starter-project. Modifique o nome do diretório de acordo com o nome do diretório do seu projeto.

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago

import os

DAG_ID = os.path.basename(__file__).replace(".py", "")

# assumes all files are in a subfolder of DAGs called dbt

with DAG(dag_id=DAG_ID, schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
    cli_command = BashOperator(
        task_id="bash_command",
        bash_command="source /usr/local/airflow/python3-virtualenv/dbt-env/bin/activate;\
cp -R /usr/local/airflow/dags/dbt /tmp;\
echo 'listing project files:';\
ls -R /tmp;\
cd /tmp/dbt/mwaa_dbt_test_project;\
/usr/local/airflow/python3-virtualenv/dbt-env/bin/dbt run --project-dir /tmp/dbt/mwaa_dbt_test_project --profiles-dir ..;\
cat /tmp/dbt_logs/dbt.log;\
rm -rf /tmp/dbt/mwaa_dbt_test_project"
    )