Processando o dump do CNPJ com Polars: do ZIP ao Parquet

O cadastro de CNPJ da Receita Federal chega como dezenas de milhões de linhas em CSV, sem cabeçalho e em latin-1. Pandas até dá conta, mas exige muita memória e paciência. O Polars, com execução preguiçosa (lazy) e motor de streaming, é uma escolha melhor: lê os arquivos em pedaços, aplica transformações em paralelo e grava Parquet sem precisar carregar tudo de uma vez.

Este tutorial mostra um pipeline completo, do ZIP baixado ao Parquet pronto para análise, com as decisões que evitam os problemas mais comuns.

Por que Polars

Há várias ferramentas capazes de processar esse volume. O Polars se encaixa bem por alguns motivos práticos:

• Roda numa máquina só. Para dezenas de milhões de linhas, um cluster distribuído adiciona complexidade sem necessidade.
• Execução preguiçosa. Você descreve a transformação inteira e o otimizador decide a ordem, empurra filtros para a leitura e descarta colunas que não serão usadas.
• Streaming. O motor processa os dados em lotes, então o tamanho do arquivo não precisa caber na memória.
• API expressiva e tipada. As transformações ficam legíveis, testáveis e fáceis de revisar em pull request.

Se o seu time já domina SQL, o DuckDB é uma alternativa equivalente para boa parte deste pipeline. As decisões de modelagem abaixo valem para os dois.

Visão geral do pipeline

ZIPs do mês ──► CSV UTF-8 ──► scan_csv (lazy, tudo texto) ──► tipagem e limpeza ──► Parquet por tabela
                                                                                     │
                                                         consultas e junções ◄───────┘

Três princípios guiam as decisões:

1. Ler tudo como texto e tipar explicitamente. Inferência automática erra com CNPJ (zeros à esquerda), CEP e datas zeradas.
2. Nunca materializar o arquivo inteiro sem necessidade. Trabalhe com LazyFrame até o último passo.
3. Guardar o resultado por mês de referência. Cada carga é uma fotografia; você vai querer compará-las depois.

Passo 1: baixar e organizar por mês

Baixe os ZIPs da pasta do mês no servidor de dados abertos da Receita e guarde-os numa estrutura versionada:

dados/
  2026-08/
    zip/   Empresas0.zip … Empresas9.zip, Estabelecimentos0.zip …, Socios0.zip …, Simples.zip, Cnaes.zip …
    utf8/  (gerado no passo 2)
    parquet/ (gerado no passo 4)

Manter os ZIPs originais permite reprocessar qualquer mês se você mudar a lógica de limpeza.

Passo 2: converter de latin-1 para UTF-8

O leitor de CSV do Polars trabalha com UTF-8. Ler latin-1 diretamente estraga todos os acentos (SÃO PAULO vira lixo) ou falha. A forma mais robusta e rápida é converter em streaming, sem descompactar para o disco antes:

