Great Expectations ile Python Veri Kalitesi Testleri: Pandas ve Airflow Entegrasyonu (2026)

Güncellendi: 3 Ağustos 2026

Great Expectations (GX), Python veri boru hatlarında satır sayısından şema kısıtlarına, benzersizlikten dağılım kaymalarına kadar bir tabloya dair her "beklentiyi" test edilebilir kurallara dönüştürmenize yarayan açık kaynaklı bir veri kalitesi çerçevesidir. Bu rehberde, 2024'te yayımlanan Great Expectations 1.0 ve sonrasındaki 1.x sürümleriyle gelen sadeleştirilmiş Fluent API'yi baz alarak; pandas ve Airflow ile veri kalitesini nasıl otomatik test edeceğinizi, Checkpoint'leri nasıl kuracağınızı ve production'da hangi tuzaklara düşmemeniz gerektiğini adım adım göstereceğim.

• Great Expectations 1.x, eski great_expectations.yml tabanlı bloklu yapıyı büyük ölçüde bıraktı; artık Fluent API üzerinden Python'da Data Source → Batch Definition → Expectation Suite → Validation Definition → Checkpoint zincirini kuruyorsunuz.
• Pandas DataFrame'lerini test etmek için tek satırlık bir ephemeral Data Context yeterli; test amaçlı Airflow DAG'lerinde bu yapı yerel kalır ve state biriktirmez.
• Airflow ile entegrasyon için airflow-provider-great-expectations paketinin GXValidateCheckpointOperator'ünü kullanmak, DAG'i "Checkpoint kırıldıysa görev başarısız olsun" mantığına indirger.
• Data Docs, ekibin veri kalitesi metriklerini birlikte görebildiği tek arayüzdür. S3 veya GCS'e senkronize edip Slack'e link atmak, tek başına önemli bir alışkanlık değişikliği yaratır.
• Backfill'lerde Batch Parameters ile ay/gün bazlı Batch'ler tanımlamazsanız, her koşuda tüm tabloyu doğrulamaya çalışıp saatlerinizi kaybedersiniz; bu en sık atlanan detaydır.

Great Expectations nedir ve neden kullanılır?

Great Expectations, veri boru hattınızda akan tabloların yapısal ve istatistiksel özelliklerini "beklenti" (expectation) adı verilen deklaratif kurallara çevirip her koşuda otomatik olarak doğrulayan bir Python kütüphanesidir. Kısacası: SQL'de CHECK kısıtlarının, pandas'ta assert'lerin ve dbt'de tests: bloklarının yaptığı işi, kaynak sistemden data warehouse'a kadar tek bir soyutlama altında topluyor. Ben data engineer olarak yıllardır fark ediyorum ki, pipeline'ları asıl uykumuzdan eden şey şema değişikliği değil, sessiz dağılım kaymaları: bir gün "amount" alanı ortalama 12₺ iken ertesi gün 1.200₺ olur, ETL "başarılı" görünür ve dashboard'lar yalan söyler.

GX işte tam olarak bu tür sinsi bozulmaları yakalar. Bir tablonun satır sayısının belirli bir aralıkta olması, bir kolonun her zaman non-null olması, country_code'un yalnızca ISO 3166 listesindeki değerleri içermesi, updated_at'in son 24 saatte olması gibi kuralları tek bir Expectation Suite'te toplarsınız. Suite, tabloyla değil iş kuralıyla ilişkilidir; bu yüzden aynı Suite Snowflake'teki üretim tablosuna da, geliştirme ortamındaki pandas kopyasına da bağlanabilir. Benzer bir yaklaşımı DataFrame şeması düzeyinde tercih ediyorsanız, daha önce yazdığım Pandera ile pandas DataFrame doğrulama rehberi pandera'nın statik tip desteğiyle nasıl tamamlayıcı bir rol üstlendiğini anlatıyor.

GX'i "başka bir test kütüphanesi" olarak görmek yanıltıcı olur; kütüphane aynı zamanda bir observability katmanı sunar. Her koşuda üretilen Validation Result JSON'ları ve otomatik HTML rapor sayfaları (Data Docs), sadece "geçti/kaldı" sinyali vermez; hangi kuralın hangi satırlarda kaç kez ihlâl edildiğini de gösterir. Bu, incident sonrasında blame-storming yerine "veri kanıtına" bakmayı mümkün kılar.

Kurulum ve 1.x sürüm notları

2024 Ağustos'unda çıkan Great Expectations 1.0, 0.x döneminin en büyük şikâyeti olan "çok fazla yaml, çok az Python" sorununu ciddi biçimde çözdü. Artık her şey Python'da; great_expectations init zorunlu bir başlangıç adımı değil, opsiyonel bir kolaylık. Ayrıntılar için Great Expectations 1.x resmi dokümantasyonunu takip edebilirsiniz. Kurulum en yalın haliyle şöyle:

# Sanal ortam açın; ben uv'yi tercih ediyorum ama pip de olur.
python -m venv .venv && source .venv/bin/activate

# 1.x sürümünü sabitleyerek yükleyin; PyPI'de "great_expectations".
pip install "great_expectations>=1.3,<2.0" pandas

# Airflow entegrasyonu için opsiyonel provider.
pip install "airflow-provider-great-expectations>=0.4.0"

Great Expectations 1.x'te temel akış üç Python nesnesine indirgenmiştir: Data Source (nereden okuyorsunuz), Batch Definition (hangi dilimi test ediyorsunuz) ve Expectation Suite (hangi kurallar geçerli). Bu üçlüyü bir Validation Definition altında birleştirir, bir Checkpoint ile zamanlanabilir hale getirirsiniz. Kavramsal olarak bu, dbt'deki sources.yml + schema.yml + dbt test üçlüsünün Python'daki karşılığıdır.

Pandas ile ilk Expectation Suite'i oluşturmak

Pandas kullanıcıları için en pratik başlangıç, ephemeral Data Context'tir: diske hiçbir dosya yazmaz, tek Python oturumu için bellekte yaşar ve testlerde harikadır. Aşağıda küçük bir sipariş tablosunu test eden minimal bir örnek var; kod parçacıklarını olduğu gibi kopyalayıp Jupyter'de veya bir .py dosyasında çalıştırabilirsiniz.

import pandas as pd
import great_expectations as gx
from great_expectations.core.expectation_suite import ExpectationSuite
from great_expectations import expectations as gxe

# 1) Ephemeral Data Context, diske hiçbir şey yazmıyor.
context = gx.get_context(mode="ephemeral")

# 2) Örnek veri; gerçek hayatta bunu S3, Snowflake ya da BigQuery'den çekersiniz.
df = pd.DataFrame({
    "order_id":   [1001, 1002, 1003, 1004, 1005],
    "customer":   ["ali", "veli", "ayse", "mehmet", None],
    "amount_try": [120.5, 55.0, 890.0, 33.9, 210.0],
    "country":    ["TR", "TR", "DE", "TR", "TR"],
})

# 3) Pandas Data Source ve DataFrame Asset kaydı.
data_source   = context.data_sources.add_pandas(name="orders_ds")
data_asset    = data_source.add_dataframe_asset(name="orders")
batch_def     = data_asset.add_batch_definition_whole_dataframe("orders_batch")

# 4) Suite ve Expectation'ları tanımla.
suite = context.suites.add(ExpectationSuite(name="orders_suite"))
suite.add_expectation(gxe.ExpectColumnValuesToNotBeNull(column="order_id"))
suite.add_expectation(gxe.ExpectColumnValuesToBeUnique(column="order_id"))
suite.add_expectation(gxe.ExpectColumnValuesToBeInSet(
    column="country", value_set=["TR", "DE", "FR", "NL"]
))
suite.add_expectation(gxe.ExpectColumnValuesToBeBetween(
    column="amount_try", min_value=0, max_value=100_000
))

# 5) Validation Definition ve tek seferlik run.
validation_def = context.validation_definitions.add(
    gx.ValidationDefinition(
        name="orders_validation",
        data=batch_def,
        suite=suite,
    )
)
result = validation_def.run(batch_parameters={"dataframe": df})
print(result.success, "->", result.statistics)

Bu betiği çalıştırdığınızda, customer alanındaki None değeri hakkında bir kural tanımlamamış olsak da, order_id'nin non-null ve unique olması, country'nin izin verilen kümede kalması ve amount_try'nın makul bir aralıkta olması test edilir. GX 1.x'te en sık kullanacağınız Expectation türleri şunlardır: ExpectColumnValuesToNotBeNull, ExpectColumnValuesToBeUnique, ExpectColumnValuesToBeInSet, ExpectColumnValuesToBeBetween, ExpectTableRowCountToBeBetween, ExpectColumnMeanToBeBetween. Bu kural setiyle, veri temizleme aşamasında yakaladığınız problemleri gerileme (regression) olarak sabitleyebilirsiniz — konuyla ilgili daha derin bir örneği Pandas 3.0 ile veri temizleme rehberi içinde bulabilirsiniz.

Checkpoint kurulumu ve çalıştırma

Tek seferlik validation_def.run() geliştirme sırasında iyidir; ama üretimde asıl işi Checkpoint yapar. Checkpoint, birden fazla Validation Definition'ı sıraya dizer, sonuçları merkezi Store'lara yazar, action'lar (Slack bildirimi, Data Docs güncellemesi, hata fırlatma) tetikler. Kavramsal olarak: pytest'in test_*.py keşif+koşma+raporlama davranışının veri versiyonudur.

from great_expectations.checkpoint import (
    Checkpoint, SlackNotificationAction, UpdateDataDocsAction,
)

checkpoint = context.checkpoints.add(
    Checkpoint(
        name="orders_daily_checkpoint",
        validation_definitions=[validation_def],
        actions=[
            UpdateDataDocsAction(name="update_docs"),
            SlackNotificationAction(
                name="slack_alert",
                slack_webhook="${SLACK_WEBHOOK_URL}",  # ortam değişkeninden
                notify_on="failure",
                show_failed_expectations=True,
            ),
        ],
        result_format={"result_format": "COMPLETE"},
    )
)

result = checkpoint.run(batch_parameters={"dataframe": df})
if not result.success:
    raise ValueError("orders_daily_checkpoint failed, see Data Docs")

Ekibim için altın kural şudur: Bir Checkpoint bir Airflow task'ına, bir task da bir iş kuralına eşit olmalıdır. "Bütün tabloları tek Checkpoint'te doğrula" cazip gelir ama başarısızlık durumunda ne kırıldığını bulmak için loglara dalmanız gerekir; Airflow'un görev düzeyinde retry, alert ve SLA mantığından da faydalanamazsınız.

Airflow ile Great Expectations entegrasyonu

Great Expectations'ı üretime almanın en yaygın yolu, Apache Airflow içinde bir DAG olarak koşturmaktır. 2024'ten itibaren topluluk, airflow-provider-great-expectations paketini 1.x uyumlu hale getirdi ve pipeline yazımını ciddi biçimde kısalttı. Aşağıda, günlük olarak S3'ten Snowflake'e sipariş verisi taşıyan ve her yükten önce veri kalitesi kapısı (quality gate) çalıştıran bir DAG örneği var.

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from great_expectations_provider.operators.validate_checkpoint import (
    GXValidateCheckpointOperator,
)

def build_checkpoint(context, batch_parameters):
    # Sadece Checkpoint objesini döndürüyoruz; operator kendisi çalıştırır.
    import great_expectations as gx
    from great_expectations.checkpoint import Checkpoint
    validation = context.validation_definitions.get("orders_validation")
    return Checkpoint(
        name="orders_daily_checkpoint",
        validation_definitions=[validation],
    )

default_args = {
    "owner": "data-eng",
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="orders_daily_quality_gate",
    schedule="0 6 * * *",  # her sabah 06:00 UTC
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args=default_args,
    tags=["quality", "gx"],
) as dag:

    validate = GXValidateCheckpointOperator(
        task_id="validate_orders",
        configure_checkpoint=build_checkpoint,
        # Batch parametresi ile ilgili günün dilimi test edilir.
        batch_parameters={"year": "{{ ds_nodash[:4] }}",
                          "month": "{{ ds_nodash[4:6] }}",
                          "day": "{{ ds_nodash[6:8] }}"},
    )

    load_to_warehouse = PythonOperator(
        task_id="load_to_snowflake",
        python_callable=lambda **_: None,  # gerçekte SnowflakeOperator vb.
    )

    validate >> load_to_warehouse

Yukarıdaki kalıp iki nedenle güçlüdür: birincisi, validate task'ı düştüğünde load_to_snowflake hiç çalışmaz; kötü veri warehouse'a girmez. İkincisi, Airflow'un native alerting'i (email, Slack, PagerDuty) devrededir; ayrıca ek entegrasyona ihtiyacınız yoktur. Ben genellikle bunu bir on_failure_callback ile birleştirip ihlâl eden satırların bir örneğini Slack thread'ine post ediyorum; oncall için ilk 3 dakikayı kazandırıyor.

Özel (custom) Expectation nasıl yazılır?

Yerleşik Expectation'lar %80 duruma yeter, ama iş kuralları bazen çok tuhaflaşır: "kargo ücreti free_shipping_over_amount'un üzerindeyse 0 olmalı", "kupon kodu regex'i ile ürün kategorisi tutarlı olmalı" gibi. GX 1.x, custom Expectation yazımını 0.x'e göre önemli ölçüde sadeleştirdi; artık bir Python sınıfı yeterli.

from great_expectations.expectations.expectation import (
    ColumnMapExpectation,
)
from great_expectations.expectations.metrics import (
    ColumnMapMetricProvider, column_condition_partial,
)

class ColumnValuesMatchTurkishPhone(ColumnMapMetricProvider):
    condition_metric_name = "column_values.match_turkish_phone"
    condition_value_keys = ()

    @column_condition_partial(engine="pandas")
    def _pandas(cls, column, **kwargs):
        # 5XX XXX XX XX kalıbı; boşluk/tire toleransı için sadeleştirilmiş regex.
        return column.astype(str).str.replace(r"[\s-]", "", regex=True) \
                     .str.match(r"^5\d{9}$")

class ExpectColumnValuesToMatchTurkishPhone(ColumnMapExpectation):
    # Turkiye cep telefonu formatina (5XXXXXXXXX) uygunlugu test eder.
    map_metric = "column_values.match_turkish_phone"
    success_keys = ("mostly",)
    args_keys = ("column",)

Bu sınıfları plugins/expectations/ altına koyup projeye register ettikten sonra, tıpkı yerleşiklerdeki gibi suite.add_expectation(ExpectColumnValuesToMatchTurkishPhone(column="msisdn", mostly=0.98)) diyerek kullanabilirsiniz. mostly parametresi, "en az %98 uyumluysa geçir" mantığıdır ve gerçek dünyada çok iş görür (çünkü tarihsel veriler nadiren %100 temizdir).

Data Docs'u ekiple paylaşmak

Data Docs, Great Expectations'ın en hafife alınan özelliği. Her Checkpoint sonrası otomatik olarak güncellenen statik HTML sayfaları üretir; Suite'lerdeki her Expectation'ın açıklamasını, son çalışma sonuçlarını, kaç kere geçip kaç kere kaldığını tek arayüzde toplar. Bu, "veri sözleşmesi" (data contract) tartışmalarını dokümantasyon PR'ları yerine bir link paylaşımına indirger.

from great_expectations.data_context.data_context.file_data_context import (
    FileDataContext,
)

context = gx.get_context(project_root_dir="./gx", mode="file")

# S3'e statik olarak host etmek icin site config.
context.add_data_docs_site(
    site_name="team_docs",
    site_config={
        "class_name": "SiteBuilder",
        "store_backend": {
            "class_name": "TupleS3StoreBackend",
            "bucket": "my-data-docs",
            "prefix": "gx-docs/",
        },
        "site_index_builder": {"class_name": "DefaultSiteIndexBuilder"},
    },
)
context.build_data_docs()

Ekibimde build_data_docs'ı DAG'in son adımı olarak koşuyoruz; başarılı ya da başarısız her koşuda güncellenir. Slack'e günlük bir "kalite özeti" postu şart değil — insanlar zamanla scroll'lar; kritik sinyal yine PagerDuty'den gelmeli. Data Docs, incident sonrası retrospektif ve yeni takım arkadaşlarını yetiştirirken devreye giriyor.

Üretimde dikkat edilecekler ve tipik tuzaklar

Aşağıdaki maddeler, GX'i büyük ölçekli veri boru hatlarına taşırken kendi ekibimin bedel ödeyerek öğrendiği çıkarımlar. Her biri, bir 3 sabaha kalkma hikayesinin karşılığıdır.

Batch Definition olmadan üretime çıkmayın

Yeni başlayanların en sık hatası, "her koşuda tüm tabloyu test edelim" refleksidir. 3 milyar satırlık bir fact tablosunda bu, Snowflake credit'lerini eritir. Bunun yerine tabloyu year, month, day gibi partition kolonlarına göre bölüp add_batch_definition_yearly/monthly/daily API'lerini kullanın; Airflow'daki {{ ds }} ile sadece o günün dilimini test edin. Backfill'lerde de aynı Batch tanımı çalışır; sadece parametreleri değişir.

Suite'leri kodla versionlayın

Suite'lerin JSON serialize edilebilir olması güzel bir özellik, ancak "yaml dosyasında editleyelim" tuzağına düşmeyin. Suite'leri Python fabrika fonksiyonlarında oluşturup git'te tutun; PR review, kod inceleme kültürünüzün doğal parçası olur. dbt schema.yml'lerini nasıl review ediyorsanız GX Suite'lerini de öyle review edin.

Şüpheli veriye "quarantine" yaklaşımı

Checkpoint kırılınca DAG'i durdurmak her zaman doğru refleks değildir. Kritik bir günlük raporun boşluğu bırakması, "%3 hatalı satırla devam etmek"ten daha maliyetli olabilir. Ben tipik olarak iki katmanlı bir strateji uyguluyorum: error-level ihlâller (unique key duplike, null primary key) pipeline'ı durdurur; warning-level ihlâller (dağılım kayması, format uyumsuzluğu) devam eder ama satırları bir quarantine tablosuna kopyalar ve Slack'e link atar. Bu, backfill'lerde büyük fark yaratır.

Feature engineering pipeline'larınızı da doğrulayın

Great Expectations'ı sadece "ham veri" katmanına uygulamak yaygın bir hata. ML boru hatlarındaki feature tabloları da en az kaynak veriler kadar kalite testine ihtiyaç duyar; hem eğitim zamanı hem inference-time skew'i yakalamak için birebir aynı Expectation Suite'i her iki tarafta koşturun. Konuya derinlemesine girmek isterseniz Scikit-learn Pipeline rehberim feature transformation adımlarına GX kapıları eklemek için doğal bir başlangıç noktası verir.

GX Cloud'a geçmeden önce iyi düşünün

2024'ten beri GX Cloud yönetilen bir çözüm olarak sunuluyor. Doğru kullanım senaryosu var: küçük ekipler, hızlı prototipleme, business kullanıcılara Suite yazdırma. Ama Suite'lerinizi Python'da versionlamak, Airflow ile entegre etmek istiyorsanız, açık kaynak sürüm zaten fazlasını sunar. Bu kararı vendor lock-in ve maliyet perspektifinden değerlendirin.

Sıkça Sorulan Sorular

Great Expectations ile Pandera arasındaki fark nedir?

Pandera, DataFrame şemasını Python sınıfları/tipleriyle tanımlar ve genellikle fonksiyon parametresi düzeyinde çalışır; statik tip kontrolü ve pytest ile mükemmel uyuşur. Great Expectations ise tablo/veri seti düzeyinde bir observability katmanıdır: Data Docs, Checkpoint'ler ve orchestration entegrasyonu ile üretim boru hatlarına odaklanır. İkisi rakip değil, tamamlayıcıdır — pandera fonksiyon içi doğrulama, GX ise cross-run tablo doğrulama için idealdir.

Great Expectations 1.0 ile 0.18 arasında geçiş zor mu?

Evet, breaking change'ler var. Datasource/DataConnector kavramları Data Source + Batch Definition'a evrildi ve yaml tabanlı proje yapısı büyük ölçüde Python'a taşındı. Küçük projelerde yeniden yazmak, migration script'iyle uğraşmaktan hızlıdır. Büyük projeler için resmi migration rehberini takip edin ve production'a geçmeden staging'de en az bir tam backfill koşturun.

Great Expectations Spark ile çalışır mı?

Evet. context.data_sources.add_spark(...) ile PySpark DataFrame'lerini aynı Suite API'siyle test edebilirsiniz; motor değişse de Expectation kodu genellikle taşınabilir kalır. Databricks ortamında ephemeral Data Context'i notebook başında yaratıp Checkpoint'i job task'ı içinden koşturmak sık kullanılan bir kalıptır.

Checkpoint her koşuda tüm tabloyu okuyor, nasıl hızlandırırım?

Bu neredeyse her zaman eksik Batch Definition sorunudur. Tabloyu add_batch_definition_daily/monthly/yearly ile bölüp Airflow'un {{ ds }} değişkeniyle sadece ilgili dilimi doğrulayın. Ayrıca result_format'u BASIC'te tutarak ihlâl eden satırların tamamının JSON'a yazılmasını engelleyin. Bu ikisi birlikte tipik olarak 10x hız kazandırır.

Great Expectations dbt tests'in yerini alır mı?

Tam olarak değil. dbt tests, transform edilmiş modeller üzerinde SQL-native ve dbt'in DAG'i içinde çalışır; buna karşın GX, raw ingestion, ML feature'ları ve dbt dışı boru hatlarında da devreye girer. Pratikte ekipler ikisini beraber kullanır: dbt unique/not_null gibi hızlı kontroller için, GX ise dağılım kayması, referans veri uyumu ve custom iş kuralları için idealdir.