groupBy yazmak üç saniye sürüyor; peki o üç saniyelik satır cluster’da neleri harekete geçiriyor? Gelin perdeyi birlikte aralayalım.

Apache Spark ile çalışan çoğumuz için hikâye şöyle başlar: Bir DataFrame okuruz, birkaç dönüşüm (transformation) zincirleriz, sonuca bakarız. Kod çalışır, iş biter. Ama işler yavaşlamaya başladığında — ki büyük veride er ya da geç başlar — perdenin arkasında olup biteni bilmeyenler için Spark bir kara kutuya dönüşür. “Shuffle çok olmuş” der birileri, “skew var” der bir başkası; peki bunlar fiziksel olarak ne demek? Veri o sırada gerçekten nereden nereye gidiyor?

Bu yazıda kod kalabalığına boğulmadan, diyagramlar üzerinden bir Spark job’ının yolculuğunu takip edeceğiz. Her bölümün başına, o bölümdeki akışı tetikleyen iki üç satırlık minik bir “çapa kod” koyacağım; asıl anlatım şekillerde olacak. Yolculuk boyunca elimizde tek bir senaryo olacak, hep ona döneceğiz:

# Senaryomuz: S3'teki sipariş verisi + Postgres'teki müşteri tablosu
orders    = spark.read.parquet("s3://datalake/orders/")        # ~2 TB
customers = spark.read.jdbc(url, "public.customers", ...)      # ~200 MB

result = (orders.join(customers, "customer_id")
                .groupBy("city")
                .agg(F.sum("amount").alias("total")))
result.write.parquet("s3://datalake/city_totals/")

Bu masum görünen beş satır; bir plan derlemesini, S3’ten paralel okuma yapılmasını, bir veritabanına sorgu atılmasını, ağ üzerinden terabaytlarca verinin taşınmasını ve yüzlerce görevin (task) koordinasyonunu tetikliyor. Sırayla göreceğiz.

1. Sahnedeki Oyuncular: Driver, Cluster Manager, Executor

Önce oyuncuları tanıyalım. Bir Spark uygulaması (application) üç ana bileşenden oluşur: işin beynini oluşturan sürücü (driver), kaynakları tahsis eden cluster manager (küme yöneticisi) ve asıl işi yapan yürütücüler (executors)^[1].

Buradaki en kritik ve en çok yanlış bilinen nokta şu: veri, normal akışta driver’dan geçmez. Driver S3’teki dosyaları indirmez, veritabanındaki satırları çekmez. Onun işi metadata ile: hangi dosyalar var, kaç parçaya bölünecek, hangi task hangi executor’a gidecek^[1][16]. Veri her zaman kaynaktan doğrudan executor’lara akar. Bu yüzden 2 TB’lık bir tabloyu işlerken driver’a 4 GB bellek çoğu zaman yeter — ama birazdan göreceğimiz iki istisna vardır: collect() çağrısı ve broadcast join’in toplama aşaması.

2. Kod Çalışmadan Önce: Tembellik ve Plan

result.explain(mode="formatted")   # daha tek satır veri okunmadan planı gösterir

PySpark’ta dönüşümler tembel (lazy evaluation) çalışır: join, groupBy yazdığınız anda hiçbir veri kıpırdamaz. Spark sadece “yapılacaklar tarifini” biriktirir. Siz write ya da count gibi bir eylem (action) tetiklediğinizde bu tarif, Catalyst Optimizer’dan geçerek mantıksal plandan (logical plan) fiziksel plana (physical plan) çevrilir; filtreler mümkün olduğunca kaynağa itilir (predicate pushdown), gereksiz sütunlar elenir^[7].

Fiziksel plan daha sonra aşamalara (stages) bölünür ve işte bu bölünmenin kuralı yazının kalbi: Spark, planı shuffle sınırlarından keser. Verinin partition’lar (bölümler) arasında yer değiştirmesi gerekmeyen her operasyon zinciri tek bir stage’de ardışık çalışır; veri taşınması gereken her nokta yeni bir stage başlatır. Bizim senaryodaki join ve groupBy bu sınırları oluşturacak. O halde önce verinin cluster’a ilk girişine, sonra da o meşhur taşınma anına bakalım.

3. Veri Kümeye Nasıl Girer? S3’ten Parquet Okumak

orders = spark.read.parquet("s3://datalake/orders/")
orders.rdd.getNumPartitions()   # kaç parçaya bölündü?

