Grande Tutorial de Expectativas: Validando Dados com Python

Configurando o Great Expectations

Neste tutorial, você aprenderá como usar o GX Core, a versão de código aberto do Great Expectations, para validar um DataFrame do Pandas. Vamos percorrer a configuração de um contexto, registrar uma fonte de dados do Pandas, definir expectativas e validar lotes de dados.

Nota: Recomendamos que você siga com o caderno DataLab, mas você também pode criar seu próprio script em Python.

1. Instalando o Great Expectations

Pré-requisitos

• Python 3.9 a 3.12 instalado.
• Para evitar conflitos, é altamente recomendável que você instale o Great Expectations dentro de um ambiente virtual (aviso: a configuração de ambientes virtuais está além do escopo deste artigo).
• Um conjunto de dados de exemplo.

Nota: Se estiver utilizando o bloco de notas DataLab fornecido, esses pré-requisitos já foram atendidos. Sinta-se à vontade para ignorá-los.

Use o seguinte comando para instalar o GX via pip:

pip install great_expectations

Este comando instala o pacote principal e todas as dependências necessárias.

2. Inicializando o contexto de dados

O Great Expectations requer um contexto de dados para gerenciar configurações. Usamos um contexto de dados efêmero para evitar persistir configurações.

import great_expectations as gx

# Obter o Contexto de Dados Efêmero
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"

Criando sua Primeira Suíte de Validação de Dados

Agora que o GX está configurado, vamos criar uma suíte de validação de dados.

1. Conectando a uma fonte de dados e criando um ativo de dados

Uma fonte de dados conecta o Great Expectations aos seus dados, enquanto um ativo de dados representa um subconjunto específico de dados (por exemplo, uma tabela, DataFrame ou arquivo).

Neste caso, vamos preparar tudo para nos conectarmos a um DataFrame chamado inventory_parts_df. O conjunto de dados de exemplo está disponível no DataLab fornecido e é criado assim que executamos o bloco SQL:

Se você não estiver usando o DataLab, crie seu próprio DataFrame com dados de exemplo.

Agora, crie sua fonte de dados e ativo:

# Adicionar uma Fonte de Dados do Pandas
data_source = context.data_sources.add_pandas(name="inventory_parts")
# Adicionar um Ativo de Dados à Fonte de Dados
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")

2. Adicionando uma definição de lote

Uma definição de lote identifica e organiza seus dados para validação. Aqui, adicionamos uma definição de lote que abrange todo o DataFrame:

# Definir o nome da Definição do Lote
batch_definition_name = "inventory_parts_batch"
# Adicionar a Definição do Lote
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name

3. Recuperando um lote

Um lote é uma coleção de dados vinculada a uma definição de lote. Para validar os dados, você precisará recuperar e vincular o lote ao seu DataFrame, neste caso inventory_parts_df:

# Definir os Parâmetros do Lote
batch_parameters = {"dataframe": inventory_parts_df}
# Recuperar o Lote
batch = batch_definition.get_batch(batch_parameters=batch_parameters)

4. Criando uma suíte e definindo expectativas

Expectativas são regras para validar dados. Neste exemplo, definiremos as seguintes expectativas simples:

1. Assegurar que os valores de inventory_id não sejam nulos.
2. Assegurar que os valores de part_num sejam únicos.

# Criar uma Suite de Expectativas
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
# Adicionar Expectativas
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
# Adicionar a Suite de Expectativas ao Contexto
context.suites.add(suite)

Você pode explorar todas as expectativas disponíveis na Galeria de Expectativas. Nós encorajamos você a adicionar mais algumas!

Após definir as expectativas, o GX gera a configuração da suite de expectativas:

{
  "name": "inventory_parts_suite",
  "id": "b2de0b69-0869-4163-8dde-6c09884483f7",
  "expectations": [
    {
      "type": "expect_column_values_to_not_be_null",
      "kwargs": {
        "column": "inventory_id"
      },
      "meta": {},
      "id": "53d6c42a-d190-412f-a113-783b706531f4"
    },
    {
      "type": "expect_column_values_to_be_unique",
      "kwargs": {
        "column": "part_num"
      },
      "meta": {},
      "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
    }
  ],
  "meta": {
    "great_expectations_version": "1.2.4"
  },
  "notes": null
}

