LOTEP · Plataforma de dados
Ingestão e carga das transações da Loteria do Estado da Paraíba, de API com Kafka a pipeline orquestrado com Airflow
Repositório privado
Sobre o projeto
A plataforma recebe as transações do jogo (apostas, prêmios, pontos de venda e operadores lotéricos), valida cada payload e as leva até um data warehouse PostgreSQL de onde a operação é analisada. São dois caminhos de entrada que convivem: o tempo real, por API, e a carga em lote de arquivos enviados pelos operadores. O caminho em tempo real é uma API FastAPI autenticada por JWT que publica em Kafka de forma assíncrona, com proteção de contrapressão: quando o buffer enche, a API responde 503 em vez de aceitar o que não vai conseguir processar. Um worker consome o tópico e persiste no banco. O formato de payload migrou de XML para JSON validado por Pydantic, com valores monetários em decimal para não perder precisão, e cada transação recebe um protocolo determinístico gerado por HMAC. O caminho em lote evoluiu de um script de carga para um pipeline orquestrado por Airflow. O extrator busca arquivos de um servidor remoto, inclusive subdiretórios inteiros, detecta o cabeçalho e o padrão de cada arquivo, processa em lotes paralelos e registra auditoria por arquivo. Uma marca d'água por data de modificação garante carga incremental, sem reprocessar o que já entrou.
Como funciona
Na entrada por API, a requisição é autenticada, validada contra os modelos Pydantic e publicada no tópico Kafka. O produtor é assíncrono e monitora o próprio buffer, devolvendo indisponibilidade temporária quando está saturado, o que preserva a integridade da fila. O worker consumidor lê o tópico e escreve no PostgreSQL de destino. Na carga em lote, uma DAG do Airflow dispara o loader: o extrator lista o servidor de arquivos, compara a data de modificação de cada subdiretório com a marca d'água guardada em tabela de controle e baixa apenas o que mudou. O detector de cabeçalho e o casador de padrões identificam o layout, os registros são processados em lotes paralelos e a carga é registrada com log de sucesso e erro por arquivo, além de sinalizar dados que sofreram modificação para reprocessamento controlado. A infraestrutura do Airflow roda completa, com scheduler, worker Celery, triggerer e Redis como intermediário, e mantém o banco de metadados separado do banco de destino. O mesmo código do loader é usado dentro e fora da orquestração.
Arquitetura
- 01
API FastAPI com autenticação JWT: recebe transações e publica em Kafka
- 02
Produtor assíncrono com contrapressão: responde 503 quando o buffer satura
- 03
Worker consumidor do Kafka: persiste no PostgreSQL de destino
- 04
Modelos Pydantic em JSON no lugar dos antigos schemas XML
- 05
Loader ETL: extrator de arquivos remotos, detector de cabeçalho e casador de padrões
- 06
Processamento em lotes paralelos com marca d'água por data de modificação
- 07
Airflow com scheduler, worker Celery, triggerer e Redis, com metadados em banco separado
- 08
Auditoria por arquivo processado, com log de sucesso e de erro
Funcionalidades
- Ingestão de transações por API autenticada, publicando em Kafka
- Contrapressão explícita quando a fila satura, em vez de aceitar e perder
- Migração de payload de XML para JSON validado, com decimal para valores monetários
- Protocolo determinístico por transação
- Carga incremental de arquivos remotos por marca d'água de modificação
- Detecção de cabeçalho e de padrão por arquivo
- Processamento em lotes paralelos
- Orquestração por Airflow com auditoria por arquivo
Tecnologias
- Backend
- Python · FastAPI · Pydantic
- Infrastructure
- Kafka · Apache Airflow · Celery · Redis · Docker
- Database
- PostgreSQL · SQLAlchemy
- Data
- pandas
- Tools
- pytest