Terminoloji notu: Bu yazıda partition dediğimizde, Spark’ın veriyi bellekte işlemek için kestiği parçayı — yani bir task’in işlediği birimi — kastediyoruz. Türkçeye “bölüm” diye çevrilir ama sahada herkes İngilizcesini kullanır; biz de öyle yapacağız. Diskteki date=2026-08-01 tarzı klasör yapısı ise ayrı bir kavramdır: depolama bölümlemesi (Hive-style partitioning). İkisi aynı kelimeyi paylaşır ama aynı şey değildir; klasör bölümlemesine bu bölümün sonunda ayrıca değineceğiz.

Driver önce S3’e bir listeleme isteği atar: klasörde hangi dosyalar var, boyutları ne? Sonra bu dosya listesini görev dilimlerine (input splits) böler. Buradaki temel kural basittir: hiçbir partition spark.sql.files.maxPartitionBytes değerinden — varsayılan 128 MB — büyük olamaz^[2][12]. Büyük dosyalar 128 MB’lık dilimlere kesilir, küçük dosyalar ise (bir miktar açılış maliyeti, openCostInBytes, hesaba katılarak) aynı dilime paketlenebilir^[13][17].

Partition sayısının neden bu kadar önemli olduğunu şimdi görüyoruz: 1 partition = 1 task = 1 çekirdeğin (core) işi. 2 TB’lık orders klasörümüz kabaca 16.000 partition’a kesilecek; 100 çekirdeklik bir cluster’da bu, her çekirdeğin sırayla ~160 task işlemesi demek. Binlerce minicik dosyanız varsa tam tersi dert başlar: partition başına iş azalır ama task açma-kapama maliyeti (overhead) toplamı ezer. Meşhur “küçük dosya problemi” (small files problem) budur^[13].

Peki 16.000 task, 100 core’a nasıl sığıyor? Süpermarket kasası gibi: 100 kasa açık, 16.000 müşteri sırada. Driver’daki Task Scheduler stage başında task’leri bir kuyruğa dizer ve ilk 100’ünü boş slot’lara dağıtır; hangi task biterse onun slotu kuyruktan sıradakini çeker^[1][16]. Yani stage boyunca cluster’da aynı anda hep ~100 task koşar ve akış kesilmez — dalga dalga değil, sürekli bir bant gibi. Sıralama kabaca partition index’ini (P0, P1, P2…) izler; driver veri yerelliğini (data locality) de gözetir ama S3 gibi uzak kaynaklarda bunun pek önemi kalmaz. Garantili bir işlem sırası yoktur — zaten olmamalı da: “önce şu partition işlensin” varsayımına dayanan bir mantık kurduysanız, dağıtık dünyada yanlış yoldasınız demektir.

Parquet’nin sütunlu (columnar) yapısı burada bir hediye daha verir: task’ler dosyanın tamamını değil, sadece sorguda geçen sütunları indirir; dosya altbilgisindeki (footer) istatistikler sayesinde işe yaramayan satır grupları (row groups) hiç okunmadan atlanabilir^[7]. Klasör date=2026-08-01 gibi bölümlenmişse (partitioned), filtrenize uymayan klasörler listelenmez bile — partition pruning (bölüm budama) denen şey tam olarak bu.

4. Ya Kaynak Bir Veritabanı Tablosuysa? JDBC’nin Tuzağı

customers = spark.read.jdbc(url, "public.customers", properties=props)
customers.rdd.getNumPartitions()   # → 1  (evet, sadece 1!)

S3 okumasının doğal paralelliğine alışınca insan veritabanı tablosunu da aynı rahatlıkla okuyacağını sanıyor. Ama sürpriz: Spark, JDBC kaynağını varsayılan olarak tek bağlantı, tek partition halinde okur^[3]. Yani 100 çekirdeklik cluster’ınız seyrederken, tek bir task SELECT * FROM customers sorgusunun tamamını tek başına çeker. Dosya sisteminin aksine veritabanının Spark’a “beni şuradan şuraya kadar oku” diyen doğal bir blok yapısı yoktur; paralellik istiyorsanız bölme kuralını sizin söylemeniz gerekir^[8]:

customers = (spark.read.format("jdbc")
    .option("url", url).option("dbtable", "public.customers")
    .option("partitionColumn", "customer_id")   # sayısal / tarih kolonu
    .option("lowerBound", "1").option("upperBound", "4000000")
    .option("numPartitions", "4")
    .load())