A suite inclui os seguintes detalhes:

1. Nome e ID da Suite: Um nome único (inventory_parts_suite) e identificador para rastrear e gerenciar a suite.
2. Expectativas: Cada regra especifica:
• O tipo de verificação (por exemplo, garantir que uma coluna não tenha valores nulos ou entradas únicas).
• Parâmetros, como a coluna a ser validada.
• Metadados e um ID único para cada expectativa, permitindo um rastreamento e personalização mais fáceis.
3. Metadados: Informações de versão para o Great Expectations, garantindo compatibilidade com a ferramenta.
4. Notas: Um espaço reservado para adicionar comentários descritivos sobre o conjunto (opcional).

Esta saída estruturada atua tanto como documentação quanto como uma configuração reutilizável para validar seu conjunto de dados, de forma que suas expectativas estejam claramente definidas, rastreáveis e prontas para uso futuro.

5. Validando os dados

Por fim, valide o lote de acordo com as expectativas definidas e avalie os resultados.

# Valide os Dados Contra o Conjunto
validation_results = batch.validate(suite)
# Avalie os Resultados
print(validation_results)

Após executar a validação, o Great Expectations fornece um relatório detalhado sobre se o conjunto de dados atende às expectativas definidas:

{
  "success": false,
  "results": [
    {
      "success": true,
      "expectation_config": {
        "type": "expect_column_values_to_not_be_null",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "inventory_id"
        },
        "meta": {},
        "id": "53d6c42a-d190-412f-a113-783b706531f4"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 0,
        "unexpected_percent": 0.0,
        "partial_unexpected_list": [],
        "partial_unexpected_counts": [],
        "partial_unexpected_index_list": []
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    },
    {
      "success": false,
      "expectation_config": {
        "type": "expect_column_values_to_be_unique",
        "kwargs": {
          "batch_id": "inventory_parts-inventory_parts_asset",
          "column": "part_num"
        },
        "meta": {},
        "id": "362a2bdc-616d-4b3a-b7f0-c73808caee78"
      },
      "result": {
        "element_count": 580069,
        "unexpected_count": 568352,
        "unexpected_percent": 97.98006788847535,
        "partial_unexpected_list": [
          "48379c01",
          "paddle",
          "11816pr0005",
          "2343",
          "3003",
          "30176",
          "3020",
          "3022",
          "3023",
          "30357",
          "3039",
          "3062b",
          "3068b",
          "3069b",
          "3069b",
          "33291",
          "33291",
          "3795",
          "3941",
          "3960"
        ],
        "missing_count": 0,
        "missing_percent": 0.0,
        "unexpected_percent_total": 97.98006788847535,
        "unexpected_percent_nonmissing": 97.98006788847535,
        "partial_unexpected_counts": [
          {
            "value": "3069b",
            "count": 2
          },
          {
            "value": "33291",
            "count": 2
          },
          {
            "value": "11816pr0005",
            "count": 1
          },
          {
            "value": "2343",
            "count": 1
          },
          {
            "value": "3003",
            "count": 1
          },
          {
            "value": "30176",
            "count": 1
          },
          {
            "value": "3020",
            "count": 1
          },
          {
            "value": "3022",
            "count": 1
          },
          {
            "value": "3023",
            "count": 1
          },
          {
            "value": "30357",
            "count": 1
          },
          {
            "value": "3039",
            "count": 1
          },
          {
            "value": "3062b",
            "count": 1
          },
          {
            "value": "3068b",
            "count": 1
          },
          {
            "value": "3795",
            "count": 1
          },
          {
            "value": "3941",
            "count": 1
          },
          {
            "value": "3960",
            "count": 1
          },
          {
            "value": "48379c01",
            "count": 1
          },
          {
            "value": "paddle",
            "count": 1
          }
        ],
        "partial_unexpected_index_list": [
          0,
          3,
          4,
          5,
          6,
          7,
          8,
          9,
          10,
          11,
          12,
          13,
          14,
          15,
          16,
          17,
          18,
          19,
          20,
          21
        ]
      },
      "meta": {},
      "exception_info": {
        "raised_exception": false,
        "exception_traceback": null,
        "exception_message": null
      }
    }
  ],
  "suite_name": "inventory_parts_suite",
  "suite_parameters": {},
  "statistics": {
    "evaluated_expectations": 2,
    "successful_expectations": 1,
    "unsuccessful_expectations": 1,
    "success_percent": 50.0
  },
  "meta": {
    "great_expectations_version": "1.2.4",
    "batch_spec": {
      "batch_data": "PandasDataFrame"
    },
    "batch_markers": {
      "ge_load_time": "20241129T122532.416424Z",
      "pandas_data_fingerprint": "84a1e1939091fcf54324910def3b89cd"
    },
    "active_batch_definition": {
      "datasource_name": "inventory_parts",
      "data_connector_name": "fluent",
      "data_asset_name": "inventory_parts_asset",
      "batch_identifiers": {
        "dataframe": "<DATAFRAME>"
      }
    }
  },
  "id": null
}

