--- title: 'Akış İşleme' source: 'https://academia.sh/tr/kurslar/veri-hatlari/akis-isleme' course: 'Veri Hatları ve Orkestrasyon' language: tr updated: '2026-08-17T18:08:49+00:00' license: 'CC BY-SA 4.0' --- # Akış İşleme Pencere kararının ve durum taşımanın sayıya çevrilmesi: dönem penceresi kapandığı anda kapatıldığında bir dönem geciken 40 kayıt düşüyor ve 55 özet hücresinin 10'u kirleniyor, pencere bir dönem daha açık tutulduğunda düşen satır 0'a iniyor ama 2026-07 hücresi 2026-08'de yazılıyor ve taşınan durum 400 girdiden 600'e çıkıyor. Tolerans gecikmeye yapılan bir bahistir: aynı kayıtlar iki dönem geciktiğinde tolerans 1 sonucu tolerans 0 ile birebir aynı oluyor. Dokuz kurulumun hiçbirinde sunulan beş değer değişmiyor, yalnız kirli hücre sayısı değişiyor. Pencere hiç kapanmazsa durum 2400'e çıkıyor ve kip toplu işlemeye dönüyor; denetim noktası olmadan kaybedilen durumda 200 satır hesaplanamıyor, özet 50 hücreye düşüyor ve sunum her bölgede kayıyor. Ö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. ```python # 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. ```python 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.