Repository files navigation

Este repositório contém a implementação de uma pipeline ETL utilizando a bibliotecaPandera para validação de contratos de dados em forma de DataFrame. O material foi apresentado em umalive no Youtube no canal doLuciano Galvão Filho.

Os dados são extraídos de um arquivo CSV local, transformados e inseridos em um banco de dados PostgreSQL. Durante o processo, os dados passam por validação em duas etapas: a primeira ocorre ao serem lidos diretamente do arquivo CSV, e a segunda é realizada após a transformação dos dados.

graph TD;    A[Extrai Dados do CSV] -->|Validação de Entrada| B[Calcula Novas Métricas]    B -->|Validação de Saída| C[Carrega Dados no PostgreSQL]

Adicionalmente, foram implementados testes unitários para as validações dos Schemas utilizandopytest, integrados com uma rotina de CI utilizando GitHub Actions para todas aspull requests.

O servidor do banco de dados PostgreSQL, assim como oclient PGAdmin, foram criados localmente utilizando Docker.

• Contexto
• Pandera
• Testes com Pytest
• CI com GitHub Actions
• Instruções para executar este projeto localmente

O arquivodata/dados_financeiros.csv contém o resultado operacional de uma empresa que possui três setores diferentes: vendas, manutenção e reparação. O objetivo é criar uma ETL que leia esses dados, calcule algumas métricas operacionais e carregue os dados em um banco PostgreSQL. Durante o processo, os dados devem seguir a seguinte regra de negócio:

• Dados ao entrar:

• Dados após transformação:

"Data validation for scientists, engineers, and analysts seeking correctness."

Pandera é um projeto open source da Union.ai que oferece uma API flexível e expressiva para realizar validação de dados em objetos semelhantes a dataframes, tornando os pipelines de processamento de dados mais legíveis e robustos.

Integrações:

pip install'pandera[hypotheses]'# verificações de hipótesespip install'pandera[io]'# utilitários de IO para esquemas yaml/scriptpip install'pandera[strategies]'# estratégias de síntese de dadospip install'pandera[mypy]'# habilita a verificação estática de tipos do pandaspip install'pandera[fastapi]'# integração com FastAPIpip install'pandera[dask]'# validação de dataframes Daskpip install'pandera[pyspark]'# validação de dataframes PySparkpip install'pandera[modin]'# validação de dataframes Modinpip install'pandera[modin-ray]'# validação de dataframes Modin com Raypip install'pandera[modin-dask]'# validação de dataframes Modin com Daskpip install'pandera[geopandas]'# validação de geodataframes GeoPandaspip install'pandera[polars]'# validação de dataframes Polars

Abaixo segue oSchema do contrato de dados utilizado no projeto na entrada dos dados.

importpanderaaspaimportpandasaspdfrompandera.typingimportSeriesclassMetricasFinanceirasBase(pa.DataFrameModel):setor_da_empresa:Series[str]receita_operacional:Series[float]=pa.Field(ge=0)data:Series[pa.DateTime]percentual_de_imposto:Series[float]=pa.Field(in_range= {"min_value":0,"max_value":1})custo_operacionais:Series[float]=pa.Field(ge=0)classConfig:strict=Truecoerce=True@pa.check("setor_da_empresa",# <--- Coluna que vai receber a checagem customizadaname="Checagem código dos setores",error="Cógido do setor da empresa é inválido")defcheca_codigo_setor(cls,codigo:Series[str])->Series[bool]:returncodigo.str[:4].isin(['REP_','MNT_','VND_'])

• A classeMetricasFinanceirasBase herda depa.DataFrameModel e define oschema com as colunas/índices sendo atributos de classes.

• pa.Field contém os argumentos nativos da biblioteca e representa características específicas de cada coluna/índice. A lista completa está emCheck.

• Com o decorator@pa.check podemos criar checagens customizadas.

• Emclass Config podemos estabelecer opções de esquema geral.

  • coerce define se a classe, ao validar, vai tentar converter a Coluna para o tipo de dados estabelecido.
  • strict estabelece que a validação não aceita colunas adicionais além das que estão na classe.

• Lista dostipos de dados que o pandera aceita.

Abaixo o modelo criado para oSchema do contrato de dados após sua transformação.

classMetricasFinanceirasOut(MetricasFinanceirasBase):valor_do_imposto:Series[float]=pa.Field(ge=0)custo_total:Series[float]=pa.Field(ge=0)receita_liquida:Series[float]=pa.Field(ge=0)margem_operacional:Series[float]=pa.Field(ge=0)transformado_em:Optional[pa.DateTime]@pa.dataframe_checkdefcheca_margem_operacional(cls,df:pd.DataFrame)->Series[bool]:returndf["margem_operacional"]== (df["receita_liquida"]/df["receita_operacional"])

• Sendo apenas uma extensão dos dados de entrada, a classeMetricasFinanceirasOut herda da classeMetricasFinanceirasBase.

• O decorador@pa.dataframe_check é utilizado para criarmos checagens a nível do dataframe. Esta abordagem difere do decorador@pa.check, que chega a nível de colunas isoladas.

• Com Decoradores:

  • @pa.check_input(<CONTRATO_ENTRADA>): Checa os dados na entrada da função.
  • @pa.check_output(<CONTRATO_SAÍDA>): Checa os dados na saída da função.
  • @pa.check_io(df1 =<CONTRATO_ENTRADA> , df2 = <CONTRATO_ENTRADA>, output = <CONTRATO_SAÍDA>): Checa os dados na entrada e na saida da função.

• Chamando diretamente o método.validate() da classe criada e passando o dataframe a ser validado como argumento.

MetricasFinanceirasBase.validate(df)

O frameworkpytest facilita a escrita de testes pequenos e legíveis, e pode escalar para suportar testes funcionais complexos para aplicações e bibliotecas.

A estrutura utilizada para criarmos testes unitários dos contratos de dados criados com o Pandera foi:

deftest_<NOME_DO_TESTE>():df_test=pd.DataFrame(        {<ESTRUTURA-DO-DF>}    )MetricasFinanceirasBase.validate(df_test)# ouwithpytest.raises(pa.errors.SchemaError):MetricasFinanceirasBase.validate(df_test)# Para quebra do contrato de dados

Para rodar os testes:

pytest test/<ARQUIVO-DO-TESTE> -v

Aflag-v serve para vermos todos os resultados dos testes no terminal.

Para criar uma rotina de Integração Contínua (CI) utilizando GitHub Actions, você deve criar um arquivoci.yaml dentro da pasta.github/workflows.

Para ativar omerge de novaspull requests apenas se passar nos testes, siga os passos abaixo:

1. Vá paraConfigurações.
2. SelecioneBranches no menu lateral esquerdo.
3. Clique emAdicionar regra de proteção de branch.

4. Escolha o nome da regra (qualquer um que você desejar).

5. Marque a caixa paraExigir que verificações de status sejam aprovadas.
6. Marque a caixa paraExigir que os branches estejam atualizados antes da fusão.
7. Clique emAdicionar verificações.
8. Escolha o nome da verificação.
9. Salve as alterações.

Os passos abaixo foram executados para um terminal do tipobash.

1. Clone o repositório localmente:

git clone https://github.com/lealre/pandera-aovivo.git

2. Acesse a pasta:

cd pandera-aovivo

3. Crie um ambiente virtual:

python -m venv .venv

4. Ative o ambiente virtual:

source .venv/bin/activate

5. Baixe as dependências:

pip install -r requirements.txt

6. Crie o banco de dados com o Docker:

docker compose up -d

Para acessar o banco de dados:

7. Entre no link do localhostem seu navegador:http://localhost:8888/

8. Acesse o PGAdmin com as credencias estipuladas no arquivodocker-compose.yaml:

Username:user-name@domain-name.comPassword:pgadmin

9. Defina a senha mestra (ao acessar pela primeira vez).

10. Clique com o botão direito no servidor para conectar o pgAdmin ao banco de dados.

11. Defina o nome do servidor (pode ser qualquer nome que você queira).

12. Conecte-se ao banco de dados usando as credenciais definidas no arquivodocker-compose.yaml.

• Nome do host:db
• Senha:postgres