Como testar DAGs (fluxos de trabalho)

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

Como testar 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 que você verifique se há erros de sintaxe e tarefa. O teste local não verifica as dependências nem se comunica com o status para o banco de dados.

Recomendamos colocar os DAGs em uma pasta data/test no ambiente de teste.

Como criar um diretório de teste

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

    gsutil

    gsutil cp -r BUCKET_NAME/dags BUCKET_NAME/data/test
    

    Em que:

    • BUCKET_NAME é o nome do bucket do Cloud Storage associado ao ambiente do Cloud Composer.

    Por exemplo:

    gsutil

    gsutil cp -r gs://us-central1-test-environment-a12bc345-bucket/dags gs://us-central1-test-environment-a12bc345-bucket/data/test
    

Como verificar erros de sintaxe

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

  2. Para verificar se há erros de sintaxe, insira o seguinte comando gcloud:

    Airflow 1

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

    Airflow 2

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

    Em que:

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

    Por exemplo:

    Airflow 1

    gcloud composer environments run \
    test-environment --location us-central1 \
    list_dags -- -sd /home/airflow/gcs/data/test
    

    Airflow 2

    gcloud composer environments run \
    test-environment --location us-central1 \
    dags list -- --subdir /home/airflow/gcs/data/test
    

Como verificar erros na tarefa

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

  2. Para verificar se há erros específicos da tarefa, digite o seguinte comando gcloud:

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

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

Em que:

  • 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 fins de modelos. Independentemente da data especificada aqui, o DAG é executado imediatamente.

Por exemplo:

Airflow 1

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

Airflow 2

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

Como atualizar e testar um DAG implantado

Para testar as 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 foi executado corretamente.
  5. Desative o DAG no ambiente de teste.
    1. Acesse a página da interface do usuário do Airflow &gt. DAGs.
    2. Se o DAG estiver sendo constantemente executado, desative-o.
    3. Para acelerar tarefas pendentes, clique na tarefa e em Marcar como bem-sucedida.
  6. Implante o DAG no ambiente de produção.
    1. Desative o DAG no ambiente de produção.
    2. Faça upload do DAG atualizado para a pasta dags/ no ambiente de produção.

Perguntas frequentes sobre testes de DAGs

Como 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 é compartilhado por todas as DAGs. Você quer atualizar o código-fonte na produção ou no teste sem interferir nos DAGs em execução.

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

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

Use nomes exclusivos para evitar interferência. Por exemplo, você pode prefixar os IDs das tarefas com o nome da ramificação.

Qual é a prática recomendada para testar a integração com o Airflow?

Recomendamos que você use um ambiente dedicado para 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 seus próprios casos de teste de integração.

Como posso colaborar de forma 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 selecionados automaticamente pelo programador do Airflow ou servidor da Web.

Os colaboradores do DAG podem criar execuções manuais do 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.

Por exemplo:

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

No Airflow 2, use o comando tasks test:

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

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

Para gerenciar o acesso, siga estas etapas:

Para implantar do desenvolvimento à produção:

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

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

    Por exemplo:

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

    Na CLI do Airflow 2.0, use o comando variables set:

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

Próximas etapas