cd dados/2026-08
mkdir -p utf8
for f in zip/*.zip; do
  nome=$(basename "$f" .zip)
  unzip -p "$f" | iconv -f LATIN1 -t UTF-8 > "utf8/${nome}.csv"
done

O unzip -p joga o conteúdo na saída padrão e o iconv converte em fluxo. A memória usada é mínima, e o processo pode ser paralelizado por arquivo com xargs -P se você tiver núcleos sobrando.

Passo 3: ler de forma preguiçosa, tudo como texto

O arquivo não tem cabeçalho, então você informa os nomes das colunas na ordem do dicionário de dados. Para Estabelecimentos, são 30:

import polars as pl

ESTAB_COLS = [
    "cnpj_basico", "cnpj_ordem", "cnpj_dv", "matriz_filial", "nome_fantasia",
    "situacao", "data_situacao", "motivo_situacao", "cidade_exterior", "pais",
    "data_inicio", "cnae_principal", "cnaes_secundarios", "tipo_logradouro",
    "logradouro", "numero", "complemento", "bairro", "cep", "uf", "municipio",
    "ddd1", "telefone1", "ddd2", "telefone2", "ddd_fax", "fax", "email",
    "situacao_especial", "data_situacao_especial",
]

def scan_raw(pattern: str, columns: list[str]) -> pl.LazyFrame:
    return pl.scan_csv(
        pattern,
        separator=";",
        has_header=False,
        new_columns=columns,
        infer_schema=False,  # tudo como texto: a tipagem é nossa
        quote_char='"',
    ).with_columns(pl.all().str.strip_chars().replace("", None))  # "" vira null

Dois detalhes importam aqui:

• infer_schema=False mantém tudo como texto. Assim, "00012345" continua com os zeros e nenhuma coluna é inferida como número por acaso.
• A última linha transforma strings vazias em nulos. Sem isso, filtros como is_null() não pegam campos vazios, e médias e contagens ficam erradas.

O scan_csv aceita padrões de arquivo, então utf8/Estabelecimentos*.csv lê as dez partes como uma tabela só.

Passo 4: tipar, limpar e gravar Parquet

Agora aplicamos as regras de negócio que valem para qualquer uso:

def to_date(col: str) -> pl.Expr:
    # "0", "00000000" e datas impossíveis viram null em vez de quebrar o pipeline
    return pl.col(col).str.to_date("%Y%m%d", strict=False)

estab = (
    scan_raw("utf8/Estabelecimentos*.csv", ESTAB_COLS)
    .with_columns(
        cnpj=pl.concat_str("cnpj_basico", "cnpj_ordem", "cnpj_dv"),
        matriz=pl.col("matriz_filial") == "1",
        situacao=pl.col("situacao").cast(pl.Int8),
        data_inicio=to_date("data_inicio"),
        data_situacao=to_date("data_situacao"),
        cnaes_secundarios=pl.col("cnaes_secundarios").str.split(","),
    )
    .drop("matriz_filial", "cidade_exterior", "ddd_fax", "fax")
)

estab.sink_parquet("parquet/estabelecimentos.parquet")

O sink_parquet executa o plano em streaming: lê, transforma e grava em lotes, sem carregar o arquivo inteiro na memória.

Repare nas escolhas:

• CNPJ como texto, montado a partir das três partes. Isso já deixa o pipeline pronto para o CNPJ alfanumérico, que tem letras na raiz e na ordem.
• CNAEs secundários como lista. A coluna original é uma string separada por vírgulas; transformá-la em lista permite filtrar com list.contains e explodir quando precisar.
• Situação como inteiro pequeno. 2 é ativa, 8 é baixada; ocupa pouco e compara rápido.

A tabela Empresas tem um cuidado extra: o capital social usa vírgula decimal.

EMP_COLS = [
    "cnpj_basico", "razao_social", "natureza_juridica",
    "qualificacao_responsavel", "capital_social", "porte", "ente_federativo",
]

empresas = scan_raw("utf8/Empresas*.csv", EMP_COLS).with_columns(
    capital_social=pl.col("capital_social")
    .str.replace(",", ".", literal=True)
    .cast(pl.Decimal(18, 2)),
)
empresas.sink_parquet("parquet/empresas.parquet")

Use Decimal em vez de Float64 para valores monetários: você evita surpresas de arredondamento ao somar capitais.

E as outras tabelas?

O mesmo padrão vale para Sócios, Simples e as tabelas de domínio. Cada uma tem as suas particularidades:

• Sócios: o documento do sócio pessoa física vem mascarado (***123456**) e o de pessoa jurídica vem completo. Guarde o tipo de sócio como inteiro e não tente "desmascarar" o CPF: além de tecnicamente impossível com segurança, seria um problema de LGPD.
• Simples: as colunas de opção vêm como S ou N. Converta para booleano e tipe as datas de opção e exclusão com a mesma função to_date.
• Domínios (CNAEs, municípios, naturezas, qualificações): são pequenas. Leia com read_csv, guarde em Parquet e use em junções para trazer as descrições só na hora de exibir.

Uma regra que evita retrabalho: guarde códigos nas tabelas grandes e descrições nas tabelas de domínio. Os códigos ocupam menos espaço, comparam mais rápido e não mudam quando a Receita ajusta a grafia de uma descrição.

Passo 5: consultar em streaming

Com o Parquet pronto, as consultas ficam rápidas e baratas. Um exemplo: estabelecimentos e empresas ativas por UF e CNAE principal, com o capital social mediano.

ativas_por_cnae = (
    pl.scan_parquet("parquet/estabelecimentos.parquet")
    .filter(pl.col("situacao") == 2)
    .join(pl.scan_parquet("parquet/empresas.parquet"), on="cnpj_basico", how="left")
    .group_by("uf", "cnae_principal")
    .agg(
        estabelecimentos=pl.len(),
        empresas=pl.col("cnpj_basico").n_unique(),
        capital_mediano=pl.col("capital_social").median(),
    )
    .sort("estabelecimentos", descending=True)
    .collect(engine="streaming")
)

O Polars empurra o filtro de situação e a seleção de colunas para a leitura do Parquet (predicate e projection pushdown). Ele só lê do disco as colunas e os grupos de linhas de que precisa.

Para trabalhar com as atividades secundárias, explodimos a lista:

secundarios = (
    pl.scan_parquet("parquet/estabelecimentos.parquet")
    .select("cnpj", "cnaes_secundarios")
    .explode("cnaes_secundarios")
    .drop_nulls("cnaes_secundarios")
    .rename({"cnaes_secundarios": "cnae"})
)

O resultado é uma tabela de relação CNPJ × CNAE, útil para buscas do tipo "empresas que têm a atividade X como principal ou secundária".

Passo 6: validar antes de publicar

Pipelines de dados quebram em silêncio. Algumas checagens baratas pegam a maioria dos problemas:

df = pl.scan_parquet("parquet/estabelecimentos.parquet")

checagens = df.select(
    linhas=pl.len(),
    cnpj_duplicado=pl.len() - pl.col("cnpj").n_unique(),
    cnpj_tamanho_errado=(pl.col("cnpj").str.len_chars() != 14).sum(),
    sem_uf=pl.col("uf").is_null().sum(),
    data_inicio_nula=pl.col("data_inicio").is_null().sum(),
).collect()
print(checagens)

Compare esses números com os do mês anterior. Uma variação muito grande no total de linhas, CNPJs duplicados ou uma explosão de datas nulas indicam arquivo incompleto, mudança de layout ou erro de conversão. Nesses casos, pare a publicação antes que o problema chegue aos usuários.

Dicas de desempenho

• Disco rápido importa mais que CPU na etapa de conversão. Se possível, use SSD local.
• Compressão do Parquet: o padrão (zstd) equilibra bem tamanho e velocidade. Mantenha.
• Colunas que você não usa podem ser descartadas cedo, como fizemos com fax e cidade no exterior.
• Memória limitada: prefira sink_parquet e collect(engine="streaming"). Evite read_csv e collect() sem streaming em tabelas inteiras.
• Reprocessamento: como os ZIPs originais ficam guardados, você pode recriar o Parquet de qualquer mês ao mudar uma regra.

Conclusão

Com Polars, o dump do CNPJ cabe num notebook comum: converta para UTF-8 em fluxo, leia tudo como texto, tipe explicitamente, grave Parquet por mês e consulte em streaming. A partir daí, você tem uma base pronta para contagens de mercado, segmentação e, guardando os meses, para enxergar mudanças ao longo do tempo.

Se você quer os dados já tratados e pesquisáveis sem manter o pipeline, experimente a busca da CNPJpedia.