Este relatório detalha a qualidade dos seus dados, destacando sucessos e falhas. Aqui está uma explicação simplificada dos resultados:

Validação geral: O resultado da validação foi parcialmente bem-sucedido: 50% das expectativas foram atendidas e 50% falharam. Uma expectativa não atendida indica um problema de qualidade dos dados que precisa de atenção. Neste caso, uma coluna não atendeu à regra definida.

Expectativa 1: inventory_id não deve ter valores ausentes

• Resultado: Aprovado
• Explicação: Todos os valores na coluna inventory_id estão presentes, sem entradas nulas ou ausentes. Isso indica boa completude de dados para esta coluna.

Expectativa 2: part_num deve ter valores únicos

• Resultado: Falhou
• Explicação: A coluna part_num contém 97,98% de valores duplicados, o que significa que apenas alguns valores são únicos.
• Destaque:
• Exemplos de valores duplicados incluem “3069b” e “33291”.
• A ferramenta também mostra com que frequência essas duplicatas aparecem e suas posições de linha, facilitando a localização e correção dos problemas.

Claro, este é apenas um conjunto de dados de exemplo e incluímos propositalmente uma expectativa de sucesso e uma de falha para que você possa ver ambos os resultados de validação.

É isso! Você executou com sucesso validações de dados de ponta a ponta.

Integrando o Great Expectations em Pipelines de Dados

Em um ambiente de produção, as validações devem ser incorporadas diretamente no fluxo de trabalho para monitorar continuamente a qualidade dos dados em cada etapa.

Nesta seção, discutiremos como você pode integrar o Great Expectations em seus pipelines de dados.

Estes são exemplos para lhe dar uma ideia e podem ser necessárias configurações adicionais não incluídas aqui. Consulte a documentação de cada ferramenta para obter a sintaxe atualizada!

Integração com ferramentas ETL

A integração do Great Expectations com ferramentas ETL populares como Apache Airflow ou Prefect é relativamente simples. Inserir etapas de validação diretamente nos processos ETL permitirá que você identifique e resolva problemas de dados em tempo real antes que eles afetem a análise subsequente.

Vamos passar por um exemplo simples de integração do Great Expectations com o Prefect para executar a validação de dados como parte de um fluxo de trabalho ETL automatizado:

from prefect import task, Flow
import great_expectations as ge
# Defina uma tarefa para executar a validação do Great Expectations
@task
def validate_data():
    context = ge.data_context.DataContext()
    batch_kwargs = {"path": "path/to/your/datafile.csv", "datasource": "your_datasource"}
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
	    
    # Verifique os resultados da validação e emita um alerta se a validação falhar
    if not results["success"]:
        raise ValueError("Data validation failed!")
