ETL Framework Overview

​

Purpose

• Scale: Processing millions of entities (50M+ companies, 80M+ people)
• Reliability: Production-grade error handling, retry logic, and data validation
• Transparency: Full audit trails with IngestionRun nodes tracking every pipeline execution
• Extensibility: Plugin architecture for adding new data sources

​

Architecture

​

Key Features

​

Multi-Format Support

• CSV files: Headerless Receita Federal CSVs (; delimiter, latin-1 encoding)
• BigQuery exports: Base dos Dados datasets with automatic column mapping
• REST APIs: TSE, Câmara, Senado open data APIs
• Streaming ingestion: Memory-efficient chunk processing for 100GB+ datasets

​

Data Quality

• Document validation (CPF/CNPJ with check-digit verification)
• Name normalization (case, diacritics, whitespace)
• Date parsing (handles multiple Brazilian date formats)
• Schema validation using Pandera
• Deduplication on primary keys

​

Operational Monitoring

MATCH (r:IngestionRun)
WHERE r.source_id = 'cnpj'
RETURN r.status, r.rows_in, r.rows_loaded, r.started_at
ORDER BY r.started_at DESC
LIMIT 1

​

Pipeline Base Class

class Pipeline(ABC):
    name: str
    source_id: str

    @abstractmethod
    def extract(self) -> None:
        """Download raw data from source."""

    @abstractmethod
    def transform(self) -> None:
        """Normalize, deduplicate, and prepare data for loading."""

    @abstractmethod
    def load(self) -> None:
        """Load transformed data into Neo4j."""

    def run(self) -> None:
        """Execute the full ETL pipeline."""
        self.extract()
        self.transform()
        self.load()

​

Data Sources

​

Tech Stack

​

Dependencies

• neo4j>=5.27.0 — Database driver
• pandas>=2.2.0 — Data processing
• httpx>=0.28.0 — HTTP client for downloads
• click>=8.1.0 — CLI framework
• pydantic>=2.10.0 — Settings management
• pandera>=0.21.0 — Schema validation

• bigquery — Google BigQuery integration
• resolution — Splink entity resolution

​

Performance

• Batch mode: Loads full dataset into memory (up to 32GB)
• Streaming mode: Fixed 2GB memory footprint with chunking

​

Next Steps