Akış İşleme

Önceki üç derste veri hatta alındı: hangi yöntemle alınacağı, kaynağın kısıtı altında nasıl alınacağı ve mesajların hangi bölümden geleceği yazıldı. Alınan satırın hatta eksiksiz varıp varmadığı sorulmadı. Bu ders kaybın ilk biçimini ölçer ve o biçim taşıma hatası değildir: kapanmış bir pencereye geç gelen kayıttır.

Toplu işlemede pencere sorusu yoktur, çünkü dönem kapandıktan sonra bakılır. Akış işlemede kaynak bitmez; bir toplam almak için akışın bir yerinden kesilmesi gerekir. Kesme kararı üç şeyi birden belirler: hangi kaydın hesaba girdiğini, sonucun hangi dönemde yazıldığını ve koşumun ne kadar durum taşıdığını. Bu ders o üç sayıyı yan yana koyar.

• VA41. Kod kurgudur; gerçek bir akış işleyici, mesaj sistemi ya da orkestratör çağrılmaz. Kaynak M29’un kurgu sayaç okumalarıdır: 200 abone, 12 dönem, 2400 satır, tohum 20260218.
• VA42. Toplu ile akış kipinin karşılaştırması tekrarlanmaz; Veri Mühendisliğine Giriş kursundaki Toplu ve Akış İşleme dersinde ölçüldü. Buradaki taban çizgisi o dersin akış kipidir; eklenen tek şey pencerenin ne kadar açık kaldığıdır.
• VA43. Pencere bir dönemdir. İlerleme damgası (watermark) kapanmış sayılan en son dönemi gösterir; tolerans, pencerenin kapanmadan önce kaç dönem daha açık kaldığıdır.
• VA44. Geç gelen kayıt “hiç gelmedi” değil bir dönem sonra geldi sayılır: 2026-07 döneminin kırk okuması 2026-08’de geliyor. Bozulma sınıfı ortak tanımın gec_gelen sınıfıdır.
• VA45. Taşınan durum iki parçadır: açık pencerelerde bekleyen satırlar ve abone başına son endeks. Ölçülen büyüklük ikisinin toplamının koşum boyunca gördüğü en büyük değerdir.
• VA46. Akış kipinde tüketim durumdan hesaplanır: satırın tüketimi, aynı abonenin durumda duran son endeksinden çıkarılır. Durumda karşılığı olmayan satır hesaplanamaz.
• VA47. Gecikme dönem olarak yazılır, ham süre yazılmaz. Özet hücresi oranının adımı 1/55, yani 0,0182; sunum üzerinde oran basılmaz, değerin kendisi yazılır.
• VA48. Denetim noktası, taşınan durumun koşum dışında saklanmasıdır. Yordamı bu derste yazılmaz; yalnız kaybının bedeli sayılır.

Pencere, Tolerans ve Durum

Aşağıdaki blok ortak tanımı kurar ve üzerine tek bir koşucu ekler. akis_pencere her dönemde o dönemde gelen satırları alır, ilerleme damgasının kapattığı pencereye ait olanları düşürür, kalanları açık pencerelerde bekletir ve sırası gelen pencereyi kapatıp özet hücrelerini yazar.

# KURGUDUR. Gercek bir akis isleyici , mesaj sistemi 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


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):                              # toplu kip: donem farkindan tuketim
    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
        cikti.append(dict(k, m3=int(k["endeks"]) - int(onceki["endeks"])))
    return cikti, dusen


def a4_ozet(kayit):
    top = {}
    for k in kayit:
        ad = (k["bolge"], k["donem"])
        top[ad] = top.get(ad, 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


def gelisli(kayit, gecikme):
    """Her satira bir gelis donemi verir. gec_gelen sinifi: 2026-07 doneminin
    A001..A040 okumalari `gecikme` donem sonra geliyor."""
    veri = []
    for k in kayit:
        i = DONEM.index(k["donem"])
        g = gecikme if k["donem"] == "2026-07" and k["abone"] <= "A040" else 0
        veri.append(dict(k, gelis=DONEM[i + g]))
    return veri


def akis_pencere(veri, tolerans, sifirla=None):
    """Donem penceresi tolerans kadar donem daha acik kalir , sonra kapanir.
    sifirla: bu donemin basinda tasinan durum kaybolur (denetim noktasi yok)."""
    tampon, son_endeks, hucre, yazim = {}, {}, {}, {}
    dusen, durum, hesaplanamayan = 0, 0, 0

    def kapat(p, i):
        nonlocal hesaplanamayan
        v, _ = a1_ayikla(tampon.pop(p, []))
        v, _ = a2_tekille(v)
        c = []
        for k in v:
            if k["abone"] in son_endeks:
                c.append(dict(k, m3=k["endeks"] - son_endeks[k["abone"]]))
            elif DONEM.index(p) > 0:
                hesaplanamayan += 1
        for k in v:
            son_endeks[k["abone"]] = k["endeks"]    # durum: abone basina son endeks
        o, _ = a4_ozet(c)
        if o:
            hucre.update(o)
            yazim[p] = DONEM[i]

    for i, d in enumerate(DONEM):
        if d == sifirla:
            son_endeks.clear()
        for k in veri:
            if k["gelis"] != d:
                continue
            if DONEM.index(k["donem"]) < i - tolerans:
                dusen += 1                          # ilerleme damgasi pencereyi kapatti
                continue
            tampon.setdefault(k["donem"], []).append(k)
        durum = max(durum, sum(len(v) for v in tampon.values()) + len(son_endeks))
        if i - tolerans >= 0:
            kapat(DONEM[i - tolerans], i)
    for j in range(max(0, len(DONEM) - tolerans), len(DONEM)):
        kapat(DONEM[j], len(DONEM) - 1)
    return {"dusen": dusen, "durum": durum, "hucre": hucre, "yazim": yazim,
            "hesaplanamayan": hesaplanamayan}


ham = kaynak_uret()
v, _ = a1_ayikla(ham)
v, _ = a2_tekille(v)
v, _ = a3_tuketim(v)
TABAN, _ = a4_ozet(v)


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


def fark(h):
    s, t = a5_sun(h)[0], a5_sun(TABAN)[0]
    d = {b: round(s[b] - t[b], 2) for b in t if round(s.get(b, 0) - t[b], 2)}
    return d if d else "bos"


print("taban: tuketim satiri", len(v), ", ozet hucresi", len(TABAN))
print("taban sunum", a5_sun(TABAN)[0])
print()
print("40 satirin gecikmesi  tolerans  dusen  durum  kirli/55  07 yazildi  sunum farki")
for gc in (0, 1, 2):
    for t in (0, 1, 2):
        r = akis_pencere(gelisli(ham, gc), t)
        print(f"{gc:>15} donem {t:>9} {r['dusen']:>6} {r['durum']:>6}"
              f" {kirli(r['hucre']):>9}  {r['yazim']['2026-07']:>9}  {fark(r['hucre'])}")

taban: tuketim satiri 2200 , ozet hucresi 55
taban sunum {'merkez': 1321.45, 'kuzey': 1212.55, 'dogu': 881.36, 'bati': 833.09, 'guney': 1065.55}

40 satirin gecikmesi  tolerans  dusen  durum  kirli/55  07 yazildi  sunum farki
              0 donem         0      0    400         0    2026-07  bos
              0 donem         1      0    600         0    2026-08  bos
              0 donem         2      0    800         0    2026-09  bos
              1 donem         0     40    400        10    2026-07  bos
              1 donem         1      0    600         0    2026-08  bos
              1 donem         2      0    800         0    2026-09  bos
              2 donem         0     40    400        10    2026-07  bos
              2 donem         1     40    600        10    2026-08  bos
              2 donem         2      0    800         0    2026-09  bos

Pencere Kararının Değiştirdiği Sonuç

İlk üç satır kararın bedelini kayıp olmadan gösterir. Hiçbir kayıt geç gelmese bile tolerans 1 taşınan durumu 400 girdiden 600’e, tolerans 2 800’e çıkarıyor ve 2026-07 hücresini bir ya da iki dönem geç yazıyor. Bu bedel her koşumda ödenir; geç kayıt gelip gelmemesi onu değiştirmez. Bir pencereyi açık tutmak, gelmemiş bir kayıt için yer ayırmaktır ve o kayıt gelmese de yer ayrılmıştır.

Ortadaki üç satır karşılığını veriyor. Kırk okuma bir dönem geç geldiğinde tolerans 0 onları düşürüyor: pencere kapanmıştır ve ilerleme damgası geri alınmaz. Sonuç 55 özet hücresinin 10’unun kirlenmesidir — 2026-07 hücreleri eksik kalıyor, 2026-08 hücreleri ise şişiyor, çünkü o kırk abonenin tüketimi iki dönemlik farktan hesaplanıyor. Bir pencerede düşen kayıt yalnız kendi hücresini değil, ardından gelen dönemin hücresini de bozuyor; sebebi tüketimin durumdan hesaplanmasıdır. Tolerans 1’de aynı kayıtlar hâlâ açık pencereyi buluyor: düşen satır 0, kirli hücre 0.

Son üç satır toleransın ne olduğunu tanımlıyor: bir tahmindir. Aynı kırk kayıt iki dönem geciktiğinde tolerans 1 onları da düşürüyor ve sonuç tolerans 0’la birebir aynı oluyor — 40 düşen satır, 10 kirli hücre. Tolerans, gecikmenin ne kadar olacağına yapılan bir bahistir; bahis tutmadığında ödenen bedel geri alınmaz, üstelik doğru cevap alınamadan ödenmiştir. Tolerans 2 bu kez kazanıyor, ama tolerans 2 satırlarının hepsinde durum 800 girdiye çıkmıştır.

Son sütun dersin uyarısıdır. Dokuz satırın hepsinde sunum farkı boştur. On kirli hücreye rağmen sunulan beş bölge değeri tabanla birebir aynıdır, çünkü 2026-07’de eksilen 2026-08’de fazlalanıyor ve on bir dönemin ortalaması oynamıyor. Pencere kararının yanlış olduğunu gösteren tek kanıt hücre düzeyindeki karşılaştırmadır; çıktıya bakan hiçbir gözlem bu kararı sorgulayamaz. Doğru cevabı veren satırların da bir bedeli var ve o bedel aynı tabloda duruyor: 2026-07 hücresi toleransla birlikte bir ya da iki dönem geç yazılıyor. Pencereyi açık tutmak sonucu geciktirmektir.

Durumun İki Bedeli

Toleransı büyütmek doğruluğu bir yere kadar satın alır ve bunun sınırı vardır. Aşağıdaki bloğun ilk tablosu sınırı gösteriyor: pencere hiç kapanmazsa ne oluyor. İkinci tablo durumun asıl niteliğini ölçüyor — taşınan durum bir bellek büyüklüğü değil, hattın doğru cevap üretebilmesinin koşuludur. Akış kipinde tüketim durumdaki son endeksten hesaplanır; durum kaybolursa satır hesaplanamaz. İkinci tablo tolerans 1 kurulumunu iki kez koşturuyor: durum koşumlar arasında saklanmışken ve 2026-08 başında kaybolmuşken.

print("pencere            tolerans  durum  dusen  kirli/55  07 yazildi")
for ad, t in (("her donem kapanir", 1), ("hic kapanmaz", 11)):
    r = akis_pencere(gelisli(ham, 1), t)
    print(f"{ad:<18} {t:>8} {r['durum']:>6} {r['dusen']:>6} {kirli(r['hucre']):>9}"
          f"  {r['yazim']['2026-07']:>9}")

print()
print("durum         hesaplanamayan  hucre  kirli/55")
for ad, sf in (("saklandi", None), ("kayboldu", "2026-08")):
    r = akis_pencere(gelisli(ham, 1), 1, sifirla=sf)
    print(f"{ad:<14} {r['hesaplanamayan']:>13} {len(r['hucre']):>6} {kirli(r['hucre']):>9}")
    print("   sunum", a5_sun(r["hucre"])[0])

pencere            tolerans  durum  dusen  kirli/55  07 yazildi
her donem kapanir         1    600      0         0    2026-08
hic kapanmaz             11   2400      0         0    2026-12

durum         hesaplanamayan  hucre  kirli/55
saklandi                   0     55         0
   sunum {'merkez': 1321.45, 'kuzey': 1212.55, 'dogu': 881.36, 'bati': 833.09, 'guney': 1065.55}
kayboldu                 200     50         5
   sunum {'merkez': 1325.6, 'kuzey': 1208.7, 'dogu': 873.7, 'bati': 835.2, 'guney': 1068.2}

İlk tablo toleransın üst sınırını veriyor. Pencere hiç kapanmazsa hiçbir kayıt düşmez ve hiçbir hücre kirlenmez, ama taşınan durum 2400 girdiye, yani kaynağın tamamına çıkar ve 2026-07 hücresi son dönemde, 2026-12’de yazılır. Bu satır artık akış işleme değildir; bütün veriyi biriktirip sonda bir kez işleyen bir toplu koşumdur. Pencere, akış işlemeyi akış yapan şeydir: kapanmayan bir pencere kipin bütün tazelik kazancını geri verir.

İkinci tablo durumun kırılganlığını ölçüyor. Durum kaybolduğunda 2026-08 penceresindeki 200 satırın hiçbiri hesaplanamıyor, çünkü hiçbir abonenin önceki endeksi elde yoktur. Özet 55 hücreden 50’ye iniyor: bir dönem hiç yazılmıyor. Bunun sunuma etkisi bu kez görünür — beş bölge değerinin hepsi kayıyor, üstelik iki yönde birden, merkez yükselirken kuzey düşüyor. Sebebi bölge ortalamasının artık on bir değil on dönem üzerinden alınmasıdır; düşen dönemin ortalamadan farkı işareti belirler.

İki bedel yan yana konduğunda akış kipinin tuhaf bir dengesi görünür. Geç gelen kayıt sunumu hiç oynatmadan on hücreyi bozdu; kaybolan durum sunumu oynattı ama beş hücreyi eksik bırakarak, yani sayılabilir bir iz düşürerek. Eksik hücre bir denetim için sorulabilir bir sorudur, yanlış hücre değildir. Durumun koşum dışında saklanması — denetim noktası (checkpoint) — bu ikinci bedeli ortadan kaldırır, birincisini, yani pencere kararını kaldırmaz.

İki bedelin ortak yanı, ikisinin de pencerenin bir sonucu olmasıdır. Toplu koşumda böyle bir soru yoktur: her koşum kaynağın tamamını yeniden okur, geç gelen satır bir sonraki koşumda kendiliğinden hesaba girer ve taşınacak bir durum kalmaz. Akış kipi okunan satırı düşürdüğü için bu bedava düzeltmeyi kaybeder; onun yerine iki karar bırakır — pencere ne kadar açık kalacak ve durum nerede saklanacak. Bu iki kararın hiçbiri veriye bakılarak verilemez; gecikmenin dağılımına ve koşumun kesilme sıklığına bakılarak verilir. Bir hattın sayısı bu yüzden yalnız kodundan okunmaz.

Özet

• Akış işlemede pencere kararı üç şeyi birden belirler: hangi kaydın hesaba girdiğini, sonucun hangi dönemde yazıldığını ve koşumun ne kadar durum taşıdığını.
• Tolerans 0’da bir dönem geciken 40 kayıt düşüyor ve 55 özet hücresinin 10’u kirleniyor; tolerans 1’de düşen satır ve kirli hücre 0’a iniyor, bedeli 2026-07 hücresinin bir dönem geç yazılmasıdır.
• Tolerans gecikmeye yapılan bir bahistir: aynı kayıtlar iki dönem geciktiğinde tolerans 1 sonucu tolerans 0 ile birebir aynı oluyor, üstelik 600 girdilik durum ödenmiş olarak.
• Toleransın bedeli koşuma bağlı değildir: taşınan durum geç kayıt olsun olmasın 400, 600 ve 800 girdidir; pencere hiç kapanmazsa durum 2400’e, yani kaynağın tamamına çıkar.
• Dokuz kurulumun hiçbirinde sunulan beş değer değişmiyor; on kirli hücrenin tek tanığı hücre düzeyindeki karşılaştırmadır.
• Denetim noktası olmadan kaybedilen durum 200 satırı hesaplanamaz kılıyor, özeti 55 hücreden 50’ye düşürüyor ve sunumu her bölgede oynatıyor.

Sonraki Adım

Bu derste kaybın kaynağı hattın kendi kararıydı: pencere kapandıktan sonra gelen kayıt bilerek düşürüldü, kaç tanesinin düştüğü sayıldı ve karar değiştirildiğinde sayı sıfıra indi. Kayıt hatta ulaşmıştı; yalnız geç ulaşmıştı. Bir kayıt hiç ulaşmazsa ya da iki kez ulaşırsa sayının ne olacağı sorulmadı. Alma ile hattın arasındaki teslim, ders boyunca kusursuz varsayıldı: gönderilen her satır tam olarak bir kez varmış gibi işlendi. Sonraki ders bu varsayımı kaldırır ve üç teslim kipini yan yana koyar: satırın kaybolabildiği, iki kez varabildiği ve anahtarla yazıldığı kurulumlar. Ölçülecek soru şudur: kayıp ile yinelenmeden hangisi sunulan sayıyı bozar, hangisi bir denetime takılır ve ikisinin onarılabilirliği neden aynı değildir.