Great Expectations 1.20 în Python: Ghid Complet pentru Testarea Calității Datelor în 2026

Actualizat: 16 august 2026

Great Expectations (GX) este o bibliotecă Python open-source pentru testarea calității datelor. Definești așteptări declarative, gen „coloana user_id nu are valori NULL" sau „revenue este între 0 și 1.000.000", apoi le rulezi ca Checkpoints peste DataFrame-uri pandas, tabele SQL sau job-uri PySpark. Primești rapoarte HTML plus alerte Slack când ceva se rupe. În acest ghid actualizat pentru GX Core 1.20.0 (lansat pe 7 august 2026), îți arăt cum funcționează noul model Fluent, cum înlocuiești API-ul YAML v0.x, și de ce trebuie să încetezi să mai folosești CloudDataContext. GX Cloud a fost oprit oficial în mai 2026.

• GX Core 1.20.0 (august 2026) este versiunea stabilă curentă. Rulează pe Python 3.10–3.13 și acceptă acum și Apache Spark 4 (începând cu 1.19.0).
• GX Cloud a fost oprit în v1.17.2 (mai 2026); CloudDataContext aruncă acum excepție. Folosește FileDataContext sau EphemeralDataContext.
• Modelul mental v1.x: DataContext → Data Source → Data Asset → Batch Definition → Expectation Suite → Validation Definition → Checkpoint → Actions. Data Asset descrie formatul, Batch Definition descrie partiționarea.
• API-ul YAML/CLI vechi și SimpleCheckpoint/RuntimeBatchRequest nu mai există. Orice tutorial din 2023–2024 este obsolet.
• Pentru validare rapidă în cod aplicație (FastAPI, servicii ML) preferă Pandera. Pentru gate-uri de pipeline SQL/Spark cu rapoarte partajabile, Great Expectations rămâne standardul.
• Checkpoints sunt single-threaded. Nu profila tabele Snowflake de miliarde de rânduri; validează pe batch-uri partiționate zilnic.

Ce este Great Expectations în Python?

Great Expectations este un framework open-source pentru testarea calității datelor, echivalentul pytest pentru pipeline-uri. Definești o Expectation Suite care conține reguli („expectations") despre cum ar trebui să arate datele: unicitate pe cheie primară, interval numeric, patternuri regex, distribuție statistică, prospețimea unei coloane timestamp. La runtime, un Checkpoint rulează suita pe un Batch de date și produce un obiect ExpectationSuiteValidationResult plus un raport HTML (Data Docs) pe care oricine din echipă îl poate deschide fără să știe Python.

Am ajuns la GX după ani în care am scris teste ad-hoc cu assert în cadrul job-urilor Airflow. Problema cu abordarea aia era că alerta ajungea la 3 dimineața ca stacktrace crud în log, iar analista de business care raporta metricii n-avea idee de ce dashboard-ul era gol. Cu GX, alerta ajunge în Slack cu link către un raport care spune „85% din rânduri au trecut, 15% au picat expectația expect_column_values_to_not_be_null pe coloana user_id". Diferența în timpul de rezoluție a incidentelor este uriașă.

Filosofia principală: datele sunt cod, iar codul are nevoie de teste. Dacă ai deja un stack modern cu dbt, GX nu îl înlocuiește. Completează stratul de testare cross-model și oferă rapoarte accesibile stakeholderilor non-tehnici. dbt-ul tău testează transformările; GX testează sursele și ieșirile.

Ce s-a schimbat în GX 1.x și de ce a dispărut GX Cloud

Dacă ai deschis vreun tutorial de Great Expectations scris înainte de mijlocul lui 2024, aruncă-l. GX 1.0 (august 2024) a fost o ruptură completă. API-ul YAML plus great_expectations init CLI au fost eliminate și înlocuite de Fluent API declarativ în Python. Clase pe care le vedeai peste tot în articole vechi (SimpleCheckpoint, RuntimeBatchRequest, batching_regex, blocuri datasources: în YAML) nu mai există.

Al doilea lucru care surprinde oamenii care revin la GX în 2026: GX Cloud a fost oprit. Pull request-ul #11894, mergeat în v1.17.2 pe 14 mai 2026, face constructorul CloudDataContext să arunce direct excepție. Site-ul greatexpectations.io încă are pagini de marketing care descriu ExpectAI și GX Cloud. Ignoră-le. Pentru producție, folosește FileDataContext (persistă configurația pe disc/S3) sau EphemeralDataContext (totul în memorie, potrivit pentru joburi Airflow one-shot și pentru CI).

Alte eliminări din v1.17.0 (mai 2026): argumentele legacy data_context, datasource_name, batch_parameters, batch_kwargs pe metodele Batch. Config-urile evaluation_parameter_store_name, notebooks și include_rendered_content nu mai sunt acceptate. Parametrii de evaluare runtime se scriu acum ca expectation_parameters. Consultă direct releases-urile oficiale de pe GitHub pentru changelog complet; nu te baza pe articole terțe.

Cum instalezi și configurezi Great Expectations 1.20

Instalarea este simplă în principiu (un pip install), dar dependința grea (SQLAlchemy, Marshmallow, Jinja2, pyarrow) intră frecvent în conflict cu pin-urile din proiecte existente. Regula mea după ani de bătălii: întotdeauna într-un venv proaspăt și pin la o versiune exactă. Fără >=, fără ^.

python -m venv .venv
source .venv/bin/activate
pip install "great-expectations==1.20.0"

# Backend-uri opționale, instalează doar ce folosești
pip install "great-expectations[snowflake]==1.20.0"
pip install "great-expectations[bigquery]==1.20.0"
pip install "great-expectations[spark]==1.20.0"    # Spark 4 acceptat din 1.19.0
pip install "great-expectations[postgresql]==1.20.0"

Verifică instalarea și inițializează un context persistent pe disc:

import great_expectations as gx

print(gx.__version__)  # ar trebui să afișeze 1.20.0

# FileDataContext creează structura de foldere gx/ în directorul curent
context = gx.get_context(mode="file", project_root_dir="./gx_project")
print(context.root_directory)

Pentru joburi CI sau pentru task-uri Airflow care n-au nevoie să persiste configurația, folosește EphemeralDataContext. Nu scrie nimic pe disc, deci nu ai probleme cu permisiuni de writer în containere read-only.

context = gx.get_context(mode="ephemeral")

Conceptele cheie ale Fluent API: DataContext, Batch Definition, Expectation Suite

Modelul mental al Fluent API este singurul lucru care încurcă echipele care migrează de la v0.x. Iată maparea explicită, în ordinea în care le construiești:

1. DataContext, obiectul rădăcină. Gestionează configurația, store-urile pentru rezultate, definițiile de checkpoints.
2. Data Source, adică unde stau datele: „pandas", „postgres", „snowflake", „spark".
3. Data Asset, adică ce fel de date: un tabel, un fișier CSV, un dataframe. Descrie formatul, nu partiționarea.
4. Batch Definition, adică cum partiționezi asset-ul: „zilnic după coloana event_date", „lunar", „întregul asset". Aici e diferența nouă față de v0.x, care confunda cele două concepte în batching_regex.
5. Batch Request, cererea concretă pentru un batch specific (ex: „ziua 2026-08-15").
6. Expectation Suite, colecția de reguli.
7. Validation Definition, leagă o Suite de o Batch Definition (perechea „ce validezi + pe ce").
8. Checkpoint, rulează una sau mai multe Validation Definitions și declanșează Actions.

Data Asset = format. Batch Definition = partiționare. Dacă ții minte doar asta, ai depășit cel mai mare hop conceptual al migrării la v1.x.

Primul pipeline complet: validare pandas pas cu pas

Uite un exemplu end-to-end complet, funcțional pe 1.20.0. Validăm un DataFrame pandas cu tranzacții: cheia primară trebuie să fie unică, amount pozitiv, iar currency să fie una din trei valori acceptate. La finalul secțiunii vei avea un raport HTML rulabil local.

import pandas as pd
import great_expectations as gx
from great_expectations import expectations as gxe

# 1. Context și date de test
context = gx.get_context(mode="file", project_root_dir="./gx_project")

df = pd.DataFrame({
    "transaction_id": [1, 2, 3, 4, 5],
    "amount":         [10.5, 22.0, 5.75, 99.9, 3.25],
    "currency":       ["EUR", "USD", "RON", "EUR", "USD"],
    "user_id":        [101, 102, 103, 104, 105],
})

# 2. Data Source + Data Asset + Batch Definition
data_source = context.data_sources.add_pandas(name="transactions_src")
data_asset  = data_source.add_dataframe_asset(name="transactions_asset")
batch_def   = data_asset.add_batch_definition_whole_dataframe(
    name="whole_df"
)

# 3. Expectation Suite
suite = context.suites.add(
    gx.ExpectationSuite(name="transactions_suite")
)
suite.add_expectation(
    gxe.ExpectColumnValuesToBeUnique(column="transaction_id")
)
suite.add_expectation(
    gxe.ExpectColumnValuesToBeBetween(column="amount", min_value=0.01, max_value=10000)
)
suite.add_expectation(
    gxe.ExpectColumnValuesToBeInSet(
        column="currency",
        value_set=["EUR", "USD", "RON"],
    )
)

# 4. Validation Definition + Checkpoint
validation_def = context.validation_definitions.add(
    gx.ValidationDefinition(
        name="transactions_validation",
        data=batch_def,
        suite=suite,
    )
)

checkpoint = context.checkpoints.add(
    gx.Checkpoint(
        name="transactions_checkpoint",
        validation_definitions=[validation_def],
        actions=[gx.checkpoint.UpdateDataDocsAction(name="update_docs")],
        result_format={"result_format": "SUMMARY"},
    )
)

# 5. Rulare
result = checkpoint.run(batch_parameters={"dataframe": df})
print("success:", result.success)
context.open_data_docs()  # deschide raportul HTML în browser

Ce se întâmplă în spate: GX construiește un plan de validare, execută fiecare expectație pe batch (aici tot DataFrame-ul), acumulează statistici de linie și emite rezultatul. Acțiunea UpdateDataDocsAction regenerează raportul HTML în ./gx_project/gx/uncommitted/data_docs/local_site/index.html. Deschide-l. Este singurul artefact pe care îl vor înțelege colegii tăi non-Python.

Checkpoints, Actions și alerte Slack

Un Checkpoint fără alertare este un test care nu se execută. Great Expectations livrează out-of-the-box SlackNotificationAction, MicrosoftTeamsNotificationAction și EmailAction, iar tu poți subclasa ValidationAction pentru orice altceva (PagerDuty, Opsgenie, Grafana OnCall). Detalii oficiale sunt în documentația de Checkpoint Actions.

from great_expectations.checkpoint import (
    SlackNotificationAction,
    UpdateDataDocsAction,
)

slack_action = SlackNotificationAction(
    name="slack_alert",
    slack_webhook="https://hooks.slack.com/services/XXX/YYY/ZZZ",
    notify_on="failure",           # doar când pică
    notify_with=["local_site"],    # link către Data Docs
    show_failed_expectations=True, # listează ce s-a rupt
)

checkpoint = context.checkpoints.add_or_update(
    gx.Checkpoint(
        name="transactions_checkpoint",
        validation_definitions=[validation_def],
        actions=[
            UpdateDataDocsAction(name="update_docs"),
            slack_action,
        ],
    )
)

Recomandarea mea, din experiență: separă alertele critice de cele informaționale. Am văzut echipe care primesc 40 de Slack-uri pe zi de la GX și le ignoră pe toate. Slack fatigue-ul este real. Folosește un Checkpoint cu notify_on="failure" pentru expectații critice (unicitate PK, prospețime) și unul separat cu notify_on="all" care postează într-un canal #data-quality-daily pentru vizibilitate low-signal.

Backend-uri: Pandas, SQL, Snowflake, BigQuery și PySpark

Puterea reală a GX vine din faptul că același Expectation Suite rulează pe pandas, SQL sau Spark fără modificări. Suita este declarativă; backend-ul o traduce în operațiuni native. Pentru un tabel Snowflake, GX generează SELECT COUNT(*) FROM ... WHERE ... IS NULL; pentru Spark, generează un plan Catalyst; pentru pandas, iterează. Câteva exemple concrete:

# Postgres / MySQL / Redshift / Snowflake / BigQuery, toate SQLAlchemy
pg_source = context.data_sources.add_postgres(
    name="warehouse",
    connection_string="postgresql://user:pass@host:5432/analytics",
)
orders_asset = pg_source.add_table_asset(name="orders", table_name="orders")

# Batch Definition partiționat zilnic
daily_batch = orders_asset.add_batch_definition_daily(
    name="daily", column="created_at"
)
# Cere batch-ul de ieri
batch = daily_batch.get_batch(
    batch_parameters={"year": 2026, "month": 8, "day": 15}
)

Pentru Snowflake specific, folosește add_snowflake(), care pasează parametrii de conectare corect (warehouse, role, database). Pentru BigQuery, add_bigquery() acceptă credentiale ADC sau service account JSON.

# PySpark (Spark 4 acceptat de la GX 1.19.0)
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

spark_source = context.data_sources.add_spark(name="spark_src")
events_asset = spark_source.add_dataframe_asset(name="events")
whole = events_asset.add_batch_definition_whole_dataframe(name="whole")

sdf = spark.read.parquet("s3://bucket/events/date=2026-08-15/")
batch = whole.get_batch(batch_parameters={"dataframe": sdf})

Dacă analizezi date cu SQL în Python, articolul nostru despre DuckDB pentru analiza datelor cu SQL este un companion natural. Poți valida rezultatele DuckDB cu GX folosind conector SQLAlchemy standard, iar Data Docs devin dovada pe care o atașezi la audit-uri.

Great Expectations vs Pandera vs Soda Core

Nu există „cel mai bun tool" universal. Alegerea depinde de unde rulezi validarea și cine citește rezultatele. Iată tabelul pe care îl folosesc în consultanță:

Criteriu	Great Expectations 1.20	Pandera 0.25	Soda Core 3.5
Definire reguli	Python declarativ (ExpectColumn...)	Schema Pydantic-style	YAML (SodaCL)
Backend-uri	pandas, SQL, Spark 4	pandas, Polars, PySpark, Dask	SQL warehouses, Spark, Dask
Rapoarte HTML partajabile	Da (Data Docs)	Nu (doar excepții Python)	Doar via Soda Cloud (comercial)
Overhead startup	Mare (context + config)	Foarte mic	Mic
Cea mai bună potrivire	Pipeline-uri batch, audit	Validare in-process, FastAPI, cod ML	Echipe cu analiști care scriu YAML
Curbă învățare	Abruptă (multe concepte)	Ușoară dacă știi Pydantic	Medie (limbaj propriu)
Alertare integrată	Slack, Teams, Email out-of-box	Doar excepții	Slack, Webhooks

Recomandarea practică: dacă construiești un serviciu ML sau un API care primește date de la clienți și vrei să respingi payload-uri invalide, folosește Pandera. Se integrează perfect cu FastAPI. Vezi ghidul nostru despre servirea modelelor ML cu FastAPI pentru un exemplu concret. Dacă ai un warehouse Snowflake/BigQuery și trebuie să produci rapoarte HTML pentru echipa de compliance, alege Great Expectations. Dacă echipa ta e mixtă (data analysts + engineers) și vrei ca oamenii să scrie teste în YAML fără să deschidă Python, Soda Core.

Integrare cu Airflow, Prefect și Dagster

În lumea reală, GX rulează într-un orchestrator. Toți cei trei mari au integrări oficiale sau semi-oficiale mature în 2026:

• Airflow: pachetul airflow-provider-great-expectations (release 0.4.0 pe 28 ianuarie 2026) oferă GreatExpectationsOperator compatibil cu GX 1.x. Pasezi numele checkpoint-ului sau path-ul contextului și DAG-ul se oprește pe fail.
• Prefect: prefect-great-expectations expune task-uri run_checkpoint_validation care mapează rezultatele GX în stări Prefect (Failed/Completed). Pentru Prefect 3.x, folosește versiunea 2.0+ a integrării.
• Dagster: Dagster tratează GX ca asset checks native. Definești un checkpoint ca @asset_check peste un asset materializat și Dagit afișează statusul lângă asset. Cel mai integrat setup dintre cei trei.

Un pattern de care mă țin ferm: nu bloca DAG-ul principal pe validări soft. Pipeline-ul de ingestie continuă chiar dacă validarea „prospețime {'<'} 6h" pică; pornește doar o alertă. Blochează doar pe validări hard (schema break, nulls în PK). Backfill-urile mari sunt suficient de dureroase și fără să se oprească la ora 4 pentru un warning cosmetic.

Great Expectations în GitHub Actions pentru CI/CD

Testarea calității în CI, înainte ca datele să ajungă în producție, este pattern-ul cel mai underexploatat din research-ul nostru. Puține articole îl arată în cod real. Uite un workflow minimal care validează un CSV de test și pică PR-ul dacă expectațiile nu trec:

# .github/workflows/data-quality.yml
name: Data Quality Gate
on: [pull_request]

jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install "great-expectations==1.20.0" pandas
      - name: Run GX checkpoint
        env:
          SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
        run: python ci/run_checkpoint.py

Scriptul ci/run_checkpoint.py creează un EphemeralDataContext, încarcă seed-ul, rulează checkpoint-ul și iese cu cod non-zero dacă result.success este False. GitHub Actions marchează PR-ul roșu, review-ul e blocat. Am salvat mai multe backfill-uri decât pot număra cu pattern-ul ăsta simplu. Schema-uri stricate prinse în PR, nu în producție la ora 3.

# ci/run_checkpoint.py
import sys, pandas as pd, great_expectations as gx
from great_expectations import expectations as gxe

ctx = gx.get_context(mode="ephemeral")
df = pd.read_csv("tests/data/orders_seed.csv")

src = ctx.data_sources.add_pandas("ci")
asset = src.add_dataframe_asset("orders")
batch_def = asset.add_batch_definition_whole_dataframe("all")

suite = ctx.suites.add(gx.ExpectationSuite("orders_ci"))
suite.add_expectation(gxe.ExpectColumnValuesToNotBeNull(column="order_id"))
suite.add_expectation(gxe.ExpectColumnValuesToBeUnique(column="order_id"))

vd = ctx.validation_definitions.add(
    gx.ValidationDefinition(name="v", data=batch_def, suite=suite)
)
cp = ctx.checkpoints.add(gx.Checkpoint(name="ci_cp", validation_definitions=[vd]))
result = cp.run(batch_parameters={"dataframe": df})

sys.exit(0 if result.success else 1)

Capcane comune și optimizări de performanță

Am strâns lista asta din issue-uri reale de pe tracker-ul GX de pe GitHub și din bătălii proprii:

• Checkpoints sunt single-threaded. Rularea action-urilor și serializarea rezultatelor se face în Python, nu delegată motorului. Pe seturi mari (GitHub #10231), devine bottleneck. Soluția: partiționează asset-ul în batch-uri zilnice și rulează checkpoint-uri paralele în Airflow/Prefect.
• UserConfigurableProfiler face OOM pe Snowflake mare (#5389). Nu profila tabele de miliarde de rânduri. Sampling către un CTE de 100k rânduri, apoi profilează.
• Format „COMPLETE" umflă memoria. Setează result_format="SUMMARY" sau "BASIC" pentru producție. „COMPLETE" e util doar în debugging local.
• Conflicte de dependințe. Pin la o versiune fixă a pyarrow și pandas la nivel de proiect. GX 1.20 vrea pyarrow <18. Verifică metadata PyPI înainte să faci upgrade.
• Data Docs pe S3. Store-ul default e local; pentru echipe distribuite configurează un site_url_prefix către un bucket S3 static hosting. Fără asta, doar tu vezi rapoartele.
• Testul pe „preț mediu", celebrul ExpectColumnMeanToBeBetween. Sună util dar e fragil: distribuția se schimbă natural. Preferă ExpectColumnValueLengthsToBeBetween sau constraint-uri pe cvantile în loc de medii.

Pentru pipeline-uri de preprocesare pandas care alimentează suitele GX, ghidul nostru Curățarea și Preprocesarea Datelor cu Pandas tratează exact ce ar trebui să valideze GX după: transformări idempotente, handling explicit al valorilor lipsă, coerciție de tipuri. Le combini și ai un pipeline hardened.

Întrebări frecvente

Este Great Expectations încă întreținut activ în 2026?

Da. GX Core 1.20.0 a fost lansat pe 7 august 2026, cu cadență lunară pe throughout 2026 (1.18 în iunie, 1.19 în iulie, 1.20 în august). Serviciul comercial GX Cloud a fost oprit în mai 2026, dar biblioteca open-source rămâne activă și primește feature-uri noi, Spark 4 support fiind cel mai recent exemplu.

Ce a înlocuit API-ul YAML din GX 0.x?

Fluent API declarativ în Python, introdus în GX 1.0 (august 2024). În loc să configurezi datasources în great_expectations.yml, apelezi context.data_sources.add_postgres(...). CLI-ul great_expectations init a fost eliminat; totul se face acum programatic. SimpleCheckpoint, RuntimeBatchRequest și batching_regex nu mai există.

Funcționează Great Expectations cu PySpark, Snowflake și BigQuery?

Da, cu toate trei nativ. Instalează extras-ul corespunzător: pip install "great-expectations[spark]", [snowflake] sau [bigquery]. Începând cu GX 1.19.0 (iulie 2026), este acceptat și Apache Spark 4. Aceeași Expectation Suite rulează pe oricare backend fără modificări. GX o traduce în operațiuni native (SQL sau Spark).

Great Expectations vs Pandera, care e mai bun?

Depinde de context. Pandera este ideal pentru validare in-process în servicii Python (FastAPI, cod ML): sintaxă Pydantic-style, overhead mic. Great Expectations este mai potrivit pentru pipeline-uri batch peste warehouse-uri, unde ai nevoie de rapoarte HTML partajabile (Data Docs) și alertare Slack/Teams out-of-box. Multe echipe le folosesc pe amândouă: Pandera în cod aplicație, GX în orchestrator.

Cum trimit alerte Slack când un Checkpoint eșuează?

Adaugă un SlackNotificationAction în lista de actions a Checkpoint-ului, cu slack_webhook="https://hooks.slack.com/..." și notify_on="failure". Setează show_failed_expectations=True ca mesajul Slack să listeze exact ce expectații au picat, nu doar un „ceva a eșuat". Citește webhook-ul dintr-o variabilă de mediu sau dintr-un secret manager, nu îl hardcoda.