Bitirme projesi: Anadolu Market web olayları ETL

Proje: Anadolu Market web olayları için ETL

Anadolu Market'in analiz ekibi huni raporunu her sabah elle düzeltiyor: bazı günlerde aynı olay iki kez sayılıyor, sistem yeniden çalışınca rakamlar şişiyor, ertesi gün gelen eski olaylar rapora girmiyor. Senden her gece çalışacak bir veri işi istiyorlar: ham dosyayı alsın, bozuk satırı ayırsın, temiz satırı SQLite'a yüklesin, aynı iş iki kez çalışsa da hiçbir şey çiftlenmesin ve sabah neyin olduğu görülebilsin.

Bu ders yeni kavramdan çok, Veri Mühendisi yolunda öğrendiklerini tek bir çalışan işe bağlar. Veri boru hattı (ETL, data pipeline): çıkar, dönüştür, yükle dersindeki parçaları gerçek veriyle ve kanıtıyla kuracaksın. Her adımda önceki adımın kodu hazır gelir; sonda sonucu testle kanıtlarsın.

Ham veri: gerçek tablo, belgelenmiş bozulma

Kaynak, anadolu_market setindeki web_events tablosu: 2.819 olay, 60 gün. Bu tabloda iki gerçek kusur zaten var: 40 olay iki kez gönderilmiş (farklı event_id, aynı kullanıcı, tür, zaman ve cihaz) ve bazı kullanıcıların istemci saati kaymış. Reddedilecek satır ise yok; bu yüzden ham_partiler() satırları günlük CSV dosyalarına bölerken numarası belli kurallara göre bozulmalar ekler. Bozulmaların kuralı kodda açıktır, rastgele değildir; yani sonuçlar her çalıştırmada aynıdır. Bu ekleme derse aittir, veri setinin kendisinde yoktur.

import csv, io
from datetime import date, datetime, timedelta
import veri

SUTUNLAR = ["event_id", "user_id", "event_type", "event_time", "device"]

def ham_partiler():
    """Gerçek web_events satırlarını günlük CSV partilerine böler ve belgelenmiş bozulmaları ekler."""
    gunler = {}
    for r in veri.kayitlar("anadolu_market", "web_events"):
        s = {k: str(r[k]) for k in SUTUNLAR}
        n = r["event_id"]
        z = s["event_time"]
        parti = z[:10]
        if n % 101 == 0:   # düzeltilebilir: harf ve boşluk
            s["event_type"] = " " + s["event_type"].upper() + " "
        if n % 103 == 0:   # düzeltilebilir: başka tarih biçimi
            s["event_time"] = z[8:10] + "." + z[5:7] + "." + z[:4] + z[10:]
        if n % 211 == 0:   # reddedilir: kullanıcı yok
            s["user_id"] = ""
        if n % 223 == 0:   # reddedilir: tanınmayan olay
            s["event_type"] = "sayfa_gorunum"
        if n % 227 == 0:   # reddedilir: imkânsız tarih
            s["event_time"] = z[:5] + "13" + z[7:]
        if n % 229 == 0:   # reddedilir: tanınmayan cihaz
            s["device"] = "buzdolabi"
        if n % 59 == 0:    # geç gelen satır: iki gün sonraki partide
            parti = (date.fromisoformat(parti) + timedelta(days=2)).isoformat()
        gunler.setdefault(parti, []).append(s)
    partiler = {}
    for parti in sorted(gunler):
        f = io.StringIO()
        y = csv.DictWriter(f, SUTUNLAR, lineterminator="\n")
        y.writeheader()
        y.writerows(gunler[parti])
        partiler[parti] = f.getvalue()
    return partiler

partiler = ham_partiler()
print(len(partiler), "parti,", sum(m.count("\n") - 1 for m in partiler.values()), "satır")
print(partiler["2024-10-01"][:170])

Doğrulama: düzeltilebilir olanla olmayanı ayır

Doğrulamanın amacı satır atmak değil, karar vermektir. VISIT ile visit aynı şeydir: düzelt. Boş kullanıcı numarası düzeltilemez: uydurma, ayır. Reddedilen satırı nedeniyle bir tabloya yazmak, kaynak ekibine "şu satırlarınız şu sebeple gelmedi" diyebilmeni sağlar.

Doğal anahtar ve idempotent yükleme

Akla ilk gelen anahtar event_id'dir, ama burada yanlıştır: yinelenen olayların numaraları farklıdır. Anahtar, bir olayı olay yapan alanlardır: kullanıcı, tür, zaman, cihaz. Kısıtı veritabanına koyarsın, çünkü bu kuralın her kod yolundan geçerli olmasını istersin.

import sqlite3
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE o (id INTEGER, kul INTEGER, tur TEXT, zaman TEXT, UNIQUE (kul, tur, zaman))")
satirlar = [(1, 5, "visit", "10:00"), (2, 5, "visit", "10:00"), (3, 6, "visit", "10:05")]
for tur_no in (1, 2):
    once = con.total_changes
    con.executemany("INSERT INTO o VALUES (?, ?, ?, ?) ON CONFLICT (kul, tur, zaman) DO NOTHING", satirlar)
    print("tur", tur_no, "eklenen:", con.total_changes - once, "toplam:", con.execute("SELECT COUNT(*) FROM o").fetchone()[0])

Olaylar değişmez, bu yüzden DO NOTHING yeter. Sipariş durumu gibi değişen veride DO UPDATE kullanırsın; Veri boru hattı (ETL, data pipeline): çıkar, dönüştür, yükle dersi onu gösterdi.

Artımlı yükleme: filigran neye bakmalı?

Her gece tüm tarihçeyi yeniden işlemek israftır; yalnız yeniyi işlemek istersin. "Buraya kadar işledim" kaydına filigran (watermark) denir. Asıl soru filigranın neye baktığıdır. Olay zamanına bakarsan geç gelen olayı kaybedersin; bu veride bu şekilde 46 olay kayboluyor ve hiçbir hata çıkmıyor. Parti tarihine bakarsan olay ne kadar eski olursa olsun, geldiği partiyle işlenir.

from datetime import date, timedelta
filigran = "2024-10-05"; geri_bakis = 2
alt = (date.fromisoformat(filigran) - timedelta(days=geri_bakis)).isoformat()
gelenler = ["2024-10-03", "2024-10-04", "2024-10-05", "2024-10-06"]
print("işlenecek:", [p for p in gelenler if p > alt])

Geri bakış neden var? Kaynak bazen son günlerin dosyasını düzeltip yeniden gönderir. Son iki günü her gece yeniden okumak bunu yakalar. Bu da ancak yükleme idempotent olduğu için güvenlidir: artımlılık ve idempotentlik birbirini tamamlar.

Parti başına tek işlem ve günlük

Her parti tek bir veritabanı işlemidir (transaction): satırlar, reddedilenler, günlük kaydı ve filigran ya birlikte yazılır ya hiç yazılmaz. Python'da with con: bloğu bunu sağlar. Günlüğün iki yüzü vardır: logging ile akan satırlar sabah ne olduğunu anlatır, calisma_gunlugu tablosu ise sayıları sorgulanabilir tutar.

import logging, sqlite3
log = logging.getLogger("ornek"); log.setLevel(logging.INFO)
log.handlers[:] = [logging.StreamHandler(__import__("sys").stdout)]; log.propagate = False
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE t (x INTEGER)")
try:
    with con:
        con.execute("INSERT INTO t VALUES (1)")
        raise RuntimeError("disk doldu")
except RuntimeError as e:
    log.error("parti geri alındı: %s", e)
print("satır:", con.execute("SELECT COUNT(*) FROM t").fetchone()[0])

Kalite testleri: hata ve uyarı ayrımı

Bazı ihlaller işi geçersiz kılar: sayılar tutmuyorsa ya da çift satır varsa raporu yayımlama. Bazıları ise gerçek dünyanın kusurudur: saati kaymış istemci gerçek bir kayıttır, silmezsin, sayar ve raporlarsın. En önemli test satır mutabakatıdır: gelen = eklenen + tekrar + red. Bir satır bu dengeye girmiyorsa bir yerde kaybolmuştur.

Bitince göreceğin sonuç

Aşağıdaki sayılar bu dersin kodu gerçek veride çalıştırılarak hesaplandı; son adımda kendi kodun da aynısını üretecek.