Airflow Patterns

Padrões de orquestração com Apache Airflow 3 que uso no dia a dia.

Esse repositório reune exemplos funcionais de DAGs, sensores e plugins que resolvo problemas recorrentes em pipelines de dados: ingestão com retry, transformação em camadas, observabilidade e espera por dependências externas. O código é genérico, sem referência a nenhum sistema real.

---

O que tem aqui

Arquivo	O que demonstra
dags/bronze_to_gold_pipeline.py	TaskFlow API, pipeline em camadas, Params, callback de falha
dags/api_ingestion_dag.py	Ingestao paginada de API REST, retry com urllib3, escrita em Parquet
dags/sensors/gcs_partition_sensor.py	Sensor customizado que aguarda partição no GCS
plugins/audit_listener.py	Listener de auditoria com logs JSON estruturados por task

---

Como rodar

Requisitos: Docker e Docker Compose instalados.

# 1. Suba o banco e inicialize o Airflow
docker compose up airflow-init

# 2. Suba os demais servicos
docker compose up -d

# 3. Acesse a UI
# http://localhost:8080  (usuario: admin / senha: admin)

Para rodar sem Docker, crie um ambiente virtual e instale as dependencias:

python -m venv .venv
source .venv/bin/activate       # Windows: .venv\Scripts\activate
pip install -r requirements.txt

export AIRFLOW_HOME=$(pwd)
airflow db migrate
airflow users create --username admin --password admin \
  --firstname Admin --lastname User --role Admin --email admin@example.com
airflow standalone

---

Estrutura

airflow-patterns/
├── dags/
│   ├── bronze_to_gold_pipeline.py
│   ├── api_ingestion_dag.py
│   └── sensors/
│       └── gcs_partition_sensor.py
├── plugins/
│   └── audit_listener.py
├── docker-compose.yaml
├── requirements.txt
└── README.md

---

Conceitos demonstrados

TaskFlow API As DAGs usam @dag e @task do Airflow 3. Isso elimina o XCom manual, deixa o fluxo de dados explícito no codigo e reduz boilerplate. Cada @task recebe e retorna Python nativo; o Airflow cuida da serialização.

Pipeline em camadas (Bronze / Silver / Gold) O bronze_to_gold_pipeline.py segue a arquitetura medalhao: dados brutos sao extraídos, limpos e transformados em cada camada com responsabilidades distintas. A camada Gold entrega dados agregados prontos para consumo analitico.

Sensores customizados O GCSPartitionSensor herda de BaseSensorOperator e implementa o método poke. Tem poke_interval e timeout configuráveis, log detalhado em cada verificacao e aceita templates Jinja nos parametros.

Listeners e auditoria O AuditListenerPlugin usa o sistema de hooks do Airflow para interceptar transicoes de estado (running, success, failed) sem modificar nenhuma DAG. Cada evento emite um JSON com dag_id, task_id, run_id, duracao e, quando aplicavel, o tipo e mensagem da excecao.

Retry e resiliencia A DAG de ingestao usa HTTPAdapter com Retry do urllib3 configurado por parametro. O default_args das DAGs define retry_exponential_backoff=True para nao sobrecarregar APIs externas em caso de falha transiente.

Airflow Variables URLs e tokens de API sao lidos via Variable.get, evitando hardcode no codigo e facilitando a troca de ambiente sem alterar DAGs.