Melhores Práticas para DAGs em Produção

Se preferir começar por um resumo, peça um TL;DR ao ChatGPT ou ao Claude .

Critérios para reduzir duplicação, saída parcial, espera sem limite e exposição de credenciais em DAGs de produção.

Índice

  1. 1. Idempotência
  2. 2. Atomicidade
  3. 3. Proteja credenciais e secrets
  4. 4. Configurar Timeouts Adequados
  5. 5. Sensores: Escolha o Modo Correto
  6. Checklist de Produção
  7. Referências

1. Idempotência

Projete cada task para que repeti-la com a mesma entrada e o mesmo intervalo lógico produza o mesmo estado de negócio esperado. Essa propriedade permite que retries e reprocessamentos convirjam, em vez de acumularem registros ou efeitos laterais.

Idempotência reduz o risco de duplicação durante retries e backfills, mas não garante um reprocessamento seguro. Versão do código, dados de entrada, concorrência e efeitos em sistemas externos também precisam ser controlados.

O exemplo abaixo pressupõe uma restrição UNIQUE sobre totais_diarios(data_intervalo). Para a mesma data e o mesmo total, a segunda execução mantém o estado de negócio produzido pela primeira:

1
2
3
4
5
6
7
8
9
10

def carregar_total(data_intervalo, total):
    db.execute(
        """
        INSERT INTO totais_diarios (data_intervalo, total)
        VALUES (%s, %s)
        ON CONFLICT (data_intervalo)
        DO UPDATE SET total = EXCLUDED.total
        """,
        (data_intervalo, total),
    )

Na prática:

  • leia e escreva partições identificadas pelo intervalo da execução, em vez de usar implicitamente o dado “mais recente”;
  • use UPSERT, substituição de partição ou escrita atômica quando uma repetição não puder criar duplicatas;
  • agrupe alterações dependentes em uma transação quando o destino oferecer esse recurso;
  • associe chamadas externas a uma chave de idempotência ou deduplicação;
  • diferencie estado de negócio de metadados operacionais, como horário e número da tentativa;
  • teste a task duas vezes para o mesmo intervalo e compare registros, arquivos, mensagens e demais efeitos observáveis.

Quando o destino não oferece transações ou deduplicação, documente o modo de falha e implemente reconciliação ou compensação. Marcar a task como concluída no Airflow não desfaz um efeito externo parcial.

2. Atomicidade

Uma task é atomicamente publicável quando termina com toda a saída esperada ou não deixa uma saída parcial disponível aos consumidores. Fazer “uma coisa” pode melhorar a compreensão, mas não garante atomicidade.

Padrões possíveis incluem:

  • escrever em staging e promover o resultado depois da validação;
  • usar transação para alterações que precisam ser confirmadas juntas;
  • gravar arquivo temporário e fazer rename ou troca atômica quando o storage oferecer essa garantia;
  • publicar marcador ou evento de conclusão somente após validar a saída;
  • remover ou reconciliar artefatos parciais antes de um retry.

Exemplo conceitual para publicar uma partição preparada em staging:

1
2
3
4
5
6
7
8
9
10
11

BEGIN;

DELETE FROM vendas
WHERE data_intervalo = :data_intervalo;

INSERT INTO vendas (data_intervalo, pedido_id, valor)
SELECT data_intervalo, pedido_id, valor
FROM vendas_staging
WHERE carga_id = :carga_id;

COMMIT;

O exemplo depende das garantias transacionais do banco e de uma staging já validada. Ele não cobre falha durante a extração nem efeitos em outro sistema.

Separar extração, transformação, carga e validação em tasks pode melhorar retry, observabilidade e isolamento de recursos quando cada fronteira possui um artefato persistido e verificável. Sem essa persistência, a divisão pode apenas aumentar transferência de dados e pontos de falha. Escolha a granularidade a partir da unidade de recuperação, não de uma regra de quantidade de operações.

3. Proteja credenciais e secrets

Não grave credenciais no código da DAG, em parâmetros versionados, logs ou XComs. Use uma Connection e controle onde seu valor sensível é armazenado e quais componentes podem recuperá-lo.

Connections são o modelo do Airflow para credenciais e configurações de acesso a serviços externos. Elas podem ser obtidas de:

  • um secrets backend externo, como Vault ou o gerenciador de secrets da nuvem;
  • variáveis de ambiente no formato AIRFLOW_CONN_<CONN_ID>;
  • banco de metadados do Airflow.

Um secrets backend é indicado quando a organização precisa centralizar rotação, auditoria e controle de acesso. Variáveis de ambiente podem ser adequadas quando o mecanismo de implantação as injeta apenas nos componentes que precisam delas. O banco de metadados pode ser aceitável em contextos compatíveis com seu modelo de ameaça, desde que Fernet, permissões, backup e rotação de chaves estejam configurados e testados.

Variable é um armazenamento genérico de configuração. Não a trate como equivalente a uma Connection apenas porque o valor pode ser criptografado ou resolvido por um secrets backend. Se uma Variable precisar conter informação sensível, aplique os mesmos controles de armazenamento, acesso, mascaramento e rotação.

No Airflow 3, uma task pode recuperar uma Connection pela API pública:

1
2
3
4
5
6
7
8
9
10

from airflow.sdk import Connection, task


@task
def chamar_servico():
    connection = Connection.get("servico_externo")
    token = connection.password

    # Use o token sem registrá-lo em logs nem retorná-lo por XCom.
    chamar_api(token=token)

