Orquestrador de DAGs do Airflow

O que Orquestrador de DAGs do Airflow faz?

Projete e opere DAGs do Apache Airflow para orquestração confiável de pipelines de dados com gerenciamento de dependências, SLAs e monitoramento.

Gatilhos de Ativação

Ativar em: "Airflow", "DAG", "operator", "sensor", "scheduler", "task dependency", "SLA", "backfill", "XCom", "TaskFlow API", "MWAA", "Cloud Composer"

NÃO para: Execução de modelos dbt → dbt-analytics-engineer (embora o Airflow possa acionar dbt) | Processamento de streams → streaming-pipeline-architect | Engine de workflow (Temporal) → distributed-transaction-manager

Início Rápido

1. Definir DAG — usar TaskFlow API (decoradores @dag, @task) para DAGs nativos em Python
2. Configurar agendamento — cron ou timetable, com catchup=False a menos que backfill seja intencional
3. Configurar retentativas — retries=2, retry_delay=timedelta(minutes=5) em cada tarefa
4. Adicionar SLAs — sla=timedelta(hours=2) em tarefas de caminho crítico
5. Testar localmente — airflow dags test my_dag 2026-01-01 antes de implantar

Capacidades Principais

Domínio	Tecnologias
Airflow	Apache Airflow 2.10+, MWAA, Cloud Composer 3
Operadores	BashOperator, PythonOperator, KubernetesPodOperator
Provedores	apache-airflow-providers-{snowflake, google, aws, dbt-cloud}
Executores	CeleryExecutor, KubernetesExecutor, LocalExecutor
Monitoramento	Perdas de SLA, duração de tarefas, métricas do Airflow → Prometheus

Padrões Arquiteturais

DAG com TaskFlow API

from airflow.decorators import dag, task
from datetime import datetime, timedelta

@dag(
    schedule="0 6 * * *",          # diariamente às 6h UTC
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "sla": timedelta(hours=2),
    },
    tags=["finance", "daily"],
)
def daily_revenue_pipeline():

    @task()
    def extract_payments() -> dict:
        """Extrair da API Stripe"""
        data = stripe_client.list_payments(date=today())
        return {"count": len(data), "path": "s3://raw/payments/"}

    @task()
    def extract_orders() -> dict:
        """Extrair da API Shopify"""
        data = shopify_client.list_orders(date=today())
        return {"count": len(data), "path": "s3://raw/orders/"}

    @task()
    def transform(payments: dict, orders: dict) -> str:
        """Juntar e transformar no DuckDB"""
        result_path = run_duckdb_transform(payments["path"], orders["path"])
        return result_path

    @task()
    def load(path: str):
        """Carregar no Snowflake"""
        snowflake_copy_into("fct_revenue", path)

    # Definir dependências via chamadas de função
    payments = extract_payments()
    orders = extract_orders()
    transformed = transform(payments, orders)
    load(transformed)

daily_revenue_pipeline()

Mapeamento Dinâmico de Tarefas (Fan-Out/Fan-In)

@task()
def get_partitions() -> list[str]:
    return ["2026-01-01", "2026-01-02", "2026-01-03"]

@task()
def process_partition(partition_date: str) -> dict:
    """Executa em paralelo para cada partição"""
    return {"date": partition_date, "rows": process(partition_date)}

@task()
def aggregate(results: list[dict]):
    """Fan-in: recebe todos os resultados das partições"""
    total = sum(r["rows"] for r in results)
    log.info(f"Processadas {total} linhas no total")

# Mapeia dinamicamente process_partition para todas as partições
partitions = get_partitions()
results = process_partition.expand(partition_date=partitions)
aggregate(results)

Integração dbt + Airflow

from airflow.operators.bash import BashOperator
from cosmos import DbtDag, ProjectConfig, ProfileConfig

# Opção 1: cosmos (recomendado)
dbt_dag = DbtDag(
    project_config=ProjectConfig("/opt/airflow/dbt/"),
    profile_config=ProfileConfig(
        profile_name="default",
        target_name="prod",
    ),
    schedule="@daily",
    dag_id="dbt_daily",
)

# Opção 2: BashOperator (simples)
dbt_run = BashOperator(
    task_id="dbt_run",
    bash_command="cd /opt/airflow/dbt && dbt build --select tag:daily",
)

Antipadrões

1. Tarefas pesadas — tarefas devem ser unidades atômicas; não coloque extract+transform+load em uma única tarefa
2. XCom para dados grandes — XCom é para metadados (caminhos, contagens); passe dados via S3/GCS, não via pickles do XCom
3. catchup=True acidentalmente — a menos que você queira backfill, defina catchup=False; senão o Airflow executa cada intervalo perdido
4. Sem retentativas — falhas transitórias são comuns; sempre defina retries >= 1 com um atraso
5. Código de nível superior em arquivos DAG — arquivos DAG são analisados a cada 30s; imports pesados ou chamadas de API no nível do módulo lentificam o agendador

Checklist de Qualidade

• TaskFlow API usada para DAGs nativos em Python (não encadeamento legado de operadores)
• catchup=False a menos que backfill seja intencional
• Retentativas configuradas em todas as tarefas (mínimo 1 retentativa)
• SLA definido em tarefas de caminho crítico
• XCom passa referências (caminhos S3), não cargas de dados
• Mapeamento dinâmico de tarefas usado para paralelismo fan-out
• DAG testado localmente com airflow dags test antes da implantação
• Idempotência de tarefa: reexecutar uma tarefa produz o mesmo resultado
• Alertas configurados para falhas de tarefa e perdas de SLA
• Tags de DAG aplicadas para filtragem na interface do Airflow