Tratamento dos dados
Classe ReceitaLT
Esta classe fornece ferramentas especializadas para a manipulação e análise de dados da Receita Federal do Brasil.
Métodos:
detect_encoding(file_pattern_or_path, num_bytes=10000)
Retorno: Dicionário com caminhos de arquivos e suas respectivas codificações.
read_data(schema_name, base_path=opcional)
Retorno: DataFrame do Spark consolidado dos arquivos.
Descrição:
- Detecta a codificação dos arquivos.
- Lê múltiplos arquivos CSV.
- Assimila arquivos com delimitador ";", sem cabeçalho e campos delimitados por aspas.
geocode_udf(address)
Retorno: Dicionário com 'latitude' e 'longitude'.
Descrição:
- Utiliza a função interna geocode_address para obter coordenadas.
Retorna um tipo de dado complexo (Struct).
process_estabelecimentos(df)
Retorno: DataFrame processado e enriquecido.
Descrição:
- Lê dataframes adicionais (países, municípios, cnaes, motivos).
- Renomeia colunas.
- Enriquece dados.
- Processa colunas de e-mail e datas.
- Deriva novas colunas.
- Valida endereços de e-mail.
- Geocodifica endereços e CEPs.
- Corrige provedores de e-mail.
process_empresas(df)
Retorno: DataFrame processado e enriquecido.
Descrição:
- Lê dataframes adicionais (naturezas jurídicas e qualificações).
- Renomeia colunas.
- Enriquece dados.
- Processa a coluna 'NOME_EMPRESA' e extrai CPFs.
- Deriva nova coluna com porte da empresa.
- Criptografa CPFs.
process_simples(df)
Retorno: DataFrame processado.
Descrição:
- Converte colunas de data.
- Seleciona apenas colunas relevantes.
save_df(df, path, num_partitions=1, file_format="parquet")
Descrição:
- Salva DataFrame no caminho especificado.
- Especifica o número de partições e o formato do arquivo.
download_and_extract_names(save_base_path=opcional)
Descrição:
- Cria diretório para armazenar o arquivo baixado.
- Realiza o download do arquivo "nomes.csv.gz" do dataset genero-nomes no Brasil.io.
- Extrai o conteúdo do arquivo .gz.
- Exclui o arquivo .gz original, mantendo apenas o arquivo CSV extraído.
process_meis(df, save_base_path=opcional, file_name=opcional)
Retorno: DataFrame processado após todas as transformações e junções.
Descrição:
- Realiza junção com DataFrames de 'naturezas' e 'qualificações'.
- Manipula e extrai dados de CPF.
- Mapeia e cria a coluna "NM_PORTE" para especificar o porte da empresa.
- Criptografa a coluna de CPF.
- Filtra baseado na probabilidade do nome ser um CPF válido.
- Lê um conjunto de dados de nomes e processa a coluna 'alternative_names'.
- Extrai o primeiro nome da coluna 'NOME_EMPRESA' e realiza junção para categorizar o primeiro nome.
process_socios(df, save_base_path=opcional, file_name=opcional)
Retorno: DataFrame processado após todas as transformações e junções.
Descrição:
-Realiza junção com DataFrames de 'países'.
-Mapeia e cria colunas "NM_FAIXA_ETARIA" e "NM_IDENTIFICADOR_SOCIO".
-Renomeia e junta com DataFrame de qualificações.
-Converte coluna de data "DATA_ENTRADA_SOCIEDADE".
-Lê e processa um conjunto de dados de nomes, segmentando e selecionando colunas relevantes.
-Extrai o primeiro nome da coluna 'NOME_SOCIO_RAZAO_SOCIAL' e realiza junção para categorizar esse nome.
geocode_address(address)
Retorno: Um par (tuple) contendo latitude e longitude.
Descrição:
- Geocodifica um endereço, convertendo-o em coordenadas de latitude e longitude.
Notas:
- Utiliza o serviço Nominatim para realizar a geocodificação.
- Tem um limitador de taxa para garantir que não se excedam os limites de requisições por segundo do serviço.
Métodos:
detect_encoding(file_pattern_or_path, num_bytes=10000)
Retorno: Dicionário com caminhos de arquivos e suas respectivas codificações.
read_data(schema_name, base_path=opcional)
Retorno: DataFrame do Spark consolidado dos arquivos.
Descrição:
- Detecta a codificação dos arquivos.
- Lê múltiplos arquivos CSV.
- Assimila arquivos com delimitador ";", sem cabeçalho e campos delimitados por aspas.
geocode_udf(address)
Retorno: Dicionário com 'latitude' e 'longitude'.
Descrição:
- Utiliza a função interna geocode_address para obter coordenadas.
Retorna um tipo de dado complexo (Struct).
process_estabelecimentos(df)
Retorno: DataFrame processado e enriquecido.
Descrição:
- Lê dataframes adicionais (países, municípios, cnaes, motivos).
- Renomeia colunas.
- Enriquece dados.
- Processa colunas de e-mail e datas.
- Deriva novas colunas.
- Valida endereços de e-mail.
- Geocodifica endereços e CEPs.
- Corrige provedores de e-mail.
process_empresas(df)
Retorno: DataFrame processado e enriquecido.
Descrição:
- Lê dataframes adicionais (naturezas jurídicas e qualificações).
- Renomeia colunas.
- Enriquece dados.
- Processa a coluna 'NOME_EMPRESA' e extrai CPFs.
- Deriva nova coluna com porte da empresa.
- Criptografa CPFs.
process_simples(df)
Retorno: DataFrame processado.
Descrição:
- Converte colunas de data.
- Seleciona apenas colunas relevantes.
save_df(df, path, num_partitions=1, file_format="parquet")
Descrição:
- Salva DataFrame no caminho especificado.
- Especifica o número de partições e o formato do arquivo.
download_and_extract_names(save_base_path=opcional)
Descrição:
- Cria diretório para armazenar o arquivo baixado.
- Realiza o download do arquivo "nomes.csv.gz" do dataset genero-nomes no Brasil.io.
- Extrai o conteúdo do arquivo .gz.
- Exclui o arquivo .gz original, mantendo apenas o arquivo CSV extraído.
process_meis(df, save_base_path=opcional, file_name=opcional)
Retorno: DataFrame processado após todas as transformações e junções.
Descrição:
- Realiza junção com DataFrames de 'naturezas' e 'qualificações'.
- Manipula e extrai dados de CPF.
- Mapeia e cria a coluna "NM_PORTE" para especificar o porte da empresa.
- Criptografa a coluna de CPF.
- Filtra baseado na probabilidade do nome ser um CPF válido.
- Lê um conjunto de dados de nomes e processa a coluna 'alternative_names'.
- Extrai o primeiro nome da coluna 'NOME_EMPRESA' e realiza junção para categorizar o primeiro nome.
process_socios(df, save_base_path=opcional, file_name=opcional)
Retorno: DataFrame processado após todas as transformações e junções.
Descrição:
-Realiza junção com DataFrames de 'países'.
-Mapeia e cria colunas "NM_FAIXA_ETARIA" e "NM_IDENTIFICADOR_SOCIO".
-Renomeia e junta com DataFrame de qualificações.
-Converte coluna de data "DATA_ENTRADA_SOCIEDADE".
-Lê e processa um conjunto de dados de nomes, segmentando e selecionando colunas relevantes.
-Extrai o primeiro nome da coluna 'NOME_SOCIO_RAZAO_SOCIAL' e realiza junção para categorizar esse nome.
geocode_address(address)
Retorno: Um par (tuple) contendo latitude e longitude.
Descrição:
- Geocodifica um endereço, convertendo-o em coordenadas de latitude e longitude.
Notas:
- Utiliza o serviço Nominatim para realizar a geocodificação.
- Tem um limitador de taxa para garantir que não se excedam os limites de requisições por segundo do serviço.
Para uma análise mais detalhada, consulte a documentação presente na classe.
Script da Classe
Exemplo do código
Exemplo 1
from ReceitaLT import ReceitaLT
from pyspark.sql import SparkSession
import findspark
spark = (SparkSession.builder
.master("local[*]")
.config('spark.sql.repl.eagerEval.enabled', True)
.config("spark.sql.repl.eagerEval.maxNumRows", 10)
.appName('dataset_cnpj')
.getOrCreate())
receitaLT_processor = ReceitaLT(spark)
df = receitaLT_processor.read_data(schema_name='empresas').cache()
df_mei = receitaLT_processor.process_mei(df)
Output
INFO:ReceitaLT:Detected encodings: {'./output\\Empresas\\Empresas0.csv': 'ascii',
'./output\\Empresas\\Empresas1.csv': 'ascii',
'./output\\Empresas\\Empresas2.csv': 'ascii', './output\\Empresas\\Empresas3.csv': 'ascii',
'./output\\Empresas\\Empresas4.csv': 'ascii', './output\\Empresas\\Empresas5.csv': 'ascii',
'./output\\Empresas\\Empresas6.csv': 'ascii', './output\\Empresas\\Empresas7.csv': 'ascii',
'./output\\Empresas\\Empresas8.csv': 'ascii', './output\\Empresas\\Empresas9.csv': 'ascii'}
INFO:ReceitaLT:Detected encodings: {'./output\\Naturezas\\Naturezas.csv': 'ISO-8859-1'}
INFO:ReceitaLT:Detected encodings: {'./output\\Qualificacoes\\Qualificacoes.csv': 'ISO-8859-1'}
Exemplo 2
from ReceitaLT import ReceitaLT
from pyspark.sql import SparkSession
import findspark
spark = (SparkSession.builder
.master("local[*]")
.config('spark.sql.repl.eagerEval.enabled', True)
.config("spark.sql.repl.eagerEval.maxNumRows", 10)
.appName('dataset_cnpj')
.getOrCreate())
receitaLT_processor = ReceitaLT(spark)
df = receitaLT_processor.read_data(schema_name='estabelecimentos').cache()
df_estabelecimentos = receitaLT_processor.process_estabelecimentos(df)
Output
INFO:ReceitaLT:Detected encodings: {'./output\\Estabelecimentos\\Estabelecimentos.csv': 'ascii', './output\\Estabelecimentos\\Estabelecimentos0.csv': 'ascii',
'./output\\Estabelecimentos\\Estabelecimentos1.csv': 'ascii', './output\\Estabelecimentos\\Estabelecimentos2.csv': 'ascii',
'./output\\Estabelecimentos\\Estabelecimentos3.csv': 'ascii', './output\\Estabelecimentos\\Estabelecimentos4.csv': 'ascii',
'./output\\Estabelecimentos\\Estabelecimentos5.csv': 'ascii', './output\\Estabelecimentos\\Estabelecimentos6.csv': 'ascii',
'./output\\Estabelecimentos\\Estabelecimentos7.csv': 'ascii', './output\\Estabelecimentos\\Estabelecimentos8.csv': 'ascii',
'./output\\Estabelecimentos\\Estabelecimentos9.csv': 'ISO-8859-1'}
INFO:ReceitaLT:Detected encodings: {'./output\\Paises\\Paises.csv': 'ISO-8859-1'}
INFO:ReceitaLT:Detected encodings: {'./output\\Municipios\\Municipios.csv': 'ascii'}
INFO:ReceitaLT:Detected encodings: {'./output\\Cnaes\\Cnaes.csv': 'ISO-8859-1'}
INFO:ReceitaLT:Detected encodings: {'./output\\Motivos\\Motivos.csv': 'ascii'}
continuação da tabela
continuação da tabela
