Tutorial de Grandes Esperanças: Validação de dados com Python

Criando grandes expectativas

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

Observação: Recomendamos que você acompanhe o notebook do DataLab, mas também pode criar seu próprio script Python.

1. Instalação de Great Expectations

Pré-requisitos

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

Observação: Se você estiver usando o notebook DataLab fornecido, esses pré-requisitos já foram atendidos. Você pode ignorá-las.

Use o seguinte comando para instalar o GX via pip:

pip install great_expectations

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

2. Inicialização do contexto de dados

O Great Expectations requer um contexto de dados para gerenciar as configurações. Usamos um contexto de dados efêmero para evitar a persistência das configurações.

import great_expectations as gx

# Get the Ephemeral Data Context
context = gx.get_context()
assert type(context).__name__ == "EphemeralDataContext"

Criando seu primeiro conjunto de validação de dados

Agora que o GX está configurado, vamos criar um conjunto de validação de dados.

1. Conectando-se 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).

Nesse caso, prepararemos tudo para que você se conecte a um DataFrame chamado inventory_parts_df. O conjunto de dados de amostra está disponível no DataLab fornecido e é criado quando executamos o bloco SQL:

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

Agora, crie sua fonte de dados e seu ativo:

# Add a Pandas Data Source
data_source = context.data_sources.add_pandas(name="inventory_parts")
# Add a Data Asset to the Data Source
data_asset = data_source.add_dataframe_asset(name="inventory_parts_asset")

2. Adição de 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:

# Define the Batch Definition name
batch_definition_name = "inventory_parts_batch"
# Add the Batch Definition
batch_definition = data_asset.add_batch_definition_whole_dataframe(batch_definition_name)
assert batch_definition.name == batch_definition_name

3. Recuperação de um lote

Um lote é uma coleção de dados vinculados 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:

# Define the Batch Parameters
batch_parameters = {"dataframe": inventory_parts_df}
# Retrieve the Batch
batch = batch_definition.get_batch(batch_parameters=batch_parameters)

4. Criação de uma suíte e definição de expectativas

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

1. Certifique-se de que os valores de inventory_id não sejam nulos.
2. Certifique-se de que os valores de part_num sejam exclusivos.

# Create an Expectation Suite
expectation_suite_name = "inventory_parts_suite"
suite = gx.ExpectationSuite(name=expectation_suite_name)
# Add Expectations
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="inventory_id")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="part_num")
)
# Add the Expectation Suite to the Context
context.suites.add(suite)

Você pode explorar todas as expectativas disponíveis na Galeria de expectativas. Incentivamos você a acrescentar mais alguns!

Depois de definir as expectativas, o GX gera a configuração do conjunto 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 suíte inclui os seguintes detalhes:

1. Nome e ID da suíte: Um nome exclusivo (inventory_parts_suite) e um identificador para rastrear e gerenciar a suíte.
2. Expectativas: Cada regra especifica:
• O tipo de verificação (por exemplo, garantir que uma coluna não tenha valores nulos ou entradas exclusivas).
• Parâmetros, como a coluna que está sendo validada.
• Metadados e um ID exclusivo para cada expectativa, o que facilita o rastreamento e a personalização.
3. Metadata: Informações sobre a versão de Great Expectations, garantindo a compatibilidade com a ferramenta.
4. Observações: Um espaço reservado para adicionar comentários descritivos sobre a suíte (opcional).

Esse resultado estruturado funciona como documentação e como uma configuração reutilizável para validar seu conjunto de dados, de modo que suas expectativas sejam claramente definidas, rastreáveis e prontas para uso futuro.

5. Validação dos dados

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

# Validate the Data Against the Suite
validation_results = batch.validate(suite)
# Evaluate the Results
print(validation_results)

Depois de 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
}

Esse relatório detalha a qualidade de seus dados, destacando os sucessos e as 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 aprovadas e 50% foram reprovadas. Uma expectativa fracassada indica um problema de qualidade de dados que precisa de atenção. Nesse caso, uma coluna não atendia à regra definida.