İki incelik daha: lowerBound ve upperBound satırları filtrelemez, yalnızca adım (stride) hesabında kullanılır — aralık dışındaki satırlar da uç partition’lara dahil edilir^[3]. Ve seçtiğiniz kolon eşit dağılmıyorsa (mesela müşteri ID’lerinin yarısı ilk milyonda yoğunlaşmışsa), veri çarpıklığını (data skew) daha kaynağı okurken kendi elinizle yaratmış olursunuz^[9]. Bir de fetchsize var: bazı sürücüler (Oracle gibi) her ağ turunda yalnızca 10 satır getirir; bunu birkaç bine çekmek okuma hızını dramatik değiştirebilir^[3]. Son olarak eşzamanlılık: numPartitions=1000 bile deseniz, veritabanına aynı anda açık bağlantı sayısı min(numPartitions, cluster’daki toplam core) kadardır — her aralık bir task’tir ve bir slot boşaldıkça sıradaki aralığın sorgusu açılır (bir önceki bölümdeki kasa kuyruğu burada da işler). Yani numPartitions aynı zamanda DB’ye binecek eşzamanlı yükün üst sınırını da belirler^[3].

5. Veri Ne Zaman Yerinde Kalır, Ne Zaman Taşınır?

orders.filter("amount > 0")            # dar: veri yerinde kalır
orders.groupBy("city").sum("amount")   # geniş: veri taşınmak zorunda

Artık verimiz executor’lara dağılmış partition’larda duruyor. Bundan sonraki her operasyon iki kategoriden birine girer ve aradaki fark, Spark performansının neredeyse tamamını açıklar. Dar dönüşümlerde (narrow transformation) — filter, select, withColumn — her çıktı partition’ı tek bir girdi partition’ından hesaplanabilir; veri olduğu makinede kalır, ağ hiç kullanılmaz. Geniş dönüşümlerde (wide transformation) — groupBy, join, distinct, repartition — ise bir çıktı partition’ı, girdi partition’larının hepsinden parça ister^[16]. “İstanbul” satırları şu an 16.000 partition’ın hepsine serpilmiş durumda; toplamı alınacaksa hepsinin aynı yere gelmesi şart.

Sağdaki o çapraz ok yumağının teknik adı karıştırma, yani shuffle. Peki o oklar fiziksel olarak nasıl işliyor? Sıradaki bölüm yazının en derin dalışı.

6. Shuffle’ın Anatomisi: Diskler, Dosyalar, Ağ

daily = joined.groupBy("city").agg(F.sum("amount"))
# Spark UI → Stages: "Shuffle Write" ve "Shuffle Read" sütunlarına bakın

Yaygın bir yanılgıyla başlayalım: shuffle, executor’ların birbirine canlı olarak veri “ışınladığı” bir süreç değildir. Araya her zaman yerel disk girer^[4][5]. Süreç iki perdede oynanır:

Map tarafı (shuffle write): Kaynak stage’in her task’i, kendi partition’ındaki satırları tek tek ele alır ve her satır için basit bir hesap yapar: hash(city) % hedef_partition_sayısı. Bu hesap, satırın hangi hedef partition’a (dolayısıyla hangi renge) ait olduğunu söyler. Task, çıktısını hedef partition numarasına göre sıralanmış tek bir shuffle dosyası + nerede hangi blokun başladığını söyleyen bir indeks dosyası olarak kendi makinesinin diskine yazar^[4][14]. Hedef partition sayısını spark.sql.shuffle.partitions belirler; varsayılanı yıllardır 200’dür^[2].

Reduce tarafı (shuffle read / fetch): Yeni stage’in task’leri başladığında her biri, cluster genelindeki bütün map çıktılarına gidip “benim numarama ait bloku ver” der ve kendi parçalarını ağ üzerinden çeker (fetch)^[5][15]. Hangi blokun nerede olduğunu söyleyen adres defterini — yine veriyi değil, metadata’yı — driver’daki MapOutputTracker tutar^[5].

Bu resim, iki klasik derdi kendiliğinden açıklıyor. Birincisi maliyet: shuffle = serileştirme + disk yazma + ağ + disk okuma + ters serileştirme. Spark’taki en pahalı fiziksel operasyon budur; “shuffle’ı azalt” tavsiyesinin sebebi de bu zincirdir^[4]. İkincisi çarpıklık (skew): satırların yarısı İstanbul’a aitse, mavi bloklar devleşir; R0 task’i saatlerce çalışırken diğerleri dakikada biter. Stage’in süresi en yavaş task’in süresidir — Spark UI’da 199 task’in bitip 1 task’in sürünmesi manzarasının fiziksel açıklaması budur.

