Building an API for +217 Million Public Records (CNPJ, PGFN, CNO, and CAFIR)

Processing open government data in Brazil is a true test for any software engineer. Government portals release massive monthly dumps that are unstable, lack delta endpoints, and are often packed in legacy formats with corrupted characters.

In this article, I share how I architected CNPJ-API: a complete ecosystem for collecting, parsing, storing, and serving Brazil’s major open datasets through an API — covering Federal Revenue CNPJ, PGFN Active Debt, National Registry of Works (CNO), Rural Property Registry (CAFIR), and RFB Tax Credits.

1. Context: The Challenge of Brazilian Open Data

The open datasets provided by the Federal Revenue and PGFN are rich, but present severe operational hurdles:

• Massive Volume: Just the CNPJ registry (Companies, Establishments, Partners, and Simples Nacional) exceeds 217 million records, generating over 80 GB of indexed relational data.
• Lack of Delta APIs (No CDC): The tax authority does not report daily changes; they publish a monthly full snapshot across dozens of .zip files.
• Data Quality Issues: CSV files encoded in LATIN1, semicolon-delimited, containing null bytes (\x00), and historical inconsistencies that break standard parsers.
• Download Instability: Government servers suffer frequent outages, throttled bandwidth, and dropped connections mid-download on multi-gigabyte files.

The goal was clear: run this entire pipeline in a 100% automated, resilient, and cost-effective way on a single VPS, without expensive Spark clusters or bulky Big Data tooling.

2. Project Goals

Transform disparate, raw government dumps into a unified platform queryable in milliseconds:

1. Autonomous Collectors: Download, validate, and ingest datasets idempotently and fault-tolerantly.
2. Zero-Downtime Updates: Update the monthly database without taking the query API offline.
3. Low-Disk Ingestion: Stream tens of gigabytes directly without unzipping massive intermediate CSVs onto VPS disk storage.
4. High-Performance API: Expose REST endpoints via FastAPI for structured queries by CNPJ, Corporate Structure (QSA), Tax Debts, Civil Works, and Rural Properties.

3. Architecture & Engineering Decisions

To prevent silent failures and corrupted data publishing, the system was designed with strict modular layers:

       External Sources (RFB, PGFN, CNO, CAFIR)
                         │
                         ▼
             [ Collector / Source Adapters ]
     (Discovery ➔ Probe ➔ Stream Download & Checksum)
                         │
                         ▼
             [ Streaming Pipeline ]
       (Null Byte Filtering ➔ Direct PostgreSQL COPY)
                         │
                         ▼
           [ PostgreSQL 16 (Partitioned) ]
   (Temporary Staging ➔ Semantic Validation ➔ Atomic Publish)
                         │
                         ▼
              [ FastAPI + Redis Cache ]
                (Queries < 50ms)

A. Streaming Unzip & Bulk Copy (Zero Disk Waste)

Rather than downloading 5 GB of ZIPs, extracting 20 GB of CSVs to disk, and then importing them (requiring over 30 GB of free scratch disk space), I implemented an in-memory streaming pipeline:

• The collector opens the ZIP archive as a stream (archive.open()).
• Chunks pass through a custom Python NullFilteringReader that strips \x00 null bytes on the fly.
• The stream is piped directly into PostgreSQL using the native high-speed COPY FROM STDIN WITH (FORMAT csv, DELIMITER ';').

This achieves an ingestion throughput of tens of thousands of rows per second with zero disk bloat.

B. Monthly Partitioning & Atomic Publishing

To ensure the API never serves half-loaded data during a 2-hour collection run, I adopted time-based snapshot partitioning:

• Data for 2026-06 is loaded into dedicated partitioned tables (cnpj.empresas_2026_06, cnpj.estabelecimentos_2026_06).
• A pointer table (cnpj.snapshot_atual) tracks the active snapshot.
• Semantic Validation: Before publishing, the system checks record counts against historical baselines (e.g. failing automatically if records drop by more than 30% compared to the prior month).
• Atomic Switch: Only when all ingestion and integrity validations pass, the pointer switches to the new month. If a run fails midway, the live API remains unaffected on the previous month’s snapshot.

C. Strict State Machine (collection_runs)

Every execution follows a strictly audited state lifecycle in PostgreSQL:

PENDING ➔ DISCOVERING ➔ PROBING ➔ DOWNLOADING ➔ DOWNLOADED ➔ VALIDATING_FILE ➔ PARSING ➔ LOADING_STAGING ➔ VALIDATING_DATA ➔ PUBLISHING ➔ SUCCESS

If a download is interrupted or an upstream schema changes, the exact failure step, URL, retry count, and error payload are logged for immediate troubleshooting.

4. Integrated Datasets

The platform currently consolidates 5 major public sources:

1. Federal Revenue (CNPJ Full): Companies, Establishments, Partners (QSA), Simples Nacional/MEI, CNAEs, Municipalities, and Legal Natures.
2. PGFN (Federal Active Debt): Non-social security, social security, and FGTS debtors with consolidated debt amounts and litigation status.
3. CNO (National Registry of Works): Registered construction works across Brazil, areas, responsible entities, and CNAEs.
4. CAFIR (Rural Property Registry): Registered rural estates, total area, and cadastral status.
5. RFB (Tax Credits & Transactions): Active tax credits, payment installments, and individual tax transactions.

5. Results & Key Takeaways

• +217 Million Managed Records: Complete Brazilian open business database running smoothly on PostgreSQL 16.
• Sub-50ms Response Times: Comprehensive CNPJ lookups (including partners, address, tax status, and active debts) returning in under 50ms.
• Resource Efficiency: The entire pipeline runs on a standard VPS with Docker Compose, utilizing distributed Redis locks to avoid collector concurrency conflicts.
• Extensibility: Source Adapters make it straightforward to add new public sources by implementing discovery, download, and parse steps without touching the core API.

6. Next Steps

• Implement an administrative monitoring dashboard for collector observability (update SLAs, consecutive failures, execution metrics).
• Optimize delta/upsert ingestion to further reduce month-over-month storage footprint.
• Expose webhooks for automated notifications on corporate status and tax standing changes.

Questions or thoughts on data engineering and distributed systems? Feel free to reach out or leave a comment!