[chore] Ajustes CNPJ

[laura-l-amaral]

Descrição

Ajustes:

• alterar o nome do conjunto para br_rf_cnpj
• [Bugfix]: br_me_cnpj__dicionario #1520
• [databug] conversão de encoding na tabela de CNPJ #1533
• [chore] ajustar estratégia de particionamento e clusterização da tabela de CNPJ #1508
• [databug] br_me_cnpj dicionario #1401
• Analisar a discrepância de contagem de natureza_juridica: https://discord.com/channels/787841210433536010/1531746530359246958

Obs: manter o conjunto com o nome anterior de pé, até fazermos comunicações para o público

Escopo afetado

• pipelines
• metadados
• modelo dbt

Referências

No response

[luizavboas]

20/07

• Mudanças do PR [Bugfix]: br_me_cnpj__dicionario adaptadas à nova infra com Prefect 3:
• Crawler dicionário
• Mudança de encoding
• Modelos com particionamento criados
• Testar flow de dicionário
• Testar demais flows
• Testar modelos
• Criar conjunto no backend
• Migrar dados de br_me_cnpj para br_rf_cnpj

[luizavboas]

23/07

• Bloqueios com execução do flow https://prefect3.basedosdados.org/runs/flow-run/d2f380bb-9639-4118-ab73-fc8b2f9943df: SOLUCIONADO
• Registrando metadados

[luizavboas]

Levantamento dos dados com caracteres inválidos

Nesta Planilha:

• Data Range: 2023-06-10 a 2026-05-10

• Colunas afetadas:
  br_me_cnpj.estabelecimentos : bairro, complemento, email, logradouro, nome_fantasia, numero, tipo_logradouro
  br_me_cnpj.empresas: razao_social
  br_me_cnpj.socios: nome

• A maioria são casos de cedilha: mais de 93%

• Opções:

1. Coletar os dados novamente
   Gargalo: tabela estabelecimentos.
   Arquivos de CNPJ da Receita Federal são divididos por pastas ano-mes.
   Total de pastas: 39
   Tempo de execução por pasta (download->tratamento->upload GCS): ~55min
   Tempo total: ~40h
2. Replace dos caracteres baseado em tratamentos particulares de cada caso.

[luizavboas]

Dados históricos br_me_cnpj unidos a br_rf_cnpj

Padrão aplicado para modelos incrementais (empresas, sócios, estabelecimentos)
Cada modelo segue a mesma estrutura em CTE:

select ... safe.parse_date('%Y-%m', data_referencia) data_referencia ...
from {{ set_datalake_project("br_rf_cnpj_staging.<tabela>") }}
where <filtro_header>
    {% if is_incremental() %}
        and data_referencia > (select max(data_referencia) from {{ this }})
    {% else %}
        union all
        select ... safe_cast(data as date) data_referencia ..., safe_cast(null as date) data_modificacao
        from {{ set_datalake_project("br_me_cnpj_staging.<tabela>") }}
        where data <= '2023-04-30'
    {% endif %}

Ou seja: no primeiro build (else), a tabela nasce com o histórico completo do br_me_cnpj (até abril/2023) unido ao br_rf_cnpj; nos builds incrementais subsequentes, só se busca o br_rf_cnpj mais recente que o max(data_referencia) já materializado — o histórico não é reprocessado. data_modificacao não existe no br_me_cnpj, então entra como NULL explícito para manter o schema idêntico entre os dois braços do UNION ALL.

[luizavboas]

Script de migração (DEV)

migrate_me_cnpj.py

Tabelas incrementais: foram criadas em sua versão "legado" no br_rf_cnpj_staging:
estabelecimentos_legado, empresas_legado, socios_legado, que serão usadas, com UNION ALL, em conjunto com as suas correspondentes atuais

select ... safe.parse_date('%Y-%m', data_referencia) data_referencia ...
from {{ set_datalake_project("br_rf_cnpj_staging.<tabela>") }}
where <filtro_header>
    {% if is_incremental() %}
        and data_referencia > (select max(data_referencia) from {{ this }})
    {% else %}
        union all
        select ... safe_cast(data as date) data_referencia ..., safe_cast(null as date) data_modificacao
        from {{ set_datalake_project("br_rf_cnpj_staging.<tabela>_legado") }}
        where data <= '2023-04-30'
    {% endif %}

O que falta:

• Executar testes dbt em dev
• Rodar a migração em prod para os dados de 2021 a 2023 e as tabelas de estabelecimentos, socios e empresas