Veri boru hattı (ETL, data pipeline): çıkar, dönüştür, yükle

Tam yükleme mi, artımlı yükleme mi?

Derste her gün gelen dosyayı yükledin. Kaynak bir veritabanı ya da API olduğunda iki yol var: her gece tüm tabloyu çekmek (tam yükleme, full load) ya da yalnız son çalıştırmadan beri değişenleri çekmek (artımlı yükleme, incremental load). Küçük tablolarda tam yükleme basit ve güvenlidir. Milyonlarca satırda artımlı yükleme şarttır; bunun için bir su seviyesi işareti (watermark, high-water mark) tutulur: "en son hangi guncellenme zamanına kadar aldım?"

Watermark'ı yalnızca yükleme başarıyla bittikten sonra ilerlet; aksi hâlde yarım kalan çalıştırmanın kayıtları sonsuza kadar kaçar. Aynı zaman damgasına sahip satırlar için (guncellenme, id) çiftini watermark olarak tutmak ya da >= ile çekip upsert ile tekrarları zararsız yapmak iki yaygın çözümdür.

import sqlite3, json

kaynak = sqlite3.connect(":memory:")
kaynak.execute("CREATE TABLE musteri(id INTEGER PRIMARY KEY, sehir TEXT, guncellenme TEXT)")
kaynak.executemany("INSERT INTO musteri VALUES (?,?,?)", [
    (1, "İzmir", "2024-05-01T10:00"), (2, "Bursa", "2024-05-01T12:00"),
    (3, "Ankara", "2024-05-02T09:00")])

durum = {"watermark": "2024-05-01T11:00"}          # gerçekte dosya/tabloda saklanır

def artimli_cek(con, son):
    return con.execute("SELECT id, sehir, guncellenme FROM musteri "
                       "WHERE guncellenme > ? ORDER BY guncellenme", (son,)).fetchall()

yeni = artimli_cek(kaynak, durum["watermark"])
print("çekilen:", [r[0] for r in yeni])
# ... yükleme başarılı olduktan SONRA:
if yeni:
    durum["watermark"] = yeni[-1][2]
print(json.dumps(durum))

Büyük dosyalar: akış ve parti

10 GB'lık bir CSV'yi list(csv.DictReader(f)) ile belleğe almak sunucuyu düşürür. Üreteçlerle (iteratorler-uretecler dersi) her adım bir sonrakine satır satır akar; bellekte aynı anda yalnız bir parti durur. Veritabanına da satır satır değil parti parti (batch) yazılır: executemany ile 500-5000 satırlık partiler, her parti bir işlem (transaction) içinde. Python 3.12'deki itertools.batched bu işi tek satıra indirir.

import csv, io, sqlite3
from itertools import batched

ham = "no,tutar\n" + "\n".join(f"{i},{i * 1.5}" for i in range(1, 2501))

def oku(metin):                       # gerçekte: open(yol, newline="", encoding="utf-8")
    yield from csv.DictReader(io.StringIO(metin))

def donustur(satirlar):
    for s in satirlar:
        yield int(s["no"]), round(float(s["tutar"]) * 100)     # kuruş

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE satis(no INTEGER PRIMARY KEY, tutar_kurus INTEGER)")
parti_sayisi = 0
for parti in batched(donustur(oku(ham)), 1000):
    with con:                          # her parti ayrı işlem: ya hep ya hiç
        con.executemany("INSERT OR REPLACE INTO satis VALUES (?, ?)", parti)
    parti_sayisi += 1
print(parti_sayisi, "parti,", con.execute("SELECT COUNT(*), SUM(tutar_kurus) FROM satis").fetchone())

Veri kalitesi kapıları

Boru hattı hata vermeden bitti diye veri doğru değildir. Kaynak sistem bir gece sütun sırasını değiştirir, tutarlar kuruş yerine lira gelir ya da dosya yarım kesilir. Yükleme sonrası (ya da yayına almadan önce) küçük kontroller çalıştır ve eşik aşılırsa işi başarısız say:

• Satır sayısı mutabakatı: kaynaktaki satır = yüklenen + reddedilen. Fark varsa satır kaybolmuştur.
• Boşluk oranı: zorunlu sütunda None oranı eşiğin (ör. %1) üstündeyse dur.
• Tekillik: anahtar sütunda tekrar olmamalı.
• Aralık ve küme: tutar 0 ile makul bir üst sınır arasında, şehir bilinen listede.
• Tazelik: en yeni kaydın tarihi beklenen iş gününden eski olmamalı.
• Hacim: bugünkü satır sayısı son 7 günün ortalamasının yarısından azsa kaynakta sorun olabilir.

Bu kontroller büyüyünce Great Expectations, Soda ya da dbt testleri gibi araçlara taşınır (orkestrasyon dersindeki kalite kapıları); fikir aynıdır.

Atomik yazma ve dosya parmak izi

Çıktı dosyasını doğrudan hedef ada yazarken iş yarıda kesilirse, sonraki adım yarım bir dosya okur. Doğru yol: önce aynı klasörde geçici bir dosyaya yaz, tamamlanınca os.replace(gecici, hedef) ile adını değiştir. Aynı dosya sistemi içinde ad değiştirme atomiktir: okuyan ya eski dosyayı ya yeni dosyayı görür, yarısını asla. Gelen dosyalar için de parmak izi (SHA-256 özeti) tut: aynı içerik farklı adla ikinci kez gelirse tanırsın.

import hashlib, os, tempfile
from pathlib import Path

def atomik_yaz(yol: Path, metin: str) -> None:
    fd, gecici = tempfile.mkstemp(dir=yol.parent, suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(metin)
            f.flush()
            os.fsync(f.fileno())       # diske gerçekten yazıldığından emin ol
        os.replace(gecici, yol)        # atomik ad değiştirme
    except BaseException:
        os.unlink(gecici)
        raise

def parmak_izi(yol: Path) -> str:
    h = hashlib.sha256()
    with open(yol, "rb") as f:
        for parca in iter(lambda: f.read(65536), b""):
            h.update(parca)
    return h.hexdigest()

klasor = Path("etl_cikti"); klasor.mkdir(exist_ok=True)
atomik_yaz(klasor / "rapor.csv", "sehir,ciro\nİzmir,1250\n")
atomik_yaz(klasor / "rapor_kopya.csv", "sehir,ciro\nİzmir,1250\n")
print(parmak_izi(klasor / "rapor.csv") == parmak_izi(klasor / "rapor_kopya.csv"))
print(sorted(p.name for p in klasor.iterdir()))

Derste kalite kapılarını bir kontrol listesi olarak gördün. Bu bölüm onları uygulamaya döküyor: kalitenin boyutlarını ölçmek, kuralları koddan ayırıp bir şemaya yazmak, kötü satırları karantinaya almak, üretici ile tüketici arasındaki beklentiyi bir veri sözleşmesine bağlamak ve her çalıştırmanın sonunda okunabilir bir rapor üretmek. Örnekler gerçek kirli tablolarla: Anadolu Market'in raw_signups kayıt dökümü ve Power Query'den tanıdığın kirli_veri bayi raporu.

Kalitenin boyutları

Her boyutu bir orana çevirmek (ör. tamlık = zorunlu alanları dolu satır / tüm satırlar) kaliteyi zaman içinde izlenebilir kılar: "bu hafta tamlık %98'den %81'e düştü" demek, "veride sorun var" demekten çok daha işe yarar.

import veri, re
from collections import Counter
r = veri.kayitlar("anadolu_market", "raw_signups")
n = len(r)
tamlik = sum(x["city"] is not None for x in r) / n
tekillik = len({x["id"] for x in r}) / n
gecerlilik = sum(bool(re.fullmatch(r"\d{4}-\d{2}-\d{2}", x["signup_date"])) for x in r) / n
yazimlar = Counter(x["city"] for x in r if x["city"])
print(f"satır={n} tamlık={tamlik:.1%} tekillik={tekillik:.1%} ISO tarih={gecerlilik:.1%}")
print("şehir yazımları:", len(yazimlar), sorted(yazimlar)[:4])

Kurallar koddan ayrı: şema

Kuralları if yığınları olarak yazmak yerine bir şemada (schema) toplarsan hem okunur hem yeniden kullanılır. Python dünyasında bunun en yaygın aracı pandera'dır (pandas/polars DataFrame şeması); satır bazlı API girdisinde ise pydantic. Mantık aynı: sütun → tür + kurallar.

import pandera.pandas as pa   # pip install pandera
sema = pa.DataFrameSchema({
    "id":          pa.Column(int, unique=True),
    "email":       pa.Column(str, pa.Check.str_matches(r"^[^@\s]+@[^@\s]+\.[a-z]+$")),
    "city":        pa.Column(str, pa.Check.isin(["İstanbul", "Ankara", "İzmir"]), nullable=False),
    "signup_date": pa.Column("datetime64[ns]"),
}, strict=True)                # şemada olmayan sütun gelirse hata
try:
    sema.validate(df, lazy=True)          # lazy: ilk hatada durma, hepsini topla
except pa.errors.SchemaErrors as e:
    print(e.failure_cases)                # hangi sütun, hangi kural, hangi değer

Karantina tablosu: ne at, ne durdur

Kötü satırla ne yapmalı? Sessizce atmak veri kaybettirir ve kimse fark etmez; tüm partiyi durdurmak tek bir bozuk satır için bütün raporu bekletir. Orta yol karantinadır (quarantine): geçen satırlar hedefe, kalanlar ham hâliyle ve neden bilgisiyle ayrı bir tabloya yazılır. Veri sahibi karantinayı inceler, kaynağı düzeltir, satırlar yeniden işlenir. Karantina oranı eşiği aşarsa (ör. %5) parti yine durdurulur; bu, derste gördüğün hata kuyruğunun veritabanındaki karşılığıdır.

import sqlite3, json
c = sqlite3.connect(":memory:")
c.execute("CREATE TABLE karantina (parti TEXT, kayit TEXT, nedenler TEXT)")
bayi = veri.kayitlar("kirli_veri", "bayi")[2:]          # ilk iki satır başlık
for satir in bayi:
    nedenler = []
    if not satir["Column1"]:
        nedenler.append("bayi boş")
    if satir["Column4"] in (None, "-"):
        nedenler.append("tutar yok")
    if satir["Column3"] and not re.fullmatch(r"\d{2}\.\d{2}\.\d{4}", satir["Column3"]):
        nedenler.append("tarih geçersiz")
    if nedenler:
        c.execute("INSERT INTO karantina VALUES (?, ?, ?)",
                  ("2024-01", json.dumps(satir, ensure_ascii=False), ", ".join(nedenler)))
for parti, nedenler in c.execute("SELECT parti, nedenler FROM karantina"):
    print(parti, "|", nedenler)

Veri sözleşmesi (data contract)

Kaynak ekip bir gün sütun adını değiştirir ve senin boru hattın kırılır. Veri sözleşmesi, üretici ile tüketici arasında bu beklentileri yazıya döker: şema, kalite kuralları, tazelik taahhüdü (SLA), sahip ve iletişim. Çoğunlukla YAML'dır, git'te durur, değişiklikleri kod incelemesinden (code review) geçer ve CI'da otomatik denetlenir.

# contracts/raw_signups.yaml
dataset: anadolu_market.raw_signups
owner: crm-ekibi@anadolu.example
sla: {tazelik_saat: 24}
columns:
  - {name: id, type: int, required: true, unique: true}
  - {name: email, type: str, required: true, pattern: "^[^@ ]+@[^@ ]+$"}
  - {name: city, type: str, required: true, allowed: [İstanbul, Ankara, İzmir]}
  - {name: signup_date, type: str, required: true, pattern: "^\\d{4}-\\d{2}-\\d{2}$"}
quality:
  max_karantina_orani: 0.05

Sözleşmeyi koda çevirmek, şema doğrulayıcını YAML'dan beslemek demektir: yaml.safe_load ile oku, her sütun tanımını bir kurala dönüştür, sonuçları bir rapora topla. Raporun sonunda tek bir karar olmalı: GEÇTİ ya da KALDI. Bu raporu JSON olarak saklamak, kalitenin zaman içindeki eğilimini panoda göstermeyi sağlar.