Bir de en çok karışan noktayı netleştirelim: shuffle write, spill değildir. Map task’in çıktısını diske yazması, bellek bol olsa bile her zaman olur; bu tasarımın kendisidir. Üstelik “bu parça başka worker’a gidecekse neden yerel diske?” sorusunun cevabı şu: o dosya başka worker’a gönderilmez, servis edilir. Her executor’ın içinde bir blok sunucusu (BlockManager / shuffle service) çalışır; reduce task ağdan “şu dosyanın 2 numaralı blokunu ver” diye istek atar, map tarafındaki sunucu bloku kendi diskinden okuyup ağa akıtır — veri itilmez (push), çekilir (pull)^[4][5]. Diske yazmak bu yüzden akıllıcadır: map stage bitip kaynaklarını serbest bırakabilir, reduce task’leri dakikalar sonra bile başlasa veri diskte sabırla bekler; çöken bir reduce task’i de map’leri baştan çalıştırmadan aynı dosyadan tekrar çeker. Spill ise bambaşka bir olaydır: bir task’in kendi içindeki bellek taşması. Reduce task’i gelen blokları birleştirirken kurduğu hash tablosu ya da sıralama tamponu execution memory’ye sığmazsa, ara veriyi geçici olarak diske boşaltır, sonra parçaları birleştirerek devam eder^[4]. Spill’de veri hiçbir yere gitmez — aynı task, aynı makine; sadece “RAM yetmedi, masanın üstünü yere indirdim” durumu. Spark UI’da “Shuffle Write” ile “Spill (Disk)”in ayrı sütunlar olmasının sebebi budur. Tek cümlede: shuffle write = başkasına servis edilmek üzere bilerek yazmak; spill = kendi işini bitirebilmek için mecburen taşmak.

İyi haber: Spark 3 ile birlikte varsayılan olarak açık gelen adaptif sorgu yürütme (Adaptive Query Execution, AQE), shuffle bittikten sonra gerçek boyutlara bakıp minik partition’ları birleştirir (coalesce) ve devleşmiş çarpık partition’ları otomatik olarak parçalara böler^[6][18]. Sihirli değnek değil ama 200 sabitiyle boğuşma devrini büyük ölçüde kapattı.

7. Join Sahnesi: İki Tablo Buluşunca Veri Nereye Gider?

joined = orders.join(customers, "customer_id")
joined.explain()   # planda "SortMergeJoin" mü "BroadcastHashJoin" mi yazıyor?

Senaryomuzun kilit anına geldik: 2 TB’lık orders ile Postgres’ten okuduğumuz 200 MB’lık customers buluşacak. Join’in temel problemi shuffle’ınkiyle aynıdır: aynı customer_id‘ye sahip satırların — iki tablodan da — fiziksel olarak aynı task’te buluşması gerekir. Spark bu buluşmayı başlıca iki stratejiyle sağlar ve ikisinin ağ maliyeti arasında uçurum vardır^[10][11].

Sıralamalı birleştirme (sort-merge join): Varsayılan yol. Her iki tablo da hash(customer_id)‘ye göre shuffle edilir — evet, 2 TB’ın tamamı ağdan geçer! Aynı ID aralığı iki taraftan da aynı partition’a düşer, partition’lar kendi içinde sıralanır ve iki sıralı liste fermuar gibi birleştirilir^[11].

Yayın birleştirmesi (broadcast hash join): Taraflardan biri küçükse — varsayılan eşik spark.sql.autoBroadcastJoinThreshold = 10 MB, ama hint ile elle de tetiklenebilir — Spark küçük tabloyu önce driver’da toplar, sonra bütün executor’lara kopyalar^[2][10]. Artık her executor’da müşteri tablosunun tam bir kopyası bellekte bir hash tablosu olarak durur; 2 TB’lık orders ise yerinden hiç kımıldamaz — her task kendi partition’ını yerel kopyayla eşleştirir.

