Toplu ve Akış İşleme

Önceki ders üç kaynak sınıfının profilini on iki dönemin hepsi elde varken çıkardı: bozulmuş dönem hattın içinde duruyordu ve kalan on bir dönemle karşılaştırılabiliyordu. Bir hat böyle çalışmaz; bir çizelgeye bağlıdır ve çizelge bozulmuş döneme ne zaman bakılacağını belirler.

Bu ders aynı hattı iki kipte koşturur. Toplu işleme (batch processing) dönem kapandıktan sonra çalışır ve o ana kadar gelmiş bütün satırları baştan işler; akış işleme (stream processing) satırı geldiği dönemde işler ve sonraki döneme durum taşır. İkisi aynı beş aşamayı, aynı altı bozulma sınıfını ve aynı denetimleri kullanır; tek farkları penceredir.

• YD47. Kod kurgudur; gerçek bir akış sistemi, kuyruk ürünü ya da orkestratör çağrılmaz. Kaynak önceki dersin kurgu sayaç okumalarıdır: 200 abone, 12 dönem, 2400 satır, tohum 20260218.
• YD48. Kuyruk, yeniden deneme ve gecikme yordamı tekrarlanmaz; M16’nın Önbellekleme, Kuyruklar ve Eşzamansız İşleme kursunda ve M19’un Uygulama Katmanı ve Servis Etkileşimi kursunda ölçüldü. Buraya eklenen tek şey bozulmanın hangi dönemde görüldüğüdür.
• YD49. Toplu kipte koşum, o ana kadar gelmiş bütün satırları baştan işler; akış kipinde satır geldiği dönemde işlenir ve ilerleme damgası (watermark) penceresi kapanmış satırı düşürür.
• YD50. Geç gelen satır bu derste “hiç gelmedi” değil bir dönem sonra geldi sayılır. Bozulma sınıfı aynıdır; eklenen tek şey satırın geliş dönemidir.
• YD51. İki kipte de aynı beş denetim açıktır: şema, tür, tekillik, bölüm ve dağılım. Bölüm denetimi kusurlu olanıdır — bölge kümesine bakar, dönemi görmez.
• YD52. Gecikme iki birimde yazılır: dönem ve aşama. Özet hücresi oranının adımı 1/55, yani 0,0182.

İki Kip, Tek Hat

Aşağıdaki blok ortak tanımı kurar ve iki koşucu ekler. toplu her dönem başında, o ana kadar gelmiş satırları baştan işler ve hiçbir şey taşımaz. akis her dönemi kendi penceresinde işler ve iki şey taşır: görülen anahtar kümesi ile abone başına son endeks — tekilleştirme ve tüketim aşamalarının toplu kipte bedava bulduğu bilgi.

# KURGUDUR. Gercek bir akis sistemi, kuyruk ya da orkestrator cagrilmaz; olcum
# aginin sayac okumalari kurgudur, tohum 20260218.
TOHUM, M32 = 20260218, 0xFFFFFFFF


def uretec(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def sonraki():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return sonraki


BOLGE = ["dogu", "bati", "kuzey", "guney", "merkez"]
TIP = ["mesken", "isyeri"]
DONEM = [f"2026-{a:02d}" for a in range(1, 13)]
ABONE = {}
for i in range(200):
    r = uretec(TOHUM + 37 * i)
    ABONE[f"A{i + 1:03d}"] = {"bolge": BOLGE[int(r() * 5)], "tip": TIP[int(r() * 2)],
                              "baslangic": 10000 + int(r() * 60000)}


def kaynak_uret():                                  # her donem her abone icin bir okuma
    kayit = []
    for i, (ab, a) in enumerate(ABONE.items()):
        e = a["baslangic"]
        for d in DONEM:
            r = uretec(TOHUM + 101 * i + 7 * DONEM.index(d))
            e += 4 + int(r() * 46)
            kayit.append({"abone": ab, "donem": d, "bolge": a["bolge"], "tip": a["tip"],
                          "endeks": e, "birim": "m3"})
    return kayit


BOZULMA = ["gec_gelen", "yinelenen", "sema_degisti", "birim_degisti", "eksik_bolum", "kayan_tip"]


def boz(kayit, sinif, donem="2026-07"):
    """Tek bir donemi bozar , kalan donemler saglam kalir."""
    yeni, sayac = [], 0
    for k in kayit:
        if k["donem"] != donem:
            yeni.append(k)
            continue
        if sinif == "gec_gelen":
            if k["abone"] <= "A040":                # bu donemde 40 abone hic gelmedi
                sayac += 1
                continue
            yeni.append(k)
        elif sinif == "yinelenen":
            yeni.append(k)
            if k["abone"] <= "A060":                # 60 kayit iki kez geldi
                yeni.append(dict(k))
                sayac += 1
        elif sinif == "sema_degisti":
            y = dict(k)
            y["endeks_degeri"] = y.pop("endeks")    # alan adi degisti
            sayac += 1
            yeni.append(y)
        elif sinif == "birim_degisti":
            y = dict(k)
            y["endeks"] = k["endeks"] * 1000        # m3 yerine litre
            y["birim"] = "m3"                       # ETIKET DEGISMEDI , sessiz bozulma
            sayac += 1
            yeni.append(y)
        elif sinif == "eksik_bolum":
            if k["bolge"] == "kuzey":               # bir bolgenin bolumu hic yazilmadi
                sayac += 1
                continue
            yeni.append(k)
        elif sinif == "kayan_tip":
            y = dict(k)
            y["endeks"] = str(k["endeks"])          # sayi metne dondu
            sayac += 1
            yeni.append(y)
    return yeni, sayac


def a1_ayikla(kayit):
    ZOR = ("abone", "donem", "bolge", "endeks", "birim")
    tut = [k for k in kayit if all(a in k for a in ZOR)]
    return tut, len(kayit) - len(tut)


def a2_tekille(kayit):
    gor, tut = set(), []
    for k in kayit:
        ad = (k["abone"], k["donem"])
        if ad in gor:
            continue
        gor.add(ad)
        tut.append(k)
    return tut, len(kayit) - len(tut)


def a3_tuketim(kayit):
    tablo = {(k["abone"], k["donem"]): k for k in kayit}
    cikti, dusen = [], 0
    for k in kayit:
        i = DONEM.index(k["donem"])
        if i == 0:
            continue
        onceki = tablo.get((k["abone"], DONEM[i - 1]))
        if onceki is None:
            dusen += 1
            continue
        try:
            f = int(k["endeks"]) - int(onceki["endeks"])
        except (TypeError, ValueError):
            dusen += 1
            continue
        cikti.append(dict(k, m3=f))
    return cikti, dusen


def a4_ozet(kayit):
    top = {}
    for k in kayit:
        top[(k["bolge"], k["donem"])] = top.get((k["bolge"], k["donem"]), 0) + k["m3"]
    return top, 0


def a5_sun(ozet):                                   # bolge basina donem ortalamasi
    bolge = {}
    for (b, d), v in ozet.items():
        bolge.setdefault(b, []).append(v)
    return {b: round(sum(v) / len(v), 2) for b, v in bolge.items()}, 0


ASAMA = [("ayikla", a1_ayikla), ("tekille", a2_tekille), ("tuketim", a3_tuketim),
         ("ozet", a4_ozet), ("sun", a5_sun)]


def d_sema(veri, alanlar=("abone", "donem", "bolge", "endeks", "birim")):
    return any(any(a not in k for a in alanlar) for k in veri)


def d_tip(veri):
    return any(not isinstance(k.get("endeks"), int) for k in veri)


def d_tekillik(veri):
    ad = [(k["abone"], k["donem"]) for k in veri]
    return len(ad) != len(set(ad))


def d_bolum_tamligi(veri):
    """Bolge kumesine bakar , donemi gormez: tum tarihte kusurlu , pencerede degil."""
    return set(k["bolge"] for k in veri) != set(BOLGE)


def d_dagilim(ozet, taban, pay=0.25):
    for ad, v in ozet.items():
        t = taban.get(ad)
        if t and abs(v - t) / max(abs(t), 1) > pay:
            return True
    return False


DENETIM = [("sema", 0, lambda v, t: d_sema(v)),
           ("tip", 0, lambda v, t: d_tip(v)),
           ("tekillik", 0, lambda v, t: d_tekillik(v)),
           ("bolum", 0, lambda v, t: d_bolum_tamligi(v)),
           ("dagilim", 4, lambda v, t: d_dagilim(v, t))]


def tam_hat(kayit):
    v, _ = a1_ayikla(kayit)
    v, _ = a2_tekille(v)
    v, _ = a3_tuketim(v)
    oz, _ = a4_ozet(v)
    return oz


def teslim_et(kayit, sinif):
    """Her satira bir gelis donemi verir ; gec gelen satir bir donem sonra gelir."""
    bozuk, n = boz(kayit, sinif)
    veri = [dict(k, gelis=k["donem"]) for k in bozuk]
    if sinif == "gec_gelen":
        veri += [dict(k, gelis="2026-08") for k in kayit
                 if k["donem"] == "2026-07" and k["abone"] <= "A040"]
    return veri, n


ASAMA_TOPLU = [a for a, _ in ASAMA]
ASAMA_AKIS = ["ilerleme_damgasi"] + ASAMA_TOPLU + ["durum_anahtar", "durum_endeks"]


def toplu(veri, taban):
    """Kosum , bir onceki donemin sonuna kadar GELMIS butun satirlari bastan isler."""
    okunan, yak, asama, donem = 0, None, None, None
    for i, d in enumerate(DONEM):
        gelmis = [k for k in veri if DONEM.index(k["gelis"]) < i]
        okunan += len(gelmis)
        v = gelmis
        for adim in range(6):
            if adim > 0 and v:
                v, _ = ASAMA[adim - 1][1](v)
            for ad, das, df in DENETIM:
                if das == adim and yak is None and v and df(v, taban):
                    yak, asama, donem = ad, adim, d
    return {"okunan": okunan, "yakalayan": yak, "asama": asama, "donem": donem,
            "durum": 0, "gec_dusen": 0, "hucre": tam_hat(veri)}


def akis(veri, taban):
    """Donem penceresi: satir geldigi donemde islenir , durum sonraki doneme tasinir."""
    gorulen, son_endeks, hucre = set(), {}, {}
    okunan, gec_dusen, durum = 0, 0, 0
    yak, asama, donem = None, None, None
    for d in DONEM:
        pencere = [k for k in veri if k["gelis"] == d]
        okunan += len(pencere)
        v = [k for k in pencere if k["donem"] == d]         # su isareti: pencere kapandi
        gec_dusen += len(pencere) - len(v)
        for adim in range(5):
            if adim == 1:
                v, _ = a1_ayikla(v)
            elif adim == 2:
                yeni, g = [], set()
                for k in v:
                    ad = (k["abone"], k["donem"])
                    if ad not in gorulen and ad not in g:
                        g.add(ad)
                        yeni.append(k)
                gorulen |= g                                # durum: gorulen anahtar
                v = yeni
            elif adim == 3:
                c = []
                for k in v:
                    try:
                        c.append(dict(k, m3=int(k["endeks"]) - int(son_endeks[k["abone"]])))
                    except (KeyError, TypeError, ValueError):
                        pass
                for k in v:
                    try:
                        son_endeks[k["abone"]] = int(k["endeks"])   # durum: son endeks
                    except (TypeError, ValueError):
                        pass
                v = c
            elif adim == 4:
                o = {}
                for k in v:
                    o[(k["bolge"], d)] = o.get((k["bolge"], d), 0) + k["m3"]
                hucre.update(o)
                v = o
            for ad, das, df in DENETIM:
                if das == adim and yak is None and v and df(v, taban):
                    yak, asama, donem = ad, adim, d
        durum = max(durum, len(gorulen) + len(son_endeks))
    return {"okunan": okunan, "yakalayan": yak, "asama": asama, "donem": donem,
            "durum": durum, "gec_dusen": gec_dusen, "hucre": hucre}


ham = kaynak_uret()
TABAN = tam_hat(ham)
saglam = [dict(k, gelis=k["donem"]) for k in ham]
st, sa = toplu(saglam, TABAN), akis(saglam, TABAN)
print("taban ozet", len(TABAN), "hucre , sunum", a5_sun(TABAN)[0])
print("kip    asama  okunan satir  tasinan durum  yakalayan")
for ad, r, n in (("toplu", st, len(ASAMA_TOPLU)), ("akis", sa, len(ASAMA_AKIS))):
    print(f"{ad:<6} {n:>6} {r['okunan']:>13} {r['durum']:>14}  {r['yakalayan']}")

taban ozet 55 hucre , sunum {'merkez': 1321.45, 'kuzey': 1212.55, 'dogu': 881.36, 'bati': 833.09, 'guney': 1065.55}
kip    asama  okunan satir  tasinan durum  yakalayan
toplu       5         13200              0  None
akis        8          2400           2600  None

Sağlam koşumda iki kip de aynı 55 özet hücresini ve aynı beş sunum değerini üretiyor, ikisinde de hiçbir denetim ateşlemiyor. Bedel satırı burada okunuyor. Toplu kip on iki koşumda 13200 satır okuyor, çünkü her koşum birikmiş tarihi baştan işliyor; akış her satırı bir kez, 2400 satır okuyor. Buna karşılık akış hattın aşamasını 5’ten 8’e çıkarıyor — ilerleme damgası, görülen anahtarın ve son endeksin saklanması — ve en çok 2600 girdi taşıyor: 2400 anahtar ile 200 abone endeksi. Toplu kipin taşıdığı durum sıfırdır.

Yakalama Dönemi

Aşağıdaki blok altı bozulma sınıfını iki kipte koşturur ve her birinin hangi dönemde, hangi denetimle yakalandığını yazar.

sonuc = {}
for sinif in BOZULMA:
    veri, n = teslim_et(ham, sinif)
    sonuc[sinif] = (n, toplu(veri, TABAN), akis(veri, TABAN))


def etiket(r):
    if r["yakalayan"] is None:
        return "yakalanmiyor"
    return f"{r['yakalayan']}/{r['donem']}/{DONEM.index(r['donem']) - DONEM.index('2026-07')}"


def kirli(h):
    return sum(1 for ad in set(h) | set(TABAN) if h.get(ad) != TABAN.get(ad))


print("bozulma        bozulan  toplu: denetim/donem/gecikme  akis: denetim/donem/gecikme")
for s, (n, t, a) in sonuc.items():
    print(f"{s:<14} {n:>7}  {etiket(t):<28} {etiket(a)}")

print("\nbozulma         kirli hucre/55   sunum degisti   akista")
print("                toplu    akis    toplu   akis    dusen satir")
for s, (n, t, a) in sonuc.items():
    e = ["hayir", "evet"]
    print(f"{s:<14} {kirli(t['hucre']):>7} {kirli(a['hucre']):>7} "
          f"{e[a5_sun(t['hucre'])[0] != a5_sun(TABAN)[0]]:>8}"
          f"{e[a5_sun(a['hucre'])[0] != a5_sun(TABAN)[0]]:>7} {a['gec_dusen']:>10}")

bozulma        bozulan  toplu: denetim/donem/gecikme  akis: denetim/donem/gecikme
gec_gelen           40  dagilim/2026-08/1            dagilim/2026-07/0
yinelenen           60  tekillik/2026-08/1           tekillik/2026-07/0
sema_degisti       200  sema/2026-08/1               sema/2026-07/0
birim_degisti      200  dagilim/2026-08/1            dagilim/2026-07/0
eksik_bolum         46  yakalanmiyor                 bolum/2026-07/0
kayan_tip          200  tip/2026-08/1                tip/2026-07/0

bozulma         kirli hucre/55   sunum degisti   akista
                toplu    akis    toplu   akis    dusen satir
gec_gelen            0      10    hayir  hayir         40
yinelenen            0       0    hayir  hayir          0
sema_degisti        10      10     evet   evet          0
birim_degisti       10      10    hayir  hayir          0
eksik_bolum          2       2     evet   evet          0
kayan_tip            0       0    hayir  hayir          0

Bir Dönem Erken Bakmak

Birinci tablo takası tek satırda veriyor: yakalanan beş sınıfın hepsinde toplu kip 2026-08, akış kipi 2026-07 yazıyor. Dönem gecikmesi toplu kipte 1, akışta 0. Aşama gecikmeleri ise birebir aynı: şema, tür ve tekillik aşama 0’da, dağılım aşama 4’te yakalıyor. Kip, bozulmanın hangi aşamada yakalandığını değiştirmiyor; yalnız hangi dönemde bakıldığını değiştiriyor. Tazelik burada bir sunum özelliği değil, bir denetim özelliğidir.

Bir dönemin ne demek olduğu çıktıda görünür: toplu kipte bozulmuş dönem, yakalandığı ana kadar bir kez sunulmuştur — 2026-07 kapanmış, özet yazılmış, sunum değeri okunmuştur. Bu fark bir tolerans ayarıyla kapatılamaz, çünkü toplu kipin gecikmesi denetimden değil çizelgeden gelir.

İkinci tablo kipten bağımsız olanı gösteriyor. Yinelenen satır, tür kayması ve birim değişimi sunulan beş değeri iki kipte de değiştirmiyor; ilk ikisi kirli hücre bile bırakmıyor, birim değişimi on hücreyi kirletiyor ama sunum kusursuz görünüyor. Bu üçünü gören tek şey denetimdir.

Pencerenin Denetimi Düzeltmesi

Beşinci satır tablonun en beklenmedik sonucudur. Eksik bölüm toplu kipte hiç yakalanmıyor, akışta aşama 0’da yakalanıyor — üstelik iki kipte de aynı denetim kodu koşuyor. Bölüm denetimi bölge kümesine bakar ve dönemi görmez; toplu koşumda önüne on iki dönemin tamamı gelir, kuzey öbür on bir dönemde göründüğü için küme tamdır ve denetim yeşil kalır. Akış koşumunda aynı denetimin önüne gelen küme tek bir dönemin penceresidir; orada kuzey yoktur ve denetim ateşler.

Bir denetimin doğruluğu yalnız kodundan değil önüne konan kümeden de gelir. Bu, akış kipinin bir üstünlüğü gibi görünse de bir yan etkidir; toplu kipte aynı sonucu almanın yolu kipi değiştirmek değil denetimi dönem duyarlı yazmaktır.

Durumun Bedeli

İkinci tablonun ilk satırı takası tersine çeviriyor. Geç gelen kırk satır bir dönem sonra geldiğinde toplu kip birikmiş tarihi baştan işler ve satırları geri doldurur: 55 özet hücresinin tamamı doğru kalır. Akış aynı satırları ilerleme damgasınde düşürür ve pencere kapandığı için bir daha bakmaz. Sonuç 10 kirli özet hücresidir, yani hücrelerin 0,1818’i.

Mekanizma taşınan durumdur. Kırk abonenin 2026-07 okuması hiç işlenmediği için son endeks durumu 2026-06’da kalır; 2026-08 geldiğinde tüketim iki dönemlik farktan hesaplanır. Toplu kipte böyle bir satır düşer, çünkü önceki dönem tabloda yoktur; akışta yanlış hesaplanır, çünkü durumda eski bir değer vardır. Eksik satır sayılabilir bir kayıptır, yanlış değer değildir.

Son sütun bu bölümün asıl uyarısıdır. On kirli hücreye rağmen sunulan beş değer iki kipte de birebir aynıdır: temmuzda eksilen ağustosta fazlalanıyor, on iki dönemin ortalaması oynamıyor. Akış kipi bu bozulmayı 2026-07’de yakalamıştı; yakalamasaydı hiçbir çıktıya bakarak bulunamayacaktı.

Özet

• Toplu kip dönem kapandıktan sonra birikmiş tarihi baştan işler, akış satırı geldiği dönemde işler; ikisi de aynı beş aşamayı ve aynı beş denetimi kullanır, tek farkları penceredir.
• Yakalanan beş sınıfın hepsinde toplu kipin dönem gecikmesi 1, akışınki 0’dır; aşama gecikmeleri birebir aynıdır. Kip, bozulmanın hangi aşamada değil hangi dönemde görüldüğünü değiştirir.
• Akışın bedeli aşama sayısının 5’ten 8’e çıkması ve 2600 girdi taşınmasıdır; karşılığında okunan satır 13200’den 2400’e iner.
• Dönem penceresi kusurlu bölüm denetimini düzeltir: aynı kod toplu kipte eksik bölümü hiç yakalamaz, akışta aşama 0’da yakalar.
• Geç gelen satırda takas tersine döner: toplu kip 55 hücrenin hepsini doğru bırakır, akış 40 satırı düşürüp 10 hücreyi kirletir; buna rağmen sunulan beş değer iki kipte de aynıdır.

Sonraki Adım

Beş derste hat kuruldu ve sayıldı: aşamaları belirlendi, her aşamanın düşürdüğü satır yazıldı, rollerin nerede durduğu ayrıldı, veri kaynak sınıflarına ayrıştırıldı ve hangi çizelgeyle koşacağı iki kipte ölçüldü. Bir şey hiç sorulmadı: hattın taşıdığı satırın doğru olup olmadığı. Sayılan her şey satırın nereden geldiği, nereye gittiği ve ne zaman bakıldığıdır; satırın kendisine bakan tek bir ölçüt yazılmadı. Denetimler hazır bulundu, ama nereden geldikleri tanımlanmadı. Sonraki konu bunu yapar: kalite boyutlarını ölçülebilir denetimlere çevirir, her denetimin yakalama gecikmesini sayar ve denetimin kendi kusurunu ölçer.