Como testar DAGs do Airflow

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Antes de implantar os DAGs na produção, é possível executar subcomandos da CLI do Airflow para analisar o código DAG no mesmo contexto em que o DAG é executado.

Teste durante a criação do DAG

É possível executar uma única instância de tarefa localmente e ver a saída do registro. A visualização da saída permite verificar se há erros de sintaxe e de tarefa. O teste local não verifica as dependências nem informa o status para o banco de dados.

Recomendamos que você coloque os DAGs em uma pasta data/test no ambiente de teste.

Criar um diretório de teste

No bucket do ambiente, crie um diretório de teste e copie seus DAGs para ele.

gcloud storage cp BUCKET_NAME/dags \
  BUCKET_NAME/data/test --recursive

Substitua:

  • BUCKET_NAME: o nome do bucket associado ao seu ambiente do Cloud Composer.

Exemplo:

gcloud storage cp gs://us-central1-example-environment-a12bc345-bucket/dags \
  gs://us-central1-example-environment-a12bc345-bucket/data/test --recursive

Para mais informações sobre o upload de DAGs, consulte Adicionar e atualizar DAGs.

Verificar se há erros de sintaxe

Para verificar se há erros de sintaxe em DAGs enviados para a pasta /data/test, digite o seguinte comando gcloud:

Airflow 2 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  dags list -- --subdir /home/airflow/gcs/data/test

Airflow 1 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  list_dags -- -sd /home/airflow/gcs/data/test

Substitua:

  • ENVIRONMENT_NAME: o nome do ambiente.
  • ENVIRONMENT_LOCATION: a região em que o ambiente está localizado.

Verificar se há erros de tarefa

Para verificar se há erros específicos da tarefa em DAGs que você enviou para a pasta /data/test, execute o seguinte comando gcloud:

Airflow 2 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  tasks test -- --subdir /home/airflow/gcs/data/test \
  DAG_ID TASK_ID \
  DAG_EXECUTION_DATE

Airflow 1 

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  test -- -sd /home/airflow/gcs/data/test DAG_ID \
  TASK_ID DAG_EXECUTION_DATE

Substitua:

  • ENVIRONMENT_NAME: o nome do ambiente.
  • ENVIRONMENT_LOCATION: a região em que o ambiente está localizado.
  • DAG_ID: o ID do DAG.
  • TASK_ID: o ID da tarefa.
  • DAG_EXECUTION_DATE: a data de execução do DAG. Essa data é usada para modelos. Independentemente da data especificada aqui, o DAG é executado de imediato.

Exemplo:

Airflow 2 

gcloud composer environments run \
  example-environment \
  --location us-central1 \
  tasks test -- --subdir /home/airflow/gcs/data/test \
  hello_world print_date 2021-04-22

Airflow 1 

gcloud composer environments run example-environment \
  --location us-central1 \
  test -- -sd /home/airflow/gcs/data/test \
  hello_world print_date 2021-04-22

Como atualizar e testar um DAG implantado

Para testar atualizações dos DAGs no ambiente de teste:

  1. Copie o DAG implantado que você quer atualizar para data/test.
  2. Atualize o DAG.
  3. Teste o DAG.
    1. Verifique se há erros de sintaxe.
    2. Verifique se há erros específicos da tarefa.
  4. Verifique se o DAG é executado com êxito.
  5. Desative o DAG no ambiente de teste.
    1. Vá para a página IU do Airflow > DAGs.
    2. Se o DAG que você está modificando está sendo executado constantemente, desative o DAG.
    3. Para acelerar as tarefas pendentes, clique na tarefa e em Marcar com êxito.
  6. Implante o DAG no ambiente de produção.
    1. Desative o DAG no ambiente de produção.
    2. Faça o upload do DAG atualizado para a pasta dags/ no ambiente de produção.

Perguntas frequentes sobre testes de DAGs

Como faço para isolar as execuções de DAG nos ambientes de produção e teste?

Por exemplo, o Airflow tem um repositório global de código-fonte na pasta dags/ que é compartilhada por todos os DAGs. Você quer atualizar o código-fonte na produção ou no teste sem interferir nos DAGs em execução.

Os DAGs não são fortemente isolados pelo Airflow. Recomendamos que você tenha ambientes separados de teste e produção do Cloud Composer para impedir que os DAGs de teste interfiram nos de produção.

Como evitar a interferência do DAG ao executar testes de integração de diferentes ramificações do GitHub

Use nomes de tarefas exclusivos para evitar interferência. Por exemplo, inclua o nome da ramificação como prefixo nos IDs de tarefas.

Qual a prática recomendada para realizar testes de integração com o Airflow?

Recomendamos que você use um ambiente especial para realizar testes de integração com o Airflow. Uma maneira de sinalizar o sucesso da execução do DAG é gravar em um arquivo em uma pasta do Cloud Storage e verificar o conteúdo nos próprios casos de teste de integração.

Como posso contribuir de maneira eficiente com outros colaboradores do DAG?

Cada colaborador pode ter um subdiretório na pasta data/ para desenvolvimento.

Os DAGs adicionados à pasta data/ não são escolhidos automaticamente pelo programador do Airflow ou pelo servidor da Web.

Os colaboradores do DAG podem criar execuções manuais de DAG usando o comando gcloud composer environments run e o subcomando test com a sinalização --subdir para especificar o diretório de desenvolvimento do colaborador.

Exemplo:

Airflow 2 

gcloud composer environments run test-environment-name \
  tasks test -- dag-id task-id execution-date \
  --subdir /home/airflow/gcs/data/alice_dev

Airflow 1 

gcloud composer environments run test-environment-name \
  test -- dag-id task-id execution-date \
  --subdir /home/airflow/gcs/data/alice_dev

Como mantenho meus ambientes de implantação e produção sincronizados?

Para gerenciar o acesso, faça o seguinte:

Para executar implantações do desenvolvimento à produção, faça o seguinte:

  • Tenha uma configuração consistente, como variáveis de ambiente e pacotes PyPI.

  • Tenha argumentos consistentes de DAG. Para evitar a codificação, recomendamos o uso de macros e variáveis do Airflow.

    Exemplo:

    Airflow 2 

    gcloud composer environments run test-environment-name \
  variables set -- DATA_ENDPOINT_KEY DATA_ENDPOINT_VALUE

    Airflow 1 

    gcloud composer environments run test-environment-name \
  variables -- --set DATA_ENDPOINT_KEY DATA_ENDPOINT_VALUE

A seguir