Bizim 200 MB’lık customers tablomuz 10 MB eşiğinin üstünde; yani Spark kendiliğinden sort-merge seçecek. Ama tablo executor belleklerine rahat sığıyorsa orders.join(F.broadcast(customers), "customer_id") diyerek 2 TB’lık shuffle’ı tek satırla yok edebiliriz — Spark performans dünyasının en yüksek kazanç/emek oranlı hamlesi budur^[10]. Üstelik AQE, çalışma zamanında (runtime) tablonun filtreden sonra küçüldüğünü fark ederse sort-merge planını kendiliğinden broadcast’e çevirebilir^[6].

8. Peki Driver Bu Sırada Ne Yapıyor?

df.collect()        # ← tüm sonucu driver'a taşır. Büyük veride: OOM!
df.write.parquet()  # ← sonuç executor'lardan S3'e yazılır, driver rahat

Yolculuğun her adımında driver’ı hep kenarda, orkestra şefi rolünde gördük; toparlayalım. Driver job boyunca DAG’ı stage’lere böler, biten task’leri takip eder, başarısız olanları yeniden dener (retry), shuffle blok adreslerini tutar ve Spark UI’ı servis eder^[1][16]. Yani sürekli meşguldür ama taşıdığı şey hep bilgi hakkında bilgidir. Verinin driver’a fiilen aktığı yalnızca iki kapı vardır: sonucu bilinçli olarak istediğiniz collect() / toPandas() çağrıları ve broadcast join’in toplama aşaması. Driver’da yaşanan bellek taşmalarının (out of memory) neredeyse tamamı bu iki kapıdan girer. Milyonlarca satırı toPandas() ile çekmeye çalışan herkes bu dersi bir kez acıyla öğrenir; sonuç büyükse doğru kapı write‘tır — çünkü yazma işlemini de executor’lar kendi partition’ları için paralel yapar.

Kapanış: Artık Perde Şeffaf

Baştaki beş satırlık koda dönün ve zihninizde film şeridini oynatın: Driver planı derleyip S3’ü listeliyor; 2 TB, 128 MB’lık dilimler halinde executor’lara akıyor; Postgres’e (umarız!) dörder paralel WHERE’li sorgu gidiyor; join için her satıra hash(customer_id) ile hedef partition numarası atanıyor (Şekil 5’teki renkler) ve satırlar bu numaraya göre gruplanıp yerel disklere yazılıyor; reduce task’leri kendi numaralarına ait blokları ağdan topluyor; groupBy için aynı dans bir kez daha, bu kez hash(city) ile dönüyor; sonuç executor’lardan S3’e paralel yazılıyor. Ve driver? O sadece telsizle konuşuyor.

Bu zihinsel model bir kez oturunca, performans soruları kendiliğinden mekanik sorulara dönüşüyor: “Job yavaş” değil, “hangi stage’de, hangi shuffle’da, hangi partition’da yavaş?” diye sormaya başlıyorsunuz. Spark UI’daki her sayının bu diyagramlardan birinde bir karşılığı var. Bir sonraki yazıda bu modeli sahada koşturacağız: Spark UI’ı açıp yavaşlayan gerçek bir job’ın izini bu şekiller üzerinden süreceğiz. O zamana kadar — shuffle’larınız küçük, partition’larınız dengeli olsun!

---

Kaynaklar

1. Apache Spark Documentation — Cluster Mode Overview
2. Apache Spark Documentation — SQL Performance Tuning
3. Apache Spark Documentation — JDBC To Other Databases
4. Alexey Grishchenko — Spark Architecture: Shuffle
5. Apache Spark Wiki — Shuffle Internals
6. Databricks — Adaptive Query Execution: Speeding Up Spark SQL at Runtime
7. Databricks — Deep Dive into Spark SQL’s Catalyst Optimizer
8. Spark By Examples — Spark JDBC Parallel Read
9. Mercedes-Benz Tech Innovation — Increasing Apache Spark Read Performance for JDBC Connections
10. CanadianDataGuy — Mastering Spark Join Strategies
11. Spark Playground — Join Strategies in Spark Explained
12. Vladimir Prus — Spark Partitioning: The Fine Print
13. DZone — Guide to Partitions Calculation for Processing Data Files in Apache Spark
14. Bartosz Konieczny (waitingforcode) — Shuffling in Spark
15. Ajay Gupta — Revealing Apache Spark Shuffling Magic
16. AWS Prescriptive Guidance — Key Topics in Apache Spark
17. Kontext — spark.sql.files.maxPartitionBytes Explained in Detail
18. Databricks Documentation — Adaptive Query Execution
19. Kapak Resmi: Photo by Lukas Tennie on Unsplash