Great Expectations em Python: Guia Completo de Testes de Dados em 2026
Descubra como usar Great Expectations 1.x em Python para validar pipelines de dados em produção. Guia com Fluent API, Expectation Suites, Checkpoints, integração com dbt e Airflow, além de padrões reais para evitar bugs em backfill.
Great Expectations (GX Core) é um framework Python open-source que trata a validação de dados como testes unitários. Você declara o que espera dos seus dados: colunas que não podem ser nulas, IDs únicos, distribuições dentro de faixas aceitáveis, e a biblioteca verifica cada batch antes que ele chegue ao dashboard, ao modelo de ML ou ao relatório financeiro. Neste guia prático de 2026, mostro como instalar, configurar a Fluent API, criar suites de expectativas, orquestrar Checkpoints em produção e integrar com dbt e Airflow para pipelines que não quebram em silêncio às 3h da manhã.
Great Expectations 1.x roda em Python 3.10 a 3.13, com suporte experimental para 3.14 via variável GX_PYTHON_EXPERIMENTAL.
A Fluent API substituiu as antigas classes YAML e é o único caminho recomendado em 2026 (context.data_sources.add_pandas() e amigos).
Um Checkpoint combina uma Batch Definition, uma Expectation Suite e ações (alertas, Data Docs). É a unidade que você agenda no Airflow ou Prefect.
Great Expectations e dbt tests são complementares: rode GX no source data e dbt tests nos modelos transformados.
Data Docs gera HTML estático que substitui planilhas de auditoria e ajuda em conversas com stakeholders não-técnicos.
Os erros mais comuns em backfill vêm de suites que assumem "hoje". Sempre parametrize Batch Definitions com datas.
O que é Great Expectations e por que importa
Great Expectations (GX Core) é uma biblioteca Python que permite escrever asserções declarativas sobre dados, chamadas Expectations, e executá-las contra qualquer fonte: DataFrames pandas, tabelas SQL, arquivos parquet no S3, streams do Spark. A ideia central é simples. Se testes unitários protegem código de regressões, Expectations protegem pipelines de dados daquelas "surpresas silenciosas", tipo o dia em que 30% dos customer_id vieram nulos e ninguém percebeu antes do relatório executivo.
Honestamente, depois de anos consertando pipelines que "funcionavam" mas entregavam dados errados, eu aprendi que a métrica mais importante não é uptime. É confiança. E confiança se constrói com validação explícita em cada etapa: antes de ingerir, depois de transformar, antes de servir. É exatamente aí que o GX se encaixa. A biblioteca vem com mais de 300 Expectations pré-construídas: ExpectColumnValuesToNotBeNull, ExpectColumnValuesToBeUnique, ExpectColumnValuesToBeBetween, ExpectTableRowCountToBeBetween, e por aí vai. Para casos específicos do seu domínio, você escreve custom Expectations em Python puro.
O framework é mantido ativamente e continua Apache 2.0. Os releases recentes de 2026 focaram em remover APIs deprecated (o antigo DeprecatedMetaMetricProvider, argumentos legados de Batch como batch_kwargs), corrigir vazamentos de conexão SQLite em Python 3.13 e adicionar suporte a Pact contracts para CRUD de checkpoints via cliente. Vale consultar as release notes oficiais no GitHub antes de fazer upgrade.
Great Expectations vs dbt tests: qual usar?
Essa é a pergunta que mais recebo. A resposta curta: não escolha, use os dois. Eles resolvem problemas diferentes. dbt tests validam transformação ("meu modelo dim_customer tem PK única?"); Great Expectations valida dados ("essa API de terceiros está mandando o campo country_code em ISO 3166 como sempre?").
Média (requer entender Contexts, Suites, Checkpoints)
Baixa se você já usa dbt
Melhor para
Alertas de qualidade profunda, profiling estatístico
Validação rápida em cada dbt build
Ambiente
Standalone Python, qualquer runtime
Warehouse SQL apenas
Na prática, meu padrão é: dbt tests em toda execução do pipeline, porque são rápidos, integrados e dão feedback imediato para quem faz merge. Great Expectations em fontes externas e em snapshots noturnos: mais lento, mais profundo, roda uma vez por dia. Se você já usa dbt e quer a semântica do GX sem sair do YAML, o pacote dbt-expectations porta boa parte das Expectations para o mundo dbt. Não substitui o GX completo, mas cobre 80% dos casos comuns.
Como instalar Great Expectations em 2026
A instalação é direta via pip. Great Expectations 1.x oficialmente suporta Python 3.10 até 3.13; se você ainda está em 3.9 ou anterior, faça o upgrade da runtime antes, porque não vale a pena carregar backports. Recomendo sempre isolar em um virtualenv ou usar uv/poetry para lock reproduzível.
# com pip padrão
pip install great_expectations
# com uv (recomendado para pipelines de dados em 2026)
uv add great_expectations
# com extras para backends específicos
pip install "great_expectations[snowflake,postgresql,s3]"
Verifique a instalação:
import great_expectations as gx
print(gx.__version__) # deve imprimir 1.x.x
context = gx.get_context(mode="ephemeral")
print(context)
Configurando o Data Context com a Fluent API
O Data Context é o coração do GX. É ele que guarda datasources, suites, checkpoints e configurações. Existem três modos:
ephemeral: em memória, ideal para notebooks e testes;
file: persistido em gx/ no disco, o padrão para desenvolvimento local e CI;
cloud: usa GX Cloud (SaaS pago), útil quando muitos times compartilham suites.
Para pipelines de produção, o modo file versionado no git é o que eu recomendo. Ele torna as suites revisáveis por PR, exatamente como código.
import great_expectations as gx
# cria a estrutura gx/ na raiz do projeto
context = gx.get_context(mode="file", project_root_dir=".")
# adiciona um datasource pandas (Fluent API)
data_source = context.data_sources.add_pandas(name="raw_pandas")
# adiciona um asset (uma "tabela lógica"); nesse caso, um DataFrame
data_asset = data_source.add_dataframe_asset(name="orders")
# define uma batch definition (como você fatia os dados)
batch_definition = data_asset.add_batch_definition_whole_dataframe(
name="all_orders"
)
Se sua fonte é SQL, o padrão muda pouco:
pg_source = context.data_sources.add_postgres(
name="warehouse",
connection_string="postgresql://user:pass@host:5432/db",
)
orders_asset = pg_source.add_table_asset(
name="orders",
table_name="public.orders",
)
# fatia por data (essencial para backfill)
daily_batch = orders_asset.add_batch_definition_daily(
name="orders_daily",
column="created_at",
)
Note o add_batch_definition_daily: essa é a peça que evita dor de cabeça em backfill. Você passa uma coluna de data e o GX consegue selecionar automaticamente a partição do dia que você quer validar, sem hardcode. Voltarei nesse ponto na seção sobre backfill.
Criando sua primeira Expectation Suite
Uma Expectation Suite é o conjunto de regras que descreve o que você espera dos dados. Você pode criar manualmente (recomendado quando você conhece o domínio) ou via data profiling automático (útil para bootstrap em fontes desconhecidas). Vou mostrar a criação manual, que é o que uso 95% do tempo.
import great_expectations as gx
from great_expectations.expectations import (
ExpectColumnValuesToNotBeNull,
ExpectColumnValuesToBeUnique,
ExpectColumnValuesToBeBetween,
ExpectColumnValuesToBeInSet,
ExpectTableRowCountToBeBetween,
)
context = gx.get_context(mode="file")
suite = context.suites.add(
gx.ExpectationSuite(name="orders_source_v1")
)
# regras de integridade
suite.add_expectation(
ExpectColumnValuesToNotBeNull(column="order_id")
)
suite.add_expectation(
ExpectColumnValuesToBeUnique(column="order_id")
)
# regras de dominio
suite.add_expectation(
ExpectColumnValuesToBeInSet(
column="status",
value_set=["pending", "paid", "shipped", "cancelled"],
)
)
suite.add_expectation(
ExpectColumnValuesToBeBetween(
column="total_amount",
min_value=0,
max_value=100000,
)
)
# regra de volume (detecta ingestion pela metade)
suite.add_expectation(
ExpectTableRowCountToBeBetween(
min_value=1000,
max_value=500000,
)
)
suite.save()
Duas dicas que valem ouro em produção. Primeira: sempre versione o nome da suite (orders_source_v1, orders_source_v2). Quando você precisa mudar as regras, cria uma nova suite em vez de sobrescrever a existente. Isso preserva o histórico de Data Docs. Segunda: capriche na ExpectTableRowCountToBeBetween. Na minha experiência, é a expectativa que pega mais bugs de ingestion silenciosos. Se ontem chegaram 200 mil linhas e hoje chegaram 30 mil, alguma coisa está errada mesmo que "todos os campos estejam preenchidos".
Rodando Checkpoints em produção
Suites sozinhas não fazem nada. O que executa a validação é o Checkpoint, uma unidade que junta uma Validation Definition (que aponta uma Batch Definition para uma Suite) e uma lista de actions (o que fazer com o resultado: gerar Data Docs, mandar Slack, escrever num banco, etc).
from great_expectations.checkpoint import (
Checkpoint,
SlackNotificationAction,
UpdateDataDocsAction,
)
# 1. cria uma validation definition
validation_definition = context.validation_definitions.add(
gx.ValidationDefinition(
name="orders_daily_check",
data=daily_batch, # Batch Definition da secao anterior
suite=suite, # Expectation Suite v1
)
)
# 2. cria o checkpoint com as actions
checkpoint = context.checkpoints.add(
Checkpoint(
name="orders_daily_checkpoint",
validation_definitions=[validation_definition],
actions=[
UpdateDataDocsAction(name="update_docs"),
SlackNotificationAction(
name="notify_data_team",
slack_webhook="${SLACK_WEBHOOK}", # via env var
notify_on="failure",
),
],
)
)
# 3. roda o checkpoint (com parametros de batch)
result = checkpoint.run(
batch_parameters={"year": 2026, "month": 7, "day": 6},
)
if not result.success:
raise RuntimeError(f"Data quality check failed: {result.describe()}")
O batch_parameters é onde a mágica acontece para backfill: você pode rodar o mesmo checkpoint para qualquer dia do passado passando as datas apropriadas. Nada de rescrever queries ou duplicar código.
Ingestão: GX valida source data (schema, ranges, volume).
dbt run: transforma os dados no warehouse.
dbt test: valida modelos transformados.
GX pós-transformação: valida datasets analíticos (marts) com regras de negócio profundas.
Um DAG mínimo no Airflow 3:
from airflow.decorators import dag, task
from datetime import datetime
import great_expectations as gx
@dag(
schedule="@daily",
start_date=datetime(2026, 1, 1),
catchup=True, # backfill automatico
)
def daily_orders_pipeline():
@task
def validate_source(ds: str):
context = gx.get_context(mode="file", project_root_dir="/opt/airflow/gx")
checkpoint = context.checkpoints.get("orders_source_checkpoint")
year, month, day = map(int, ds.split("-"))
result = checkpoint.run(
batch_parameters={"year": year, "month": month, "day": day}
)
if not result.success:
raise ValueError(f"Source validation failed for {ds}")
@task
def run_dbt():
import subprocess
subprocess.run(
["dbt", "build", "--select", "tag:orders"],
check=True,
)
@task
def validate_marts(ds: str):
context = gx.get_context(mode="file", project_root_dir="/opt/airflow/gx")
checkpoint = context.checkpoints.get("orders_marts_checkpoint")
year, month, day = map(int, ds.split("-"))
result = checkpoint.run(
batch_parameters={"year": year, "month": month, "day": day}
)
if not result.success:
raise ValueError(f"Marts validation failed for {ds}")
validate_source() >> run_dbt() >> validate_marts()
daily_orders_pipeline()
Para Prefect 3, o padrão é semelhante. Cada passo vira um @task e o fluxo é um @flow. Se você faz o deploy do seu modelo dessa pipeline usando algo como o que descrevi no guia de deploy Scikit-learn com FastAPI, faz sentido rodar um Checkpoint sobre o input do modelo antes de servir predições. Bad input, bad prediction, e o GX pega isso.
Data Docs: documentação automática de qualidade
Uma das razões que eu não trocaria o GX por soluções mais leves é o Data Docs. Cada run de checkpoint gera um HTML estático com: expectativas rodadas, quais passaram, estatísticas descritivas por coluna (min, max, média, distintos) e amostras dos valores que falharam. É documentação viva. Você aponta o Data Docs num bucket S3 com CloudFront ou num Nginx interno, e stakeholders passam a ter visibilidade sem precisar rodar nada.
Depois disso, o UpdateDataDocsAction no seu checkpoint sincroniza automaticamente. Combinado com URLs assinados ou autenticação básica no CloudFront, isso substitui planilhas de auditoria que ninguém atualiza.
Do lado de fluxos analíticos, se você depende de DuckDB para análise de dados em Python em notebooks exploratórios, dá para rodar GX diretamente no DataFrame resultado e publicar o Data Docs. Vira o "cinto de segurança" do trabalho ad-hoc antes de virar dashboard.
Erros comuns em backfill e como evitá-los
Backfill é onde as suites mal desenhadas te machucam. Aqui vão os erros que já cometi (mais de uma vez) e o que aprendi.
1. Hardcode de "hoje" em Batch Definitions
O antipadrão: uma batch que sempre valida WHERE created_at >= CURRENT_DATE. Quando você faz backfill de 90 dias, todas as runs validam o mesmo (vazio) dataset. Sempre parametrize com batch_parameters.
2. Suites com expectativas de volume absoluto
Uma ExpectTableRowCountToBeBetween(min_value=100000, max_value=200000) é ótima para hoje, mas mata backfill de 2 anos atrás quando o negócio era menor. Use expectativas relativas (por exemplo, comparar contra uma janela de dias adjacentes) ou desative essa expectativa específica em backfills antigos.
3. Falhar o pipeline em warnings estatísticos
Nem toda expectativa que "falha" precisa parar a ingestão. Separe regras bloqueantes (schema, nulos em PK, ranges impossíveis) de alertas (skewness, distinct count fora do usual). No Airflow, marque a task de alerta com trigger_rule="all_done" para não bloquear o downstream.
4. Não versionar suites com o código
Se sua suite hoje é orders_source_v3 mas o Airflow deployment antigo ainda referencia orders_source_v1, você acabou de introduzir um bug silencioso. Trate gx/ como código Python: PR, review, CI que roda context.suites.get() para garantir que todas as suites referenciadas existem.
5. Ignorar outliers como se fossem "erro do GX"
Se sua Expectation de range detecta um pico anômalo, resista à tentação de aumentar o range. Investigue primeiro, pode ser fraude, bug de upstream ou dado real que muda o negócio. Se você quer estratégias mais consistentes de análise, dei um panorama no artigo sobre detecção de outliers em Python com Pandas e Scikit-learn.
Para uma referência conceitual completa, os documentos oficiais do Great Expectations são o melhor ponto de partida. A estrutura por conceitos (Contexts, Datasources, Suites, Checkpoints) espelha o modelo mental que você precisa ter para desenhar pipelines duráveis.
Perguntas Frequentes
Great Expectations funciona com Polars?
Sim, indiretamente. O suporte nativo a Polars ainda é experimental em 2026. O caminho suportado é converter para pandas com df.to_pandas() antes de passar para o GX, ou usar um datasource SQL para conjuntos maiores. Se você está migrando pipelines, veja o guia de migração de Pandas para Polars antes de decidir onde colocar a fronteira.
Qual a diferença entre Great Expectations e Pandera?
Pandera é mais leve e focada em validação de schema/tipos dentro do próprio código Python, com sintaxe estilo Pydantic. Great Expectations é uma plataforma completa com Data Docs, checkpoints, integrações e suporte a múltiplos backends. Para validar um DataFrame dentro de uma função, Pandera é mais ágil; para governar qualidade de dados numa plataforma, GX ganha.
Great Expectations é gratuito?
Sim. O GX Core é open-source sob licença Apache 2.0 e é gratuito para sempre. Existe uma oferta comercial paga (GX Cloud) que adiciona colaboração multi-time, dashboards gerenciados e SLA, útil para grandes empresas, mas totalmente opcional.
Como validar dados de APIs com Great Expectations?
Você chama a API, transforma a resposta em DataFrame (pandas) e usa um datasource pandas com um dataframe asset. Isso permite aplicar Expectations no payload antes de ingerir. Para chamadas frequentes, encapsule a lógica num @task do Airflow que faça fetch → to_dataframe → gx checkpoint.
Quantas Expectations devo ter por tabela?
Comece com 5 a 10 regras críticas por tabela: PK não-nula, PK única, colunas obrigatórias não-nulas, ranges de campos numéricos importantes, volume esperado. Adicione mais conforme você aprende com incidentes. Suites com 200 expectativas viram ruído, então priorize o que causa dano real quando quebra.
Como servir modelos scikit-learn em produção com FastAPI: serialização com joblib, endpoint validado por Pydantic, lifespan, Gunicorn + Uvicorn e Docker.
Guia prático para detectar outliers em Python com pandas e scikit-learn: IQR, Z-score modificado, Isolation Forest e LOF, com código pronto para produção.