Quando existir um Hook ou Operator para o serviço, prefira informar o conn_id a esse componente. Assim, a integração recupera a Connection sem espalhar a manipulação da credencial pelo código da DAG.

4. Configurar Timeouts Adequados

Timeouts transformam uma execução excessivamente longa em uma falha observável. Eles devem refletir o limite operacional do workflow, e não um valor único aplicado a todas as tasks.

Parâmetro Escopo Decisão necessária
execution_timeout Cada tentativa de uma task Quanto tempo uma tentativa pode executar antes de ser interrompida
dagrun_timeout DAG Run Quanto tempo o workflow completo pode permanecer em execução
timeout de sensor Espera total do sensor Quanto tempo a condição externa pode permanecer ausente

Para definir esses limites:

  • use a distribuição histórica de duração, incluindo picos esperados;
  • reserve tempo para retries e recuperação dentro do SLO do workflow;
  • diferencie processamento lento de espera por uma condição externa;
  • confirme se interromper a task deixa efeitos parciais que exigem limpeza ou reconciliação;
  • gere alerta com contexto suficiente para distinguir timeout, indisponibilidade da dependência e falta de capacidade.

O exemplo abaixo é hipotético. Ele supõe que o DAG deva terminar em até seis horas e que uma tentativa normalmente conclua em menos de duas horas. Esses valores precisam ser substituídos pelos limites observados e acordados para cada workflow:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from datetime import datetime, timedelta, timezone

from airflow.sdk import DAG, task


with DAG(
    dag_id="meu_dag",
    schedule="@daily",
    start_date=datetime(2026, 1, 1, tzinfo=timezone.utc),
    catchup=False,
    dagrun_timeout=timedelta(hours=6),
) as dag:

    @task(
        execution_timeout=timedelta(hours=2),
        retries=1,
        retry_delay=timedelta(minutes=10),
        retry_exponential_backoff=True,
    )
    def processar():
        ...

    processar()

Retries são apropriados para falhas transitórias, como indisponibilidade temporária ou limitação de taxa. Erros de autenticação, entrada inválida e falhas determinísticas tendem a repetir o mesmo resultado. Antes de habilitar retries, confirme que a task é repetível e que uma tentativa anterior não deixou um efeito externo sem deduplicação ou compensação.

5. Sensores: Escolha o Modo Correto

Não existe um limite universal de minutos que determine a escolha. Considere a duração esperada, a latência tolerada, o número de sensores simultâneos e a capacidade dos workers, do scheduler e do triggerer.

Contexto Opção a avaliar Limite relevante
Sensor com implementação deferrable e triggerer operacional Deferrable Libera o worker durante a espera, mas transfere trabalho ao triggerer
Polling periódico sem suporte a deferral reschedule Libera o worker entre verificações, com custo de reagendamento
Espera curta, baixa concorrência e baixa latência necessária poke Ocupa um worker slot durante toda a espera

Para cada sensor:

  • verifique primeiro se o provider oferece uma implementação deferrable;
  • confirme que o deployment executa e monitora o triggerer antes de ativar deferral;
  • defina poke_interval conforme a latência aceitável e o custo da consulta ao sistema externo;
  • configure timeout para transformar uma espera excessiva em falha observável;
  • considere backoff quando a frequência de polling puder diminuir ao longo da espera;
  • teste a quantidade máxima de sensores simultâneos sem impedir a execução das demais tasks.

deferrable=True não é um comportamento universal de todo sensor: o efeito depende da implementação do Operator ou Sensor. Confirme o suporte na versão do provider utilizada.

Veja o guia completo: Sensores — Poke, Reschedule e Deferral

Checklist de Produção

  • Retries para o mesmo intervalo não duplicam registros nem efeitos externos
  • Falhas não deixam saída parcial visível aos consumidores
  • Timeouts refletem duração observada, SLO e tempo de recuperação
  • Retries são usados apenas para falhas recuperáveis em tasks repetíveis
  • Sensores não retêm worker slots sem uma decisão explícita
  • Sensores têm intervalo, timeout e capacidade validados
  • Credenciais não estão no código, nos logs nem em XComs
  • Connections sensíveis usam armazenamento e acesso compatíveis com o modelo de ameaça
  • Falhas e atrasos críticos produzem alertas testados
  • catchup corresponde à política de processamento histórico do DAG
  • Dependências entre DAGs declaram semântica de evento, espera ou disparo

Referências

APACHE AIRFLOW. Best Practices — Airflow 2.10.2. Disponível em: https://airflow.apache.org/docs/apache-airflow/2.10.2/best-practices.html

APACHE AIRFLOW. Sensors. Disponível em: https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/sensors.html

APACHE AIRFLOW. Deferrable Operators & Triggers. Disponível em: https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/deferring.html

APACHE AIRFLOW. Tasks — Timeouts and Retry Policies. Disponível em: https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/tasks.html

APACHE AIRFLOW. Managing Connections. Disponível em: https://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html

APACHE AIRFLOW. Secrets Backend. Disponível em: https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/secrets-backend/index.html

APACHE AIRFLOW. Airflow Security Model. Disponível em: https://airflow.apache.org/docs/apache-airflow/stable/security/security_model.html

APACHE AIRFLOW. Public Interface for Airflow 3.0+. Disponível em: https://airflow.apache.org/docs/apache-airflow/stable/public-airflow-interface.html