Expectativa 1: inventory_id não deve ter valores faltantes

• Resultado: Aprovado
• Explicação: Todos os valores da coluna inventory_id estão presentes, sem entradas nulas ou ausentes. Isso indica uma boa integridade dos dados para essa coluna.

Expectativa 2: part_num deve ter valores exclusivos

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

Obviamente, esse é apenas um conjunto de dados de amostra, e incluímos propositalmente uma expectativa de aprovação e uma de reprovação para que você possa ver os dois resultados de validação.

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

Integração de grandes expectativas aos pipelines de dados

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

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

Esses são exemplos para que você tenha uma ideia, e configurações adicionais não incluídas aqui podem ser necessárias. Consulte a documentação de cada ferramenta para obter a sintaxe atualizada!

Integração com ferramentas de ETL

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

Vamos examinar 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 de ETL automatizado:

from prefect import task, Flow
import great_expectations as ge
# Define a task to run Great Expectations validation
@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])
	    
    # Check validation results and raise an alert if validation fails
    if not results["success"]:
        raise ValueError("Data validation failed!")
# Define your ETL flow
with Flow("ETL_with_GE_Validation") as flow:
    validation = validate_data()
# Execute the flow
flow.run()

Neste exemplo, definimos um fluxo Prefect com uma tarefa para executar a validação de 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 emitirá um alerta, interrompendo o fluxo de trabalho e evitando erros posteriores.

Validação contínua de dados

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

Veja como você pode configurar um DAG(Directed Acyclic Graph) do Airflow para executar diariamente as validações do Great Expectations:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
import great_expectations as ge
# Define the DAG and set the schedule to run daily
default_args = {
	  'owner': 'airflow',
	  'start_date': datetime(2024, 1, 1),
	  'retries': 1,
}
dag = DAG(
      'great_expectations_validation',
	default_args=default_args,
	schedule_interval='@daily',  # Runs once a day
)
# Define the function to run the validation
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
# Set up the task in Airflow
validation_task = PythonOperator(
      task_id='run_great_expectations_validation',
      python_callable=run_validation,
      dag=dag,
)
# Set the task in the DAG
validation_task

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

A função run_validation() executa a validação carregando o contexto de Grandes Expectativas e executando o conjunto de expectativas definido em relação aos dados.

Práticas recomendadas para validação de dados com grandes expectativas

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

Comece pequeno e repita

Comece com verificações básicas da qualidade dos dados e expanda gradualmente. É melhor que você se concentre inicialmente nas expectativas básicas, pois isso ajuda a evitar a complicação excessiva do processo, o que facilita a integração e a solução de problemas. À medida que seu conhecimento do conjunto de dados for melhorando, você poderá adicionar validações mais complexas.

Colaboração entre equipes

A qualidade dos dados não é apenas uma preocupação técnica. Colabore com as equipes comerciais para definir as expectativas e garantir que a validação implementada esteja alinhada com a lógica e as metas comerciais subjacentes. Essa abordagem multifuncional garante que os dados sirvam à finalidade pretendida e atendam aos requisitos de todas as partes interessadas.

Automatize sempre que possível

Automatize o processo sempre que possível para integrar a validação de dados aos pipelines de dados. A integração de verificações de validação automatizadas permite o monitoramento contínuo da qualidade dos dados sem intervenção manual, o que melhora significativamente a eficiência.

Conclusão

Excelente trabalho! Você aprendeu a configurar e validar dados em Great Expectations. Essas técnicas ajudarão a manter a alta qualidade dos dados e a transparência em seus fluxos de trabalho.

Para continuar desenvolvendo suas habilidades, confira estes recursos:

• ETL e ELT em Python: Saiba como transformar e mover dados de forma eficaz.
• Introdução à qualidade dos dados: Explore os fundamentos do gerenciamento da qualidade dos dados.
• Limpeza de dados em Python: Técnicas de limpeza de dados mestres para garantir a precisão e a consistência.
• Folha de dicas sobre as dimensões da qualidade dos dados: Um guia prático sobre as dimensões da qualidade dos dados.