stream-windows

Motor de streaming / ETL em Python: agrega um fluxo de eventos em janelas tumbling (fixas) por chave — count, soma, mín, máx e média — com watermark para fechar janelas e tratamento de eventos atrasados. O núcleo é puro e tem cobertura de testes 100%. Fonte real via Apache Kafka.

eventos ─▶ filtro (E) ─▶ transformação (T) ─▶ janelas + agregação (L) ─▶ resultados
                                              └─ watermark fecha a janela ─┘

Destaques

• Janelas tumbling por chave — cada evento cai na janela [start, end) do seu timestamp; a agregação (count/sum/min/max/avg) é mantida por (chave, janela).
• Watermark — o maior timestamp visto governa o tempo. A janela fecha quando watermark >= fim + lateness, e o resultado é emitido. Nada de relógio de parede: o tempo é o do evento.
• Eventos atrasados — chegou depois que a janela fechou? É descartado e contado (late_events). Um lateness configurável dá uma folga antes de fechar.
• ETL de verdade — o pipeline filtra, transforma e agrega; cada etapa é plugável e testada isoladamente.
• Fonte atrás de uma porta — a API HTTP alimenta o motor por POST /ingest; o adapter Kafka (real) lê um tópico e é validado pelo docker-compose. O núcleo roda offline nos testes.
• Cobertura 100% — o Kafka (rede) fica fora da métrica.
• OpenAPI/Swagger em /docs.

Stack

• Linguagem: Python 3.12 · API: FastAPI
• Streaming: motor próprio de janelas + watermark · Fonte real: Apache Kafka
• Testes/cobertura: pytest + pytest-cov (gate de 100%) · Lint: ruff

Arquitetura

app/
├── domain/          # Event, WindowResult, Accumulator (Python puro)
├── engine/          # TumblingWindows (janelas + watermark)
├── application/     # Pipeline (filtro/transformação) + ports (Source, Sink)
├── infrastructure/  # sink em memória, KafkaSource (real)
├── api/             # FastAPI: ingest/flush/windows/stats
└── main.py          # composição

Como rodar

python -m venv .venv && . .venv/Scripts/activate   # (Linux/mac: source .venv/bin/activate)
pip install -r requirements.txt
uvicorn app.main:app --reload   # http://localhost:8000  (Swagger em /docs)

Envie eventos e veja a janela fechar quando o tempo avança:

curl -X POST localhost:8000/ingest -H "content-type: application/json" -d '{"events":[
  {"key":"sensor-1","value":20,"ts":5},
  {"key":"sensor-1","value":40,"ts":30},
  {"key":"sensor-1","value":10,"ts":65}
]}'
# -> fecha a janela [0,60): count=2, average=30  (o ts=65 abre a próxima)

curl localhost:8000/windows
curl localhost:8000/stats     # late_events, open_windows

WINDOW_SIZE e ALLOWED_LATENESS (segundos) vêm do ambiente.

Streaming real com Kafka

docker compose --profile real up --build
# o "runner" consome o tópico "events" e imprime cada janela fechada

Endpoints

Método	Rota	Descrição
POST	/ingest	Ingesta eventos; devolve as janelas fechadas
POST	/flush	Fecha todas as janelas abertas
GET	/windows	Lista as janelas já fechadas
GET	/stats	late_events e open_windows
GET	/docs	Swagger UI

Testes e cobertura

pip install -r requirements-dev.txt
pytest            # exige 100% (configurado no pyproject.toml)
ruff check . && ruff format --check .

Licença

MIT.