Programar e acionar DAGs do Airflow

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Nesta página, explicamos como o agendamento e o acionamento de DAGs funcionam no Airflow, como definir um cronograma para um DAG e como acionar ou pausar um DAG manualmente.

Sobre os DAGs do Airflow no Cloud Composer

Os DAGs do Airflow no Cloud Composer são executados em um ou mais ambientes do Cloud Composer no seu projeto. Você envia os arquivos de origem dos seus DAGs do Airflow para um bucket do Cloud Storage associado a um ambiente. A instância do Airflow do ambiente analisa esses arquivos e programa execuções de DAG, conforme definido pela programação de cada DAG. Durante uma execução de DAG, o Airflow programa e executa tarefas individuais que compõem um DAG em uma sequência definida por ele.

Para saber mais sobre os conceitos básicos do Airflow, como DAGs, execuções de DAGs, tarefas ou operadores, consulte a página Conceitos básicos na documentação do Airflow.

Sobre a programação de DAGs no Airflow

O Airflow oferece os seguintes conceitos para o mecanismo de programação:

Data lógica

Representa uma data em que uma determinada execução de DAG é realizada.

Essa não é a data real em que o Airflow executa um DAG, mas um período que uma execução específica do DAG precisa processar. Por exemplo, para um DAG programado para ser executado todos os dias às 12h, a data lógica também seria 12h em um dia específico. Como ele é executado duas vezes por dia, o período que precisa ser processado é das últimas 12 horas. Ao mesmo tempo, a lógica definida no DAG pode não usar a data lógica ou o intervalo de tempo. Por exemplo, um DAG pode executar o mesmo script uma vez por dia sem usar o valor da data lógica.

Em versões do Airflow anteriores à 2.2, essa data é chamada de data de execução.

Data de execução

Representa uma data em que uma execução específica do DAG é realizada.

Por exemplo, para um DAG programado para ser executado todos os dias às 12h, a execução real pode acontecer às 12h05, algum tempo depois da data lógica.

Intervalo da programação

Representa quando e com que frequência um DAG precisa ser executado, em termos de datas lógicas.

Por exemplo, uma programação diária significa que um DAG é executado uma vez por dia, e as datas lógicas das execuções do DAG têm intervalos de 24 horas.

Data de início

Especifica quando você quer que o Airflow comece a programar o DAG.

As tarefas no DAG podem ter datas de início individuais ou você pode especificar uma única data de início para todas as tarefas. Com base na data de início mínima das tarefas no DAG e no intervalo da programação, o Airflow programa as execuções de DAGs.

Recuperação, preenchimento e novas tentativas

Mecanismos para executar execuções de DAG em datas anteriores.

O catchup executa execuções de DAG que ainda não foram executadas. Por exemplo, se o DAG foi pausado por um longo período e depois retomado. É possível usar o preenchimento para executar execuções de DAG em um determinado período. As novas tentativas especificam quantas tentativas o Airflow precisa fazer ao executar tarefas de um DAG.

A programação funciona da seguinte maneira:

1. Depois que a data de início passa, o Airflow aguarda a próxima ocorrência do intervalo de programação.

2. O Airflow programa a primeira execução do DAG para acontecer no final desse intervalo.

   Por exemplo, se um DAG estiver programado para ser executado a cada hora e a data de início for às 12h de hoje, a primeira execução do DAG ocorrerá às 13h de hoje.

A seção Programar um DAG do Airflow deste documento descreve como configurar a programação dos seus DAGs usando esses conceitos. Para mais informações sobre execuções e programação de DAGs, consulte Execuções de DAG na documentação do Airflow.

Sobre como acionar um DAG

O Airflow fornece as seguintes maneiras de acionar um DAG:

• Acionar com base em uma programação. O Airflow aciona o DAG automaticamente com base na programação especificada no arquivo DAG.

• Acionar manualmente. É possível acionar um DAG manualmente no console doGoogle Cloud , na interface do Airflow ou executando um comando da CLI do Airflow na Google Cloud CLI.

• Acionamento em resposta a eventos. A maneira padrão de acionar um DAG em resposta a eventos é usando um sensor.

Outras maneiras de acionar DAGs:

• Acionar de maneira programática. É possível acionar um DAG usando a API REST do Airflow. Por exemplo, em um script em Python.

• Acionar de maneira programática em resposta a eventos. É possível acionar DAGs em resposta a eventos usando as funções do Cloud Run e a API REST do Airflow.

Antes de começar

• Verifique se sua conta tem um papel que pode gerenciar objetos nos buckets do ambiente e visualizar e acionar DAGs. Para mais informações, consulte Controle de acesso.

Programar um DAG do Airflow

Você define uma programação para um DAG no arquivo dele. Edite a definição do DAG da seguinte maneira:

1. Localize e edite o arquivo DAG no computador. Se você não tiver o arquivo DAG, baixe uma cópia do bucket do ambiente. Para um novo DAG, você pode definir todos os parâmetros ao criar o arquivo DAG.

2. No parâmetro schedule_interval, defina a programação. Use uma expressão Cron, como 0 0 * * *, ou uma predefinição, como @daily. Para mais informações, consulte Cron e intervalos de tempo na documentação do Airflow.

   O Airflow determina datas lógicas para execuções de DAG com base na programação definida.

3. No parâmetro start_date, defina a data de início.

   O Airflow determina a data lógica da primeira execução do DAG usando esse parâmetro.

4. (Opcional) No parâmetro catchup, defina se o Airflow precisa executar todas as execuções anteriores desse DAG, da data de início até a data atual, que ainda não foram executadas.

   As execuções de DAG realizadas durante o catchup terão a data lógica no passado, e a data de execução vai refletir o momento em que a execução do DAG foi realizada.

5. (Opcional) No parâmetro retries, defina quantas vezes o Airflow precisa tentar novamente as tarefas com falha. Cada DAG consiste em uma ou mais tarefas individuais. Por padrão, as tarefas no Cloud Composer são executadas novamente duas vezes.

6. Faça upload da nova versão do DAG para o bucket do ambiente.

7. Aguarde até que o Airflow analise o DAG. Por exemplo, é possível verificar a lista de DAGs no seu ambiente noGoogle Cloud console ou na interface do Airflow.

A definição de DAG de exemplo a seguir é executada duas vezes por dia, às 00:00 e às 12:00. A data de início é definida como 1º de janeiro de 2024, mas o Airflow não executa datas anteriores depois que você faz o upload ou pausa o DAG porque o catchup está desativado.

O DAG contém uma tarefa chamada insert_query_job, que insere uma linha em uma tabela com o operador BigQueryInsertJobOperator. Esse operador é um dos operadores do BigQueryGoogle Cloud , que podem ser usados para gerenciar conjuntos de dados e tabelas, executar consultas e validar dados. Se uma execução específica dessa tarefa falhar, o Airflow vai tentar mais quatro vezes com o intervalo padrão. A data lógica dessas novas tentativas permanece a mesma.

A consulta SQL dessa linha usa modelos do Airflow para gravar a data lógica e o nome do DAG na linha.

import datetime

from airflow.models.dag import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

with DAG(
  "bq_example_scheduling_dag",
  start_date=datetime.datetime(2024, 1, 1),
  schedule_interval='0 */12 * * *',
  catchup=False
  ) as dag:

  insert_query_job = BigQueryInsertJobOperator(
    task_id="insert_query_job",
    retries=4,
    configuration={
        "query": {
            # schema: date (string), description (string)
            # example row: "20240101T120000", "DAG run: <DAG: bq_example_scheduling_dag>"
            "query": "INSERT example_dataset.example_table VALUES ('{{ ts_nodash }}', 'DAG run: {{ dag }}' )",
            "useLegacySql": False,
            "priority": "BATCH",
        }
    },
    location="us-central1"
  )

  insert_query_job

Para testar esse DAG, acioná-lo manualmente e ver os registros de execução de tarefas.

Mais exemplos de parâmetros de programação

Veja a seguir exemplos de como funciona a programação com diferentes combinações de parâmetros:

• Se start_date for datetime(2024, 4, 4, 16, 25) e schedule_interval for 30 16 * * *, a primeira execução do DAG acontecerá às 16h30 de 5 de abril de 2024.

• Se start_date for datetime(2024, 4, 4, 16, 35) e schedule_interval for 30 16 * * *, a primeira execução do DAG acontecerá às 16h30 em 6 de abril de 2024. Como a data de início é após o intervalo da programação em 4 de abril de 2024, a execução do DAG não acontecerá no dia 5 de abril de 2024. Em vez disso, o intervalo da programação termina às 16h35 do dia 5 de abril de 2024. Portanto, a próxima execução do DAG é programada para às 16h30 do dia seguinte.

• Se start_date for datetime(2024, 4, 4) e schedule_interval for @daily, a primeira execução do DAG será programada para às 00h do dia 5 de abril de 2024.

• Se start_date for datetime(2024, 4, 4, 16, 30) e schedule_interval for 0 * * * *, a primeira execução do DAG será programada para às 18h do dia 4 de abril de 2024. Após a data e hora especificadas, o Airflow programa uma execução do DAG para ocorrer no minuto 0 de cada hora. O momento mais próximo em que isso acontece é 17h. Nesse momento, o Airflow programa uma execução do DAG para acontecer no final do intervalo da programação, ou seja, às 18h.

Acionar um DAG manualmente

Quando você aciona um DAG do Airflow manualmente, ele executa o DAG uma vez, independente da programação especificada no arquivo DAG.

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags trigger -- DAG_ID

Ver registros e detalhes da execução do DAG

No console Google Cloud , é possível:

• Confira os status das execuções anteriores do DAG e os detalhes do DAG.
• Analise registros detalhados de todas as execuções e tarefas do DAG.
• Confira as estatísticas do DAG.

Além disso, o Cloud Composer oferece acesso à interface do Airflow, que é a interface da Web do Airflow.

Pausar um DAG

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags pause -- DAG_ID

A seguir

• Escrever DAGs do Airflow
• Testar DAGs do Airflow
• Resolver problemas de DAGs do Airflow