# Defina seu fluxo ETL
with Flow("ETL_with_GE_Validation") as flow:
    validation = validate_data()
# Execute o fluxo
flow.run()

Neste exemplo, definimos um fluxo do Prefect com uma tarefa para executar a validação do Great Expectations.

A tarefa validate_data() carrega o contexto do Great Expectations, recupera o lote de dados e aplica o conjunto de expectativas.

Se os dados não atenderem aos critérios de validação, a tarefa gera um alerta, interrompendo o fluxo de trabalho e evitando erros downstream.

Validação contínua de dados

Você pode agendar trabalhos de validação usando várias ferramentas, como cron jobs em sistemas baseados em Unix ou serviços gerenciados como o Apache Airflow. Neste exemplo, demonstraremos como agendar execuções de validação usando o Airflow, que é adequado para orquestrar pipelines de dados.

Aqui está como você pode configurar um DAG do Airflow (Gráfico Acíclico Direcionado) para executar validações do Great Expectations diariamente:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# Defina o DAG e defina o agendamento para executar diariamente
default_args = {
	  'owner': 'airflow',
	  'start_date': datetime(2024, 1, 1),
	  'retries': 1,
}
dag = DAG(
      'great_expectations_validation',
	default_args=default_args,
	schedule_interval='@daily',  # Executa uma vez por dia
)
# Defina a função para executar a validação
def run_validation():
    context = ge.data_context.DataContext()
    batch = context.get_batch(batch_kwargs, suite_name="your_expectation_suite")
    results = context.run_validation_operator("action_list_operator", assets_to_validate=[batch])
    return results
# Configure a tarefa no Airflow
validation_task = PythonOperator(
      task_id='run_great_expectations_validation',
      python_callable=run_validation,
      dag=dag,
)
# Defina a tarefa no DAG
validation_task

Neste exemplo, definimos um DAG que programa a execução de uma validação uma vez por dia (@daily).

A função run_validation() executa a validação carregando o contexto do Great Expectations e executando o conjunto de expectativas definido nos dados.

Melhores Práticas para Validação de Dados com o Great Expectations

Seguir as melhores práticas é sempre aconselhável para escalabilidade e eficiência, e não é diferente para a validação de dados com o Great Expectations.

Comece pequeno e itere

Comece com verificações fundamentais de qualidade de dados e expanda gradualmente. É melhor focar nas expectativas básicas inicialmente, pois isso ajuda a evitar complicar demais o processo, o que torna a integração mais suave e a solução de problemas mais fácil. Conforme sua compreensão do conjunto de dados melhora, você pode adicionar validações mais complexas.

Colabore entre equipes

A qualidade dos dados não é apenas uma preocupação técnica. Colabore entre as equipes de negócios para definir expectativas e garantir que a validação implementada esteja alinhada com a lógica e objetivos de negócios subjacentes. Essa abordagem interfuncional garante que os dados sirvam ao seu propósito pretendido e atendam aos requisitos de todas as partes interessadas.

Automatize quando possível

Automatize o processo sempre que possível para integrar a validação de dados nos fluxos de dados. Integrar verificações automatizadas de validação permite monitorar continuamente a qualidade dos dados sem intervenção manual, o que melhora significativamente a eficiência.

Conclusão

Ótimo trabalho! Você aprendeu como configurar e validar dados no Great Expectations. Essas técnicas ajudarão a manter alta qualidade e transparência nos seus fluxos de trabalho.

Para continuar desenvolvendo suas habilidades, confira estes recursos:

• ETL e ELT em Python: Aprenda a transformar e mover dados de forma eficaz.
• Introdução à Qualidade de Dados: Explore os fundamentos da gestão de qualidade de dados.
• Limpeza de Dados em Python: Domine técnicas de limpeza de dados para garantir precisão e consistência.
• Guia de Consulta das Dimensões de Qualidade de Dados: Um guia prático para dimensões de qualidade de dados.