Artımlı veri çekme: watermark, upsert ve geri dolum tasarımı

Artımlı veri çekme, bir kaynak sistemden hedefe veri taşırken her koşumda tabloyu baştan okumak yerine yalnızca son koşumdan bu yana değişen kayıtları almak demek. Kâğıt üzerinde basit görünüyor: bir updated_at sütunu, bir “son görülen zaman” değeri, bir WHERE koşulu. Pratikte ise kaçırılan satırlar, iki kez yazılan kayıtlar ve kaynakta silinip hedefte yaşamaya devam eden hayalet veriler tam bu basit görünen yerden sızıyor.

Bu yazıda kendi boru hatlarımda zamanla oturttuğum deseni anlatıyorum: yüksek su işareti (watermark) takibi, sınır çakışmalarına karşı örtüşme payı, doğal anahtar üzerinden upsert, silinen kayıtların tespiti ve geri dolumun (backfill) normal koşumu bozmadan nasıl yürütüleceği. Örnekler Python ile; kaynak olarak PostgreSQL, hedef olarak SQLite varsaydım ama mantık ikisi arasında değişmiyor.

Hedefim şu: koşum yarıda kesilse, iki kez çalışsa ya da üç gün geç kalsa bile hedef tablo sonunda kaynağa yakınsasın. Bunu sağlayan şey hız değil, her adımın idempotent olması.

Watermark takibi: neyi, nerede saklamalı

Watermark, “bu zamana kadar olan değişiklikleri hedefe yazdım” iddiasıdır. Bu iddianın güvenilir olması için dört koşul koyarım:

• Kaynakta güvenilir bir updated_at olmalı. Uygulama kodunun elle doldurduğu sütuna güvenmem; unutulmuş tek bir UPDATE yolu o satırı sonsuza dek görünmez kılar. Sütunu veritabanı tetikleyicisiyle doldurtun.
• Watermark, satırlarla aynı işlemde (transaction) yazılmalı. Durumu ayrı bir JSON dosyasında tutup satırları veritabanına yazarsanız, ikisinin arasında kesilen bir koşum ya veri kaybeder ya da watermark'ı geride bırakır. Hedef SQLite ise durumu da aynı dosyada, küçük bir cekim_durumu tablosunda tutarım; commit ikisini birlikte yazar ya da hiçbirini yazmaz.
• Watermark'ı “şimdi” değil, gerçekten gördüğünüz en büyük updated_at yapın. Sorgu bitene kadar saat ilerler; o arada yazılan satırları görmediniz ama “şimdi” değerini kaydederseniz görmüş gibi davranırsınız.
• Her akışın kendi watermark'ı olsun. Sipariş, müşteri ve geri dolum akışları aynı tabloda ayrı satırlardır; biri takılınca diğeri durmaz.

İlk koşumda watermark yoktur; o koşum zaten tam çekimdir ve aynı kodla yapılır. Ayrı bir “ilk yükleme” betiği yazmam, çünkü iki ayrı yol iki ayrı hata kaynağı demektir.

Küçük hacimli ama sağlam boru hatlarında SQLite'ı hem hedef hem durum deposu olarak kullanmanın neden işe yaradığını ayrı bir yazıda anlatmıştım; buradaki çekici o yaklaşımın devamı.

Örtüşme payı: sınırdaki satırlar neden kaçar

Watermark'ı doğru kaydetseniz bile bir tuzak kalır: updated_at değeri işlemin başında atanır, satır ise işlem commit olduğunda görünür olur. Uzun süren bir işlem, 10:00:00 damgalı satırı 10:00:40'ta yayınlayabilir. Sizin koşumunuz 10:00:20'de çalışıp watermark'ı 10:00:30'a çektiyse o satır bir sonraki koşumda da gelmez; damgası artık watermark'ın gerisindedir.

Çözüm, her koşumda watermark'ın biraz gerisinden başlamak; buna örtüşme payı diyorum. Payın evrensel bir değeri yok; ben kaynaktaki en uzun işlem süresinin birkaç katını alırım. Bedeli, aynı satırların tekrar gelmesidir ve upsert idempotent olduğu sürece bu bedel sıfıra yakın kalır:

• Aynı updated_at ile tekrar gelen satır hedefte hiçbir şey değiştirmez; yalnızca birkaç okuma harcanır.
• Zamanı iki tarafta da UTC ve sabit biçimde tutun. SQLite'ta ISO 8601 metin olarak saklanan zamanlar ancak biçim aynıysa doğru sıralanır; “+03:00” ile “Z” karışırsa karşılaştırma sessizce bozulur.
• Örtüşme kaçan satırı kurtarır ama silinen satırı kurtarmaz; ona ayrıca bakmak gerekir.
• Kaynak sunucunun saati de hesaba girer; pay, olası saat kaymasını yutacak kadar geniş olmalı.

Doğal anahtar üzerinden upsert ve çekici fonksiyon

Hedefte kendi ürettiğiniz otomatik artan kimliğe göre çakışma çözmeye çalışmayın; çakışma hedefi kaynağın anahtarı olmalı. Birden fazla kaynak varsa anahtar bileşiktir: (kaynak_sistem, kaynak_id). SQLite 3.24 ve PostgreSQL 9.5 sonrası aynı ON CONFLICT sözdizimini kabul eder; yalnızca yer tutucu (? ile %s) ve tipler değişir. Sorguyu ayrı bir dosyada tutuyorum ki normal koşum ile geri dolum aynı metni kullansın.

-- Hedef şema (SQLite). PostgreSQL'de yalnızca tipler ve yer tutucu (? yerine %s) değişir.
CREATE TABLE IF NOT EXISTS siparis (
    kaynak_id   INTEGER PRIMARY KEY,        -- kaynağın anahtarı, hedefin değil
    musteri_no  TEXT    NOT NULL,
    tutar       INTEGER NOT NULL,           -- kuruş
    updated_at  TEXT    NOT NULL,           -- ISO 8601, UTC
    silindi_mi  INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS cekim_durumu (
    akis      TEXT PRIMARY KEY,
    watermark TEXT NOT NULL
);

-- sql/upsert_siparis.sql — SQLite 3.24+ ve PostgreSQL 9.5+ aynı sözdizimini kabul eder
INSERT INTO siparis (kaynak_id, musteri_no, tutar, updated_at)
VALUES (?, ?, ?, ?)
ON CONFLICT (kaynak_id) DO UPDATE SET
    musteri_no = excluded.musteri_no,
    tutar      = excluded.tutar,
    updated_at = excluded.updated_at,
    silindi_mi = 0                          -- yeniden görünen kayıt canlanır
WHERE excluded.updated_at > siparis.updated_at;

Sondaki WHERE bu tasarımın kilit taşı: hedefteki satır kaynaktan gelenden daha yeniyse dokunulmaz. Böylece geciken bir koşum, örtüşmeyle tekrar gelen satırlar ve geri dolum, güncel veriyi asla eskisiyle ezemez. Çekici fonksiyon bu sorgunun etrafında kuruluyor; aynı fonksiyon isteğe bağlı bir üst sınır alarak geri dolumda da kullanılıyor.

from datetime import datetime, timedelta, timezone

ORTUSME = timedelta(minutes=15)    # kaynaktaki en uzun işlem süresinin birkaç katı
BASLANGIC = datetime(2000, 1, 1, tzinfo=timezone.utc)
UPSERT_SQL = open("sql/upsert_siparis.sql", encoding="utf-8").read()

def watermark_oku(hedef, akis):
    satir = hedef.execute(
        "SELECT watermark FROM cekim_durumu WHERE akis = ?", (akis,)
    ).fetchone()
    return datetime.fromisoformat(satir[0]) if satir else BASLANGIC

def pencere_cek(kaynak, hedef, akis, alt, ust=None):
    """kaynak: psycopg bağlantısı (PostgreSQL, %s), hedef: sqlite3 bağlantısı (?)."""
    sql = "SELECT id, musteri_no, tutar, updated_at FROM siparis WHERE updated_at >= %s"
    parametreler = [alt]
    if ust is not None:
        sql += " AND updated_at < %s"
        parametreler.append(ust)
    with kaynak.cursor() as cur:
        cur.execute(sql + " ORDER BY updated_at, id", parametreler)
        satirlar = [(r[0], r[1], r[2], r[3].astimezone(timezone.utc).isoformat())
                    for r in cur]
    if not satirlar and ust is None:
        return 0                                   # normal akış: görülen yoksa ilerleme yok
    yeni_wm = ust.isoformat() if ust else max(r[3] for r in satirlar)
    with hedef:   # tek işlem: satırlar ve watermark ya birlikte yazılır ya hiç
        hedef.executemany(UPSERT_SQL, satirlar)
        hedef.execute(
            "INSERT INTO cekim_durumu (akis, watermark) VALUES (?, ?) "
            "ON CONFLICT (akis) DO UPDATE SET watermark = excluded.watermark",
            (akis, yeni_wm),
        )
    return len(satirlar)

def artimli_cek(kaynak, hedef, akis="siparis"):
    alt = watermark_oku(hedef, akis) - ORTUSME     # örtüşme payı burada
    return pencere_cek(kaynak, hedef, akis, alt)

def geri_dolum(kaynak, hedef, bas, bit, akis):
    bas = max(bas, watermark_oku(hedef, akis))     # yarıda kesildiyse kaldığı yerden
    while bas < bit:
        ust = min(bas + timedelta(days=1), bit)
        pencere_cek(kaynak, hedef, akis, bas, ust)
        bas = ust

Dikkat edilecek iki nokta var. with hedef: bloğu satırları ve watermark'ı tek işlemde yazar; hata olursa ikisi de geri alınır. Normal akışta watermark gerçekten görülen en büyük damgadır, geri dolumda ise pencerenin sonu; çünkü boş bir gün geri dolumu durdurmamalı.

Normal akışın kaçırdıkları: silinen kayıtlar ve geri dolum

Silinen kayıtlar için üç yol bilirim ve bu sırayla tercih ederim:

• Kaynakta yumuşak silme varsa (deleted_at ya da is_deleted), silme bir güncellemedir: updated_at ilerler, satır normal akıştan gelir. O sütunu hedefe taşıyıp silindi_mi bayrağını oradan beslerim; hedefte fiziksel silme yapmam.
• Kaynak fiziksel siliyorsa, günde bir kez yalnızca anahtar sütununu çekip hedefteki anahtarlarla karşılaştırırım; hedefte olup kaynakta olmayanlar silindi_mi = 1 olur. Anahtar listesi, tam tablonun yanında çok küçüktür.
• Değişiklik günlüğü (CDC) varsa silmeler oradan gelir; ama bu altyapı her ortamda yok.

Hangi yolu seçerseniz seçin, silmeyi hedefte bir bayrak olarak taşımak raporların geçmişini korur ve yanlışlıkla silinen kaydın geri gelmesini kolaylaştırır.

Geri dolum ihtiyacı üç durumda doğar: hedefe yeni bir sütun eklediniz, bir dönemin verisinden şüpheleniyorsunuz ya da kaynak bir hatayı geriye dönük düzeltti ama updated_at'e dokunmadı. Hepsinde aynı kural geçerli: geri dolum ana watermark'a dokunmaz.

• Yukarıdaki geri_dolum fonksiyonu cekim_durumu tablosunda kendi akış adıyla (siparis:geri:2025-Q1 gibi) ilerler; yarıda kesilirse kaldığı pencereden devam eder.
• Pencereler bir günlük. Tek bir dev sorgu hem kaynağı yorar hem de yarıda kesildiğinde geride hiçbir şey bırakmaz.
• Aynı upsert kullanıldığı için normal koşumla aynı anda çalışabilir; WHERE koruması, normal koşumun az önce yazdığı daha yeni satırın ezilmesini engeller.
• updated_at'e dokunulmadan düzeltilmiş veri için yalnızca o pencereye özel, koşulu >= olan bir sorgu kopyası kullanırım; kalıcı sorguyu gevşetmem.

Geri dolumu çoğu zaman normal işle aynı makinede ama ayrı bir konteynerde koşturuyorum; iki işin ağ ve hacim adlarının birbirine karışmaması için Compose proje adı izolasyonu tam bu noktada işe yarıyor.

Tutarlılığı doğrulamak

“Koşum hatasız bitti” ile “hedef kaynağa eşit” aynı şey değil. Doğrulamayı koşumun parçası yaparım, ayrı bir iş olarak değil:

• Kova bazlı sayım: iki tarafta da güne göre satır sayısı ve bir toplam alıp farkı listelerim. Fark çıkan gün, geri dolum penceresinin ta kendisidir.
• Uç değer: kaynaktaki MAX(updated_at) ile hedefteki watermark arasındaki mesafe örtüşme payını aşıyorsa alarm.
• Rastgele örnek: her koşumda birkaç anahtarı iki taraftan sütun sütun karşılaştırırım; sayımların eşit, içeriğin farklı olduğu vakaları yalnızca bu yakalar.
• Sessiz başarısızlık: sıfır satır çeken bir koşum hata değildir, ama üst üste birkaç koşum sıfır çekiyorsa kaynak durmuş ya da watermark bozulmuş olabilir. Sayaçları kaydedip eşik koyarım.

-- SQLite tarafı; PostgreSQL'de gün için (updated_at AT TIME ZONE 'UTC')::date kullanın
SELECT substr(updated_at, 1, 10) AS gun,
       COUNT(*)   AS adet,
       SUM(tutar) AS toplam
FROM siparis
WHERE updated_at >= '2025-01-01' AND silindi_mi = 0
GROUP BY gun
ORDER BY gun;

İki çıktıyı gün bazında birleştirip farkı sıfır olmayan satırları yazdıran kısa bir betik, çoğu tutarsızlığı daha kimse fark etmeden önünüze koyar.

Bu desenin güzel yanı parçaların birbirini korumasıdır: örtüşme kaçan satırı getirir, upsert koruması tekrarı zararsız kılar, ayrı akışlı geri dolum geçmişi düzeltir, kova sayımı da hepsinin gerçekten işe yarayıp yaramadığını söyler. Kendi veri akışınızda benzer bir yapı kurmak ya da mevcut “her gece her şeyi çek” işini artımlıya çevirmek istiyorsanız iletişim bölümünden yazabilirsiniz; danışmanlık çalışmalarımda en sık dokunduğum alanlardan biri tam olarak bu.