Dağıtık Veri Kümeleri

Önceki üç derste iş bölündü ve çalıştırıldı: satırlar bir anahtara göre işçilere dağıtıldı, blokların nerede durduğu okumanın nereye gideceğini belirledi, eşleme ile indirgeme birbirinden ayrıldı. Üçünün de ortak bir varsayımı vardı ve hiçbir yerde yazılmadı: sürücünün yazdığı her hesap koşar. Bir dönüşüm tanımlandığı anda küme onu işletir, sonucunu bir yere yazar ve sonrakini bekler.

Bu ders o varsayımı kaldırır ve tek bir soru sorar: yazılan hesabın ne kadarı sonuçta gerçekten kullanılıyor. Yanıt iki sayıyla verilir — hiç açılmayan bölüm ve hiç okunmayan satır. Bu iki sayıyı mümkün kılan yapı dağıtık veri kümesidir (distributed dataset): bölümlerden ve henüz koşmamış bir plandan oluşur.

• DH41. Kaynak, hat ve küme kurgudur; gerçek bir dağıtık işleme çatısı ya da kaynak yöneticisi çağrılmaz. Tohum 20260218, ölçü koşumdan bağımsızdır.
• DH42. Süre bu modelin kendi birimidir, ham milisaniye değildir: en yüklü işçinin okuduğu satır sayısı artı görev sabiti.
• DH43. Bölüm verinin birimidir, işçi hesabın birimidir. Bölüm anahtarı donem; tüketim aşaması ilk dönemi düşürdüğü için 11 bölüm kalır, her biri 200 satır.
• DH44. Bölümler işçilere sırayla verilir (0, 1, 2, 3, 0, …); üç işçi üçer, biri iki bölüm tutar.
• DH45. Dönüşüm yeni bir küme tanımlar ve sonuç döndürmez; eylem sürücüye bir sonuç döndürür ve hesabı başlatan tek şeydir.
• DH46. Sayılan birimler: açılan bölüm, okunan satır, yazılan ara satır, süre. Bu derste karıştırma yoktur, bu yüzden toplam iş okunan satıra eşittir.
• DH47. Taban çizgisi istekli değerlendirmedir: her dönüşüm tanımlandığı anda koşar ve ara sonucunu yazar.
• DH48. Çözünürlük: bölüm 11, bölüm başına satır 200, toplam 2200. Sunum üzerinde oran basılmaz, değerin kendisi yazılır.

Dönüşüm ile Eylem

Dağıtık bir kümede iki tür çağrı vardır ve aralarındaki fark bu dersin tamamıdır. Dönüşüm (transformation) mevcut kümeden yeni bir küme tanımlar: süzme, eşleme, yeniden bölümleme. Geri döndürdüğü şey veri değil, tariftir. Eylem (action) ise sürücüye bir sonuç döndürür: sayma, ilk satırları getirme, toplama. Yalnız eylem hesabı başlatır.

Tembel değerlendirme (lazy evaluation) bu ayrımın sonucudur. Dönüşümler bir plana yazılır, ilk eylem gelene kadar hiçbir bölüm açılmaz. Eylem geldiğinde plan bir bütün olarak bilinmektedir ve küme yalnız o sonucun gerektirdiği bölümleri koşturur. Karşıtı istekli değerlendirmedir: her dönüşüm yazıldığı satırda koşar, ara sonucunu yazar ve bir sonraki dönüşüm onu yeniden okur.

İlk blok kurgu kaynağı ve hattın ilk üç aşamasını kurar. Buradan çıkan 2200 satır kursun tek somut eksenidir.

# KURGUDUR. Sayac okumalari, hattin ilk uc asamasi ve kurgu kume ders icinde
# tanimlanir; gercek bir dagitik isleme catisi ya da kaynak yoneticisi cagrilmaz.
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():
    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):
    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


ISCI = 4
GOREV_SABIT = 50       # her isciye is vermenin sabit bedeli , bir secimdir


def satirlar():
    v, _ = a1_ayikla(kaynak_uret())
    v, _ = a2_tekille(v)
    v, _ = a3_tuketim(v)
    return v

Bölüm Verinin Birimi, İşçi Hesabın Birimi

Bölüm sayısı ile işçi sayısı aynı şey değildir ve bu dersin ölçtüğü kazanç tam olarak bu ayrımdan doğar. Bölüm verinin bölünmüş bir parçasıdır ve tek başına okunabilir. İşçi o parçaları okuyan hesap birimidir ve birden çok bölüm tutabilir. Burada 11 bölüm dört işçiye dağılır: üç işçi üçer, biri iki bölüm alır. Bölüm işçiden çok olduğu için bir eylem bölümlerin yalnız bir kısmını açabilir, ve açılmayan bölümün satırları hiç okunmaz.

Bölümün depolama tarafındaki kullanımı — hangi bölümün dosya düzenine bakılarak hiç okunmayacağına karar verilmesi — Veri Depolama Mimarileri kursunda ölçüldü ve burada tekrarlanmaz. Buradaki eleme hesap tarafındadır: bölümü eleyen şey dosya düzeni değil, eylemin ne istediğidir.

S = satirlar()
BOLUM = {d: [k for k in S if k["donem"] == d] for d in DONEM}
BOLUM = {d: v for d, v in BOLUM.items() if v}
AD = sorted(BOLUM)
SAHIP = {a: i % ISCI for i, a in enumerate(AD)}
SAYAC = {}


def sifirla():
    SAYAC.update({"bolum": 0, "okunan": 0, "ara": 0,
                  "isci": {i: 0 for i in range(ISCI)}})


def oku(ad, veri):
    SAYAC["bolum"] += 1
    SAYAC["okunan"] += len(veri)
    SAYAC["isci"][SAHIP[ad]] += len(veri)
    return veri


class Tembel:
    """Donusum plana yazilir , kosmaz. Hesap ancak bir eylemle baslar."""

    def __init__(self, bolum, plan=()):
        self.bolum, self.plan = bolum, tuple(plan)

    def sec(self, f):
        return Tembel(self.bolum, self.plan + (("sec", f),))

    def esle(self, f):
        return Tembel(self.bolum, self.plan + (("esle", f),))

    def _kos(self, ad):
        cikti = []
        for k in oku(ad, self.bolum[ad]):
            for tur, f in self.plan:
                if tur == "sec" and not f(k):
                    k = None
                    break
                if tur == "esle":
                    k = f(k)
            if k is not None:
                cikti.append(k)
        return cikti

    def say(self):
        return sum(len(self._kos(a)) for a in AD)

    def ilk(self, n):
        c = []
        for a in AD:
            c += self._kos(a)
            if len(c) >= n:
                break
        return c[:n]


class Istekli(Tembel):
    """Her donusum tanimlandigi anda kosar ve ara sonucunu yazar."""

    def _uygula(self, f, sec):
        yeni = {}
        for a in AD:
            v = [k for k in oku(a, self.bolum[a]) if f(k)] if sec \
                else [f(k) for k in oku(a, self.bolum[a])]
            SAYAC["ara"] += len(v)
            yeni[a] = v
        return Istekli(yeni)

    def sec(self, f):
        return self._uygula(f, True)

    def esle(self, f):
        return self._uygula(f, False)

    def _kos(self, ad):
        return self.bolum[ad]


print("bolum:", len(BOLUM), " bolum basina satir:", sorted(set(len(v) for v in BOLUM.values())))
print("isci basina bolum:", [sum(1 for a in AD if SAHIP[a] == i) for i in range(ISCI)])

bolum: 11  bolum basina satir: [200]
isci basina bolum: [3, 3, 3, 2]

İki sınıf da aynı arayüzü sunar; ayrım yalnız hesabın ne zaman koştuğudur. Tembel sınıfında sec yeni bir plan döndürür, Istekli sınıfında aynı çağrı bütün bölümleri okur.

Tembelliğin Elediği Hesap

Ölçülen iş sürücünün yazdığı sıradan bir betiktir: mesken abonelerinin okumaları süzülür, satır dört alana indirgenir ve 30 m3 üstündekiler seçilir. Betik ayrıca bir isyeri kümesi de tanımlar ve hiçbir eylem onu istemez — bu, elde tutulan hesabın ölçüsüdür. Dört kurulum aynı betiği koşturur ve eylem değişir: ilk(5) beş satır ister, ilk(100) yüz satır, say bütün sonucu.

ALAN = ("abone", "donem", "bolge", "m3")


def betik(K):
    """Surucunun yazdigi is: mesken abonelerinin 30 m3 ustundeki okumalari."""
    mesken = K.sec(lambda k: k["tip"] == "mesken")
    K.sec(lambda k: k["tip"] == "isyeri")          # hicbir eylem bunu istemiyor
    return mesken.esle(lambda k: {a: k[a] for a in ALAN}).sec(lambda k: k["m3"] > 30)


def olc(kip, eylem):
    sifirla()
    y = betik(Tembel(BOLUM) if kip == "tembel" else Istekli(BOLUM))
    n = y.say() if eylem == "say" else len(y.ilk(int(eylem)))
    return (SAYAC["bolum"], SAYAC["okunan"], SAYAC["ara"],
            max(SAYAC["isci"].values()) + GOREV_SABIT, n)


print(f"{'kip':<9}{'eylem':>7}{'acilan bolum':>14}{'okunan':>8}"
      f"{'ara satir':>11}{'sure':>6}{'sonuc':>7}")
for kip, eylem in (("istekli", "5"), ("tembel", "5"), ("tembel", "100"),
                   ("istekli", "say"), ("tembel", "say")):
    b, o, ara, sure, n = olc(kip, eylem)
    print(f"{kip:<9}{eylem:>7}{b:>14}{o:>8}{ara:>11}{sure:>6}{n:>7}")

kip        eylem  acilan bolum  okunan  ara satir  sure  sonuc
istekli        5            44    6556       3727  1838      5
tembel         5             1     200          0   250      5
tembel       100             3     600          0   250    100
istekli      say            44    6556       3727  1838    449
tembel       say            11    2200          0   650    449

İlk iki satır dersin bulgusudur ve sonuç sütunu ikisinde de aynıdır. İstekli kurulum beş satırlık sonuç için 44 bölüm açıyor, 6556 satır okuyor, 3727 ara satır yazıyor; süre 1838. Aynı beş satırı tembel kurulum tek bölüm açarak, 200 satır okuyarak ve hiç ara satır yazmadan veriyor; süre 250. Geriye kalan 10 bölüm hiç açılmıyor ve 2000 satır hiç okunmuyor.

İstekli kurulumun 6556 satırı iki yerden geliyor. Birincisi: kaynağın 2200 satırı iki kez okunuyor, çünkü mesken ve isyeri süzmeleri ayrı ayrı bütün bölümlerden geçiyor. İkincisi: isyeri dalı 2200 satır okuyup 1122 ara satır yazıyor ve sonuca hiç girmiyor. Tembel kurulumda o dal bir plandan ibarettir ve plan koşmaz.

Üçüncü satır kursun kuralını tek başına taşıyor. ilk(100) üç bölüm açıyor, okunan satır 200’den 600’e çıkıyor — toplam iş üçe katlanıyor — ama süre 250’de kalıyor. Nedeni DH44’tür: sıralı dağıtım yüzünden ilk üç bölüm üç ayrı işçidedir ve üçü aynı anda koşar. En yüklü işçi yine 200 satır okur. Hangisinin ölçüldüğü yazılmayan bir hızlanma iddiası burada eksik kalır: aynı iki kurulum arasında toplam iş üç kat, süre bir kat.

Son iki satır tembelliğin sınırını çiziyor. say bütün sonucu istediği için tembel kurulum da 11 bölümün hepsini açıyor ve 2200 satır okuyor. Fark artık atlanan bölümden değil, kullanılmayan daldan ve yazılmayan 3727 ara satırdan geliyor: 6556’ya karşı 2200 satır, 1838’e karşı 650 süre.

Bölüm Boyutu Tembelliğin Çözünürlüğüdür

Tembellik bir bölümü atlayabilir, bölümün içinden bir satırı atlayamaz. Öyleyse atlanabilen en küçük parça bölümün kendisidir ve bölüm sayısı bir karardır. Aynı 2200 satır dört, on bir ve kırk dört bölüme konur; ölçülen iş ve eylem değişmez.

KAYIT = (BOLUM, AD, SAHIP)


def yeniden_bolumle(sayi):
    """Bolum sayisi bir karardir: tembelligin atlayabildigi en kucuk birim bir bolumdur."""
    global BOLUM, AD, SAHIP
    BOLUM = {i: S[i * len(S) // sayi:(i + 1) * len(S) // sayi] for i in range(sayi)}
    AD, SAHIP = sorted(BOLUM), {i: i % ISCI for i in range(sayi)}


print(f"{'bolum':>6}{'satir/bolum':>13}{'ilk(5) acilan':>15}{'ilk(5) okunan':>15}"
      f"{'ilk(5) sure':>13}{'say sure':>10}")
for sayi in (4, 11, 44):
    yeniden_bolumle(sayi)
    b, o, _, sure, _ = olc("tembel", "5")
    _, _, _, tam, _ = olc("tembel", "say")
    print(f"{sayi:>6}{len(S) // sayi:>13}{b:>15}{o:>15}{sure:>13}{tam:>10}")
BOLUM, AD, SAHIP = KAYIT

 bolum  satir/bolum  ilk(5) acilan  ilk(5) okunan  ilk(5) sure  say sure
     4          550              1            550          600       600
    11          200              1            200          250       650
    44           50              1             50          100       600

Açılan bölüm üç kurulumda da birdir; değişen, o bölümün kaç satır taşıdığıdır. Beş satırlık sonuç için okunan satır 550’den 50’ye iner ve süre 600’den 100’e. Tembelliğin kazancı bölüm boyutuyla ölçülür: kaba bölünmüş bir kümede tembellik yazılıdır ama eleyecek bir şey bulamaz.

Son sütun ters yönü gösteriyor. Bütün sonucu isteyen say eyleminde 11 bölüm 650, dört ve kırk dört bölüm 600 süre veriyor. Neden bölüm sayısının işçi sayısına bölünmemesidir: 11 bölüm dört işçiye [3, 3, 3, 2] dağıldığı için en yüklü işçi 600 satır okurken bir işçi 400 satırda boş kalıyor. Bölümü inceltmenin görünmeyen bir bedeli de var — bu modelde süre hesabına GOREV_SABIT yalnız bir kez ekleniyor, oysa kırk dört bölüm kırk dört kez iş verilmesi demektir.

Tembelliğin Ertelediği Hata

Planın koşmaması bir kazanç olduğu kadar bir gecikmedir. Sürücü on satırlık bir dönüşüm zinciri yazdığında hiçbir satır okunmadığı için zincirin doğru olup olmadığı da sınanmaz. Aşağıdaki plan bir sayıyı bir metne bölmeye kalkışıyor ve tanımlandığı yerde hiçbir şey olmuyor.

sifirla()
bozuk = Tembel(BOLUM).esle(lambda k: dict(k, oran=k["m3"] / k["birim"]))
print("plan kuruldu, okunan satir:", SAYAC["okunan"])
try:
    bozuk.ilk(1)
except TypeError as e:
    print("eylem cagrilinca:", type(e).__name__, "-", e)
print("hataya kadar okunan satir:", SAYAC["okunan"])

plan kuruldu, okunan satir: 0
eylem cagrilinca: TypeError - unsupported operand type(s) for /: 'int' and 'str'
hataya kadar okunan satir: 200

Plan kurulurken okunan satır sıfırdır, yani hata plan yazılırken görünmez. İlk eylemde görünür ve o ana kadar 200 satır okunmuştur. İstekli kurulumda aynı hata esle çağrısının yazıldığı satırda düşerdi. Eleme kazancı ile hata gecikmesi aynı mekanizmadan gelir: hesabın ertelenmesi. Aynı mekanizmanın ikinci bir bedeli daha vardır — plan bir sonuç olmadığı için aynı küme üzerinde iki eylem çağrıldığında zincir iki kez koşar; bu bedelin ölçümü kursun ikinci konusuna aittir.

Özet

• Dağıtık veri kümesi bölümlerden ve henüz koşmamış bir plandan oluşur; dönüşüm plana yazılır, hesabı yalnız eylem başlatır ve plan kurulurken okunan satır sıfırdır.
• Bölüm verinin, işçi hesabın birimidir: 11 bölüm dört işçiye [3, 3, 3, 2] biçiminde dağılır ve bir eylem bölümlerin yalnız bir kısmını açabilir.
• Beş satırlık sonuçta istekli kurulum 44 bölüm açıp 6556 satır okurken tembel kurulum tek bölüm açıp 200 satır okuyor; 10 bölüm hiç açılmıyor, 2000 satır hiç okunmuyor.
• ilk(100) okunan satırı 200’den 600’e çıkarıyor ama süre 250’de kalıyor; bütün sonucu isteyen say eyleminde kazanç kullanılmayan daldan ve yazılmayan 3727 ara satırdan gelir (6556’ya karşı 2200 satır, 1838’e karşı 650 süre).
• Tembelliğin atlayabildiği en küçük birim bir bölümdür: aynı ilk(5) dört bölümlü kümede 550, kırk dört bölümlü kümede 50 satır okur; buna karşılık bölüm sayısı işçi sayısına bölünmediğinde tam tarama süresi 600’den 650’ye çıkar.
• Plandaki tür hatası ancak ilk eylemde ve 200 satır okunduktan sonra görünür; eleme kazancı ile hata gecikmesi aynı mekanizmadan gelir.

Sonraki Adım

Bu ders hangi hesabın koşacağına karar verdi, ama o hesabı kimin koşturacağına hiç dokunmadı. Bölümler işçilere sırayla verildi, işçi sayısı dört kabul edildi ve süre hesabında GOREV_SABIT sessizce eklendi. Gerçek bir kümede bu üç şeyin hiçbiri sürücünün elinde değildir: işi kaç işçiye böleceğine, o işçilerin kaçının aynı anda koşacağına ve bekleyen bir işin ne zaman sıra alacağına kaynak yöneticisi karar verir. Sonraki ders o kararı ölçer ve iki düğmeyi ayırır: görev sayısını büyütmek ile eşzamanlı görev sayısını büyütmek aynı şey değildir, çünkü her işçinin 50 birimlik sabit bedeli vardır ve bu bedel işçi sayısıyla birlikte büyür. Ölçülecek sayı süre kazancının hangi işçi sayısında bu bedele yenildiğidir.