cnpj-data-pipeline

[!IMPORTANT] Desde v1.3.2 — A Receita Federal migrou os arquivos CNPJ para um novo repositório Nextcloud. Esta versão já suporta a nova URL e realiza downloads via WebDAV automaticamente. Nenhuma configuração adicional necessária.

[!IMPORTANT] Desde v1.35.0 — Suporte ao CNPJ alfanumérico da Receita Federal, previsto para novas inscrições a partir de julho de 2026. cnpj_basico e cnpj_ordem aceitam 0-9 e A-Z; cnpj_dv continua numérico. CNPJs numéricos existentes seguem compatíveis sem mudança.

[!TIP] Novo — Estratégia de carga configurável. Use LOADING_STRATEGY=replace para carga completa mais rápida (TRUNCATE + INSERT) ou upsert (default) para manter disponibilidade durante a carga.

Requisitos

• uv — gerencia pacotes e versões do Python. Substitui pip e virtualenv
• just — roda os comandos do projeto. Substitui Makefile
• Docker — roda o PostgreSQL sem instalar banco local
• Python 3.11+ — uv instala automaticamente se necessário

Instalação

macOS (Homebrew):

brew install uv just

Linux / Windows: veja a instalação do uv e do just.

uv e just são binários prontos — não precisa de Rust ou compilação.

Início Rápido

cp .env.example .env
just up      # Iniciar PostgreSQL
just run     # Executar pipeline

Via Docker

Imagem pronta publicada a cada release no GitHub Container Registry. Não precisa clonar o repositório.

# Listar meses disponíveis
docker run --rm ghcr.io/caiopizzol/cnpj-data-pipeline --list

# Processar um mês em um Postgres seu
docker run --rm \
  -e DATABASE_URL=postgres://user:pass@host:5432/cnpj \
  ghcr.io/caiopizzol/cnpj-data-pipeline --month 2024-11

# Exportar para Parquet (sem banco)
docker run --rm \
  -e OUTPUT_FORMAT=parquet \
  -v $(pwd)/parquet:/app/parquet \
  ghcr.io/caiopizzol/cnpj-data-pipeline

O schema é aplicado automaticamente na primeira execução. Para rodar com o Postgres do projeto, use docker compose run --rm pipeline <args>.

Comandos

just install # Instalar dependências
just up      # Iniciar PostgreSQL
just down    # Parar PostgreSQL
just db      # Entrar no banco (psql)
just run     # Executar pipeline
just reset   # Limpar e reiniciar banco
just lint    # Verificar código
just format  # Formatar código
just test    # Rodar testes
just check   # Rodar todos (lint, format, test)

Uso

just run                          # Processar mês mais recente
just run --list                   # Listar meses disponíveis
just run --month 2024-11          # Processar mês específico
just run --month 2024-11 --force  # Forçar reprocessamento

Configuração

DATABASE_URL=postgres://postgres:postgres@localhost:5435/cnpj
BATCH_SIZE=500000
TEMP_DIR=./temp
DOWNLOAD_WORKERS=4
RETRY_ATTEMPTS=3
RETRY_DELAY=5
CONNECT_TIMEOUT=30
READ_TIMEOUT=300
STALL_TIMEOUT=30
PROGRESS_LOG_INTERVAL=30
STALL_DEGRADE_THRESHOLD=3  # Stalls acumulados até reduzir a concorrência
KEEP_DOWNLOADED_FILES=false
LOADING_STRATEGY=upsert  # "upsert" ou "replace"
OUTPUT_FORMAT=postgres   # "postgres" ou "parquet"
PARQUET_OUTPUT_DIR=./parquet
PARQUET_TYPED_OUTPUT=false  # Quando true, datas e numéricos saem tipados (Date, Float64, Int32)
PROCESS_WORKERS=1        # Arquivos do mesmo grupo em paralelo (ex: 4)

Downloads resilientes

O servidor da Receita costuma travar streams grandes (e recusa conexões quando várias baixam em paralelo). Os downloads são retomáveis via arquivo .part, e as variáveis acima não têm mais o sentido óbvio dos nomes:

• RETRY_ATTEMPTS conta falhas consecutivas sem progresso, não falhas totais. Qualquer byte novo gravado no .part zera o orçamento, então um arquivo grande sobrevive a muitos stalls produtivos.
• RETRY_DELAY é a base de um backoff exponencial (dobra a cada falha sem progresso, teto de 120s), não um atraso fixo.
• Stalls e connect timeouts repetidos reduzem a concorrência de download (DOWNLOAD_WORKERS -> 2 -> 1) a partir de STALL_DEGRADE_THRESHOLD, sem voltar a subir na mesma execução.
• Cada tentativa HTTP (inclusive as retentativas) passa pela concorrência adaptativa, então ao degradar para 1 as retentativas serializam em vez de reconectar em paralelo.

DATABASE_URL: parâmetros libpq

O DATABASE_URL é repassado direto ao driver, então qualquer parâmetro suportado pelo libpq funciona via query string. Útil para Postgres gerenciado (Railway, RDS, Supabase, Neon) e para isolar o pipeline em um schema próprio.

Sempre coloque o valor entre aspas no shell: o & da query string é metacaractere e, sem aspas, faz o bash colocar o comando em background.

# SSL obrigatório
DATABASE_URL='postgres://user:pass@host:5432/cnpj?sslmode=require'

# Pipeline em schema separado
DATABASE_URL='postgres://user:pass@host:5432/cnpj?options=-c%20search_path%3Dcnpj'

# Combinado
DATABASE_URL='postgres://user:pass@host:5432/cnpj?sslmode=require&options=-c%20search_path%3Dcnpj'

Sobre search_path: o schema precisa existir antes (CREATE SCHEMA cnpj;), o libpq não cria. Se você incluir public como fallback (search_path=cnpj,public) e o public já tiver tabelas do pipeline de uma execução anterior, o bootstrap (ensure_schema) detecta as tabelas no public e não cria nada no cnpj. Para isolamento estrito, deixe só cnpj no search_path, ou use um banco novo.

Estratégia de carga (PostgreSQL)

Estratégia	Comando	Quando usar
upsert	LOADING_STRATEGY=upsert just run	Atualização incremental. Banco continua acessível durante a carga.
replace	LOADING_STRATEGY=replace just run	Carga completa mensal. Mais rápido — faz TRUNCATE e insere direto.

Formato de saída

Formato	Comando	Quando usar
postgres	just run	Default. Carrega no PostgreSQL.
parquet	OUTPUT_FORMAT=parquet just run	Exporta direto para Parquet. Sem banco de dados — ideal para DuckDB, Pandas, Spark.

Parquet

Com OUTPUT_FORMAT=parquet, o pipeline exporta direto para arquivos Parquet com compressão ZSTD. Sem necessidade de PostgreSQL.

OUTPUT_FORMAT=parquet just run

Saída (~6GB a partir de ~85GB de CSVs):

parquet/
  cnaes.parquet
  motivos.parquet
  municipios.parquet
  naturezas_juridicas.parquet
  paises.parquet
  qualificacoes_socios.parquet
  empresas.parquet
  estabelecimentos.parquet
  socios.parquet
  dados_simples.parquet
  manifest.json

Consulte com DuckDB:

SELECT * FROM 'parquet/empresas.parquet' WHERE cnpj_basico = '00000000';
SELECT COUNT(*) FROM 'parquet/estabelecimentos.parquet' WHERE uf = 'SP';

Schema

Documentação completa: docs/data-schema.md

Sobre normalização e tabelas derivadas: docs/post-processing.md · docs/data-audit.md

EMPRESAS (1) ─── (N) ESTABELECIMENTOS
         ├─── (N) SOCIOS
         └─── (1) DADOS_SIMPLES

[!IMPORTANT] Se você já carregou dados antes da mudança para socios.socio_id, recrie socios e as receitas derivadas (socios_quality_flags, socios_clean) para preservar sócios com CPF mascarado ou documento ausente.

Veja o passo a passo em docs/upgrading.md.

Fonte de Dados

Estes dados são públicos e oficiais, disponibilizados pela própria Receita Federal do Brasil.

Fonte	Portal de Dados Abertos — CNPJ
Repositório	Receita Federal — Nextcloud
Atualização	Mensal
Formato	CSV (; separador, ISO-8859-1)
Base legal	Lei 12.527/2011 (Lei de Acesso à Informação), art. 8°
Regulamentação	Decreto 10.046/2019
Nota Técnica	RFB/COCAD n° 47/2024 (alterada pela NT 86/2024)
Metadados	Layout dos arquivos (PDF)

O conjunto de dados CNPJ no Portal de Dados Abertos contém três recursos:

Recurso	Descrição	Status
Dicionário de dados	Layout e metadados dos arquivos	Referência
Inscrições no CNPJ	Dados cadastrais de empresas, estabelecimentos, sócios e Simples Nacional	Processado por este pipeline
Regimes Tributários	Forma de tributação (ECF): Lucro Real, Presumido, Arbitrado, Imunes/Isentas	Ainda não suportado

A Coordenação-Geral de Gestão de Cadastros e Benefícios Fiscais (Cocad) classifica estes dados como dados públicos, de livre acesso a qualquer interessado (NT 47/2024, item 10). CPFs de sócios são mascarados conforme art. 198 da Lei 5.172/1966 (CTN).

Estes dados não são vazados, obtidos ilegalmente, ou protegidos por sigilo fiscal.

Contribuidores