Autorando DAGs do Airflow

O que Autorando DAGs do Airflow faz?

Quando usar

• Ao criar ou refatorar DAGs e tarefas do Airflow.
• Ao configurar agendamentos, catchup/backfill, retentativas e SLAs.
• Ao passar dados entre tarefas (XCom) ou usar conexões/variáveis.
• Não use para diagnosticar um DAG em execução com falha (use debugging-airflow-pipelines).

Fluxo de trabalho

- [ ] Torne cada tarefa idempotente e parametrizada pelo intervalo de dados
- [ ] Mantenha código pesado/de import dentro das tarefas, não no nível superior do módulo
- [ ] Defina schedule + catchup deliberadamente
- [ ] Configure retentativas, retry_delay e SLAs
- [ ] Conecte dependências via valores de retorno da TaskFlow ou operadores >>

1. Tarefas idempotentes — uma tarefa para o intervalo 2026-01-15 deve produzir o mesmo resultado independentemente de ser executada uma vez ou reexecutada. Use o intervalo de dados, não datetime.now().
2. Sem código pesado no nível superior — o scheduler parseia cada arquivo de DAG frequentemente; chamadas a banco de dados, chamadas a API ou imports grandes no nível do módulo tornam o agendamento lento e podem quebrar o parsing. Coloque-os dentro das tarefas.
3. Schedule + catchup intencionalmente — catchup=True faz backfill de todos os intervalos perdidos desde start_date; defina como False a menos que você queira isso.
4. Retentativas e SLAs — falhas transitórias são normais; defina retries e retry_delay; use SLAs/alertas para atrasos.

Padrões

DAG TaskFlow, idempotente e com dependências claras:

from airflow.decorators import dag, task
import pendulum

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=5)},
    tags=["orders"],
)
def orders_pipeline():

    @task
    def extract(data_interval_start=None):
        # Use o intervalo, não now(), para que reexecuções sejam determinísticas.
        return fetch_orders(day=data_interval_start.date())

    @task
    def load(rows):
        # Delete-insert da partição -> idempotente em caso de retentativa.
        overwrite_partition("fct_orders", rows)

    load(extract())

orders_pipeline()

Passe dados pequenos via XCom (valores de retorno); dados grandes via armazenamento — escreva para S3/GCS/warehouse e passe o caminho/chave, nunca megabytes através de XCom.

Use conexões/variáveis para segredos e configuração (BaseHook.get_connection, Variable.get), nunca credenciais hard-coded.

Armadilhas comuns

• Chamadas a API/DB ou imports pesados no nível superior — tornam o scheduler lento e podem falhar o parsing dos DAGs em todo o ambiente.
• datetime.now() dentro de tarefas — quebra a idempotência e os backfills; use data_interval_start/_end.
• catchup=True não intencional — inunda o cluster com execuções históricas na primeira implantação.
• Payloads grandes através de XCom — incha o banco de dados de metadados; passe referências.
• start_date dinâmico (ex.: days_ago) — torna os agendamentos não determinísticos; use um timestamp fixo.
• Uma tarefa monstro — divida extract/transform/load para que as retentativas sejam granulares.

Referências

• Referência de agendamento, catchup e backfill