LOTEP · Plataforma de dados
Ingestion and loading of transactions for LOTEP, the state lottery of Paraíba, from a Kafka backed API to an Airflow orchestrated pipeline
Private repository
About the project
The platform receives lottery transactions (bets, prizes, points of sale, and lottery operators), validates each payload, and carries them into a PostgreSQL data warehouse the operation is analyzed from. Two ingestion paths coexist: a real time one over the API, and a batch load of files submitted by operators. The real time path is a JWT authenticated FastAPI service that publishes to Kafka asynchronously, with backpressure protection: when the buffer fills up, the API responds with 503 instead of accepting more than it can process. A worker consumes the topic and persists it to the database. The payload format migrated from XML to JSON validated by Pydantic, with monetary values kept as decimals to avoid losing precision, and each transaction gets a deterministic protocol number generated by HMAC. The batch path evolved from a simple loading script into an Airflow orchestrated pipeline. The extractor fetches files from a remote server, including entire subdirectories, detects each file's header and layout, processes it in parallel batches, and logs an audit trail per file. A modification date watermark guarantees incremental loading, without reprocessing what has already been ingested.
How it works
On the API path, the request is authenticated, validated against the Pydantic models, and published to the Kafka topic. The producer is asynchronous and monitors its own buffer, returning a temporary unavailability response when it's saturated, which preserves the integrity of the queue. The consumer worker reads the topic and writes to the destination PostgreSQL. On the batch path, an Airflow DAG triggers the loader: the extractor lists the file server, compares each subdirectory's modification date against the watermark stored in a control table, and downloads only what changed. The header detector and pattern matcher identify the layout, records are processed in parallel batches, and the load is logged with a per file success and error log, while also flagging data that changed for controlled reprocessing. The Airflow infrastructure runs in full, with a scheduler, Celery worker, triggerer, and Redis as the broker, and keeps its metadata database separate from the destination database. The same loader code runs both inside and outside the orchestration layer.
Architecture
- 01
FastAPI service with JWT authentication: receives transactions and publishes to Kafka
- 02
Asynchronous producer with backpressure: responds 503 when the buffer saturates
- 03
Kafka consumer worker: persists to the destination PostgreSQL
- 04
Pydantic models in JSON, replacing the old XML schemas
- 05
ETL loader: remote file extractor, header detector, and pattern matcher
- 06
Parallel batch processing with a modification date watermark
- 07
Airflow with scheduler, Celery worker, triggerer, and Redis, metadata kept in a separate database
- 08
Per file audit trail, with success and error logs
Features
- Transaction ingestion via authenticated API, publishing to Kafka
- Explicit backpressure when the queue saturates, instead of accepting and losing data
- Payload migration from XML to validated JSON, with decimals for monetary values
- Deterministic protocol number per transaction
- Incremental loading of remote files via a modification watermark
- Header and pattern detection per file
- Parallel batch processing
- Airflow orchestration with a per file audit trail
Tech stack
- Backend
- Python · FastAPI · Pydantic
- Infrastructure
- Kafka · Apache Airflow · Celery · Redis · Docker
- Database
- PostgreSQL · SQLAlchemy
- Data
- pandas
- Tools
- pytest