Veri kalitesi: eksik ve saçma satırları yakala

Garbage In, Garbage Out

Veri mühendisliğinde tekrar eden en eski kural. Pipeline’a giren veri kirli ise, downstream’deki her şey — dashboard, ML modeli, alert — kirli olur. Düzeltmek saatler, bazen günler sürer; bazen kimse fark etmeden aylarca yanlış rapor yayınlanır.

Bu yüzden veri kalitesi shift-left yapılır: pipeline’ın başında, kaynağa yakın yerde yakalanır. Bugün iki mekanizma öğreneceğiz:

1. dlt schema contracts — veri gelirken reddetme.
2. Post-load checks — veri tabloda iken alert kurma.

Schema Contracts (Veriyi Reddet)

dlt’in columns parametresiyle her sütun için katı kurallar tanımlarsın. İhlal eden kayıt pipeline’ı durdurur veya atlanır (senin tercihin).

Geçen dersin pipeline’ını güncelle:

@dlt.resource(
    table_name="books",
    write_disposition="replace",
    schema_contract={
        "tables": "evolve",   # yeni tablolara izin ver
        "columns": "freeze",  # mevcut sütunların tipi sabit
        "data_type": "freeze",
    },
    columns={
        "title": {"data_type": "text", "nullable": False},
        "first_publish_year": {"data_type": "bigint", "nullable": True},
        "edition_count": {"data_type": "bigint", "nullable": True, "data_type": "bigint"},
    },
)
def harry_potter_books():
    """Sadece başlığı olan kitapları kabul et."""
    for page in client.paginate(
        "/search.json",
        params={"q": "harry potter"},
        paginator=PageNumberPaginator(base_page=1, total_path=None),
    ):
        docs = page.get("docs", [])
        # Pre-filter: başlığı olmayanları atla (en az 1 karakter)
        yield [d for d in docs if d.get("title") and len(d["title"]) > 0]

Bu kod:

• title sütunu asla null olamaz — kütüphanede adı olmayan kitap olmaz.
• first_publish_year bigint olmalı — string gelirse reddedilir.
• Şema evolve eder: API yeni bir alan eklerse tabloya eklenir (breaking change yok).

Production’da evolve -> discard_value yapabilirsin: yeni alanlar tabloda görünmez ama hataya da yol açmaz.

Post-load Checks (Tabloda Alert)

Schema contract gelen veri için. Peki ya tabloya yazıldıktan sonra mantıksal sorunlar? (Yıl = -500, başlık = "????", vs.)

quality_check.py — bu script her pipeline run’ından sonra çalışır (CI’da, Kestra’da, cron’da):

import duckdb
import sys
from typing import List, Tuple

def run_check(name: str, sql: str, threshold: int = 0) -> Tuple[bool, str]:
    """Check çalıştır, threshold ihlal edilirse fail."""
    con = duckdb.connect("open_library.duckdb", read_only=True)
    result = con.execute(sql).fetchone()[0]
    con.close()
    passed = result <= threshold
    status = "PASS" if passed else "FAIL"
    msg = f"[{status}] {name}: {result} (threshold: {threshold})"
    return passed, msg

checks = [
    ("Boş başlık",        "SELECT COUNT(*) FROM raw_books.books WHERE title IS NULL OR TRIM(title) = ''", 0),
    ("Şüpheli yıl (eski)","SELECT COUNT(*) FROM raw_books.books WHERE first_publish_year < 1450", 0),
    ("Şüpheli yıl (yeni)","SELECT COUNT(*) FROM raw_books.books WHERE first_publish_year > 2026", 0),
    ("Boş author listesi","SELECT COUNT(*) FROM raw_books.books WHERE list_size(author_name) = 0 OR author_name IS NULL", 50),
    ("Duplicate key",     "SELECT COUNT(*) - COUNT(DISTINCT key) FROM raw_books.books", 0),
]

failures: List[str] = []
for name, sql, threshold in checks:
    passed, msg = run_check(name, sql, threshold)
    print(msg)
    if not passed:
        failures.append(name)

if failures:
    print(f"\n{len(failures)} check FAILED: {failures}")
    sys.exit(1)
else:
    print("\nTüm check'ler PASS.")

Çalıştır: python quality_check.py. Çıktıda her check için PASS/FAIL görürsün. Exit code 1 dönerse CI fail eder — alarm.

Alert’leri Gerçek Dünyada Bağlamak

Post-load check’ler tek başına çalışmaz, bağlama oturtulur:

Seviye	Araç	Ne zaman
Pipeline anında	dlt schema contract	Veri gelirken reddet
Pipeline sonrası	quality_check.py + cron	Her run’da kontrol
Sürekli izleme	Monte Carlo / Bigeye / Soda	Anomali tespiti (örn. dün 10K satır, bugün 100)
Bildirim	Slack / PagerDuty / email	Check fail olunca otomatik mesaj

Bu katmanların hepsi defense in depth — biri kaçırırsa diğeri yakalar.

dbt Test ile İlişki

Modül 4’te dbt öğreneceğiz. dbt’nin not_null, unique, accepted_values, relationships testleri aynı şeyi yapar, ama SQL dosyalarında statik tanımlanır. Avantajı: version control, kolay paylaşım, documentation otomatik.

Bugünkü Python check’leri dinamik — kodla her şeyi yapabilirsin. Dezavantajı: daha çok kod, daha az keşif (discovery).

İkisini birlikte kullanmak yaygın: critical kurallar (PK unique, NOT NULL) için dbt test, custom logic (yıl aralığı, dağılım) için Python check.

Hands-on

1. quality_check.py’yi çalıştır. Çıktıyı yorumla.
2. Yeni bir check ekle: “Ortalama sayfa sayısı 50’den büyük olmalı” (eğer number_of_pages_median sütunu varsa).
3. Bir check’i bilerek fail et: threshold = 0 yerine -1 koy, exit code’un değiştiğini gör.

Sıradaki

Veri temiz, alert’ler hazır. Sırada alternatif notebook deneyimi →