Testar DAGs do Airflow

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

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

Testar durante a criação do DAG

Pode executar uma única instância de tarefa localmente e ver o resultado do registo. A visualização do resultado permite-lhe verificar a existência de erros de sintaxe e de tarefas. Os testes locais não verificam as dependências nem comunicam o estado à base de dados.

Recomendamos que coloque os DAGs numa pasta data/test no seu ambiente de teste.

Crie um diretório de teste

No contentor do seu ambiente, crie um diretório de teste e copie os seus DAGs para o mesmo.

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

Substitua o seguinte:

  • BUCKET_NAME: o nome do contentor 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 carregamento de DAGs, consulte o artigo Adicione e atualize DAGs.

Verifique se existem erros de sintaxe

Para verificar se existem erros de sintaxe nos DAGs que carregou para a pasta /data/test, introduza o seguinte comando gcloud:

Airflow 2 

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

Fluxo de ar 1 

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

Substitua o seguinte:

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

Verifique se existem erros de tarefas

Para verificar se existem erros específicos de tarefas em DAGs que carregou 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

Fluxo de ar 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 o seguinte:

  • ENVIRONMENT_NAME: o nome do ambiente.
  • ENVIRONMENT_LOCATION: a região onde 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. Esta data é usada para fins de criação de modelos. Independentemente da data que especificar aqui, o DAG é executado imediatamente.

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

Fluxo de ar 1 

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

Atualizar e testar um DAG implementado

Para testar atualizações às suas DAGs no ambiente de teste:

  1. Copie o DAG implementado que quer atualizar para data/test.
  2. Atualize o DAG.
  3. Teste o DAG.
    1. Verifique se existem erros de sintaxe.
    2. Verifique se existem erros específicos da tarefa.
  4. Certifique-se de que o DAG é executado com êxito.
  5. Desative o DAG no seu ambiente de teste.
    1. Aceda à IU do Airflow > página DAGs.
    2. Se o DAG que está a modificar for executado constantemente, desative-o.
    3. Para acelerar as tarefas pendentes, clique na tarefa e em Marcar como concluída.
  6. Implemente o DAG no seu ambiente de produção.
    1. Desative o DAG no seu ambiente de produção.
    2. Carregue o DAG atualizado para a pasta dags/ no seu ambiente de produção.

Perguntas frequentes sobre os testes de DAGs

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

Por exemplo, o Airflow tem um repositório global de código-fonte na pasta dags/ que todas as execuções de DAG partilham. Quiser atualizar o código-fonte em produção ou testar sem interferir com os DAGs em execução.

O Airflow não oferece um isolamento DAG forte. Recomendamos que mantenha ambientes de produção e de teste do Cloud Composer separados para evitar que os DAGs de teste interfiram com os DAGs de produção.

Como posso evitar a interferência de DAGs quando executo testes de integração a partir de diferentes ramificações do GitHub

Use nomes de tarefas únicos para evitar interferências. Por exemplo, pode prefixar os IDs das tarefas com o nome da ramificação.

Qual é uma prática recomendada para testes de integração com o Airflow?

Recomendamos que use um ambiente dedicado para testes de integração com o Airflow. Uma forma de sinalizar o êxito da execução do DAG é escrever num ficheiro numa pasta do Cloud Storage e, em seguida, verificar o conteúdo nos seus próprios casos de teste de integração.

Como posso colaborar de forma eficiente com outros colaboradores do DAG?

Cada colaborador pode ter uma subdiretoria na pasta data/ para desenvolvimento.

Os DAGs adicionados à pasta data/ não são selecionados automaticamente pelo agendador do Airflow nem pelo servidor Web

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

Por 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

Fluxo de ar 1 

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

Como posso manter os meus ambientes de implementação e produção sincronizados?

Para gerir o acesso:

Para implementar da programação para a produção:

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

  • Garantir argumentos DAG consistentes. Para evitar a programação direta, recomendamos que use macros e variáveis do Airflow.

    Por exemplo:

    Airflow 2 

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

    Fluxo de ar 1 

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

O que se segue?