Spark'ta partition yönetimi: repartition(), coalesce() ve Adaptive Query Execution

Bir Spark job'ının sürünmesi, çökmesi ya da düzgün çalışması çoğu zaman partition sayısına bağlı. Partition'lar az ya da dengesizse executor'ların heap'i bitiyor. Çok fazlaysa job minik task'lar arasında boğuluyor ve binlerce minik dosya yazıyor. Bu sayıyı değiştirmenin iki aracı repartition() ve coalesce(). Birbirinin yerine kullanılabilir gibi görünüyorlar ama değiller: biri shuffle yapıyor, öbürü kendisinden önce çalışan her şeyin paralelliğini sessizce çökertebiliyor.

Mekanizma değişmedi, ama bu araçlara ne sıklıkla ihtiyaç duyduğun değişti. Spark 3.2'den beri Adaptive Query Execution (AQE) varsayılan olarak açık ve bu tuning'in büyük kısmını çalışma anında, senin tahminin yerine gerçek istatistiklerle yapıyor. Bu yazıda iki çağrının gerçekte ne yaptığını, coalesce'un nerede ters tepebildiğini, AQE'nin senin yerine neleri hallettiğini ve hâlâ dokunmaya değer birkaç ayarı anlatıyorum.

İki çağrının farkı

PySpark'ın DataFrame.coalesce dokümanı gayet net. coalesce bir narrow dependency oluşturuyor: 1.000 partition'dan 100'e inerken shuffle olmuyor, yeni partition'ların her biri eskilerden 10 tanesini sahipleniyor. Elindekinden daha fazla partition istersen hiçbir şey olmuyor, mevcut sayıda kalıyorsun.

repartition ise tersini takas ediyor. Her zaman shuffle yapıyor ve sonuç hash partitioned oluyor. Bir sayı, bir ya da birkaç kolon, ya da ikisini birden verebilirsin. Yalnızca kolon verirsen varsayılan partition sayısı kullanılıyor, DataFrame'ler için bu spark.sql.shuffle.partitions demek.

df = spark.read.parquet("s3://bucket/events/")      # diyelim 2.000 input partition

df.coalesce(100)                    # shuffle yok; 100 partition, boyutlar dengesiz olabilir
df.repartition(100)                 # full shuffle; aşağı yukarı eşit 100 partition
df.repartition("country")           # country'ye göre hash; sayı = spark.sql.shuffle.partitions
df.repartition(50, "country")       # country'ye göre hash, tam 50 partition

print(df.coalesce(100).rdd.getNumPartitions())     # 100
print(df.coalesce(5000).rdd.getNumPartitions())    # yine 2000, coalesce sayıyı artırmaz

coalesce(1) tuzağı

En çok gözden kaçan kısım bu ve dokümanda açıkça uyarılıyor. coalesce narrow olduğu için Spark araya bir stage sınırı koymuyor. Azaltılmış partition sayısını veriyi üreten stage'e, yani yukarı doğru taşıyor. Bu yüzden sert bir coalesce(1), upstream hesaplamanın tamamını tek bir executor'da, tek bir task olarak çalıştırabilir.

# Masum görünüyor. Join, filtre ve UDF artık TEK bir task'ta çalışıyor.
(orders.join(customers, "customer_id")
       .filter("amount > 100")
       .withColumn("score", score_udf("amount"))
       .coalesce(1)
       .write.mode("overwrite").csv("s3://bucket/report/"))

Gerçekten tek bir çıktı dosyasına ihtiyacın varsa repartition(1) genelde daha iyi bir seçim. Bir shuffle ekliyor ama pahalı upstream iş paralel kalıyor, yalnızca son yazma adımı tek thread'e düşüyor. Daha da iyisi, tek dosyaya gerçekten ihtiyacın olup olmadığını sorgulamak. Bir klasörü okuyabilen bir downstream sistem bu derdi baştan ortadan kaldırıyor.

Adaptive Query Execution: çalışma anında tuning

AQE'nin varsayılan olarak açıldığı Spark 3.2'den önce shuffle partition sayısı pratikte plan aşamasında sabitleniyordu. spark.sql.shuffle.partitions değerini (varsayılan 200) bir kez ayarlıyordun, her join ve aggregation bu sayıyı kullanıyordu, sen de job bazında deneme yanılmayla tuning yapıyordun. Az partition'da executor'ların heap'i bitiyordu, çok partition'da minik task'lar ve minik dosyalar arasında boğuluyordun.

AQE bunu değiştiriyor. Performance tuning rehberine göre AQE en verimli planı seçmek için çalışma anındaki istatistikleri kullanıyor ve Apache Spark 3.2.0'dan beri varsayılan olarak açık. Ana anahtar spark.sql.adaptive.enabled. Burada en çok üç özellik önemli.

1. Shuffle partition'larını birleştirme

spark.sql.adaptive.coalescePartitions.enabled (varsayılan true) açıkken Spark gerçek shuffle çıktısına bakıyor ve yan yana duran küçük partition'ları birleştiriyor. Hedef boyut spark.sql.adaptive.advisoryPartitionSizeInBytes, varsayılanı 64MB.

Neredeyse kimsenin okumadığı bir detay var: spark.sql.adaptive.coalescePartitions.parallelismFirst varsayılan olarak true. Bu modda Spark 64MB hedefini yok sayıyor ve paralelliği en üst düzeyde tutmak, AQE'yi açanlarda performans gerilemesini önlemek için yalnızca coalescePartitions.minPartitionSize (varsayılan 1MB) sınırına uyuyor. Doküman, çok sayıda küçük task oluşmasın diye yoğun bir cluster'da bunu false yapmayı öneriyor. Paylaşımlı cluster'larda önce bu ayarı değiştirmeni öneririm.

spark.sql.shuffle.partitions'ın rolü de değişti. AQE açıkken bu değer pratikte birleştirmeden önceki başlangıç partition sayısı (coalescePartitions.initialPartitionNum ayarlanmamışsa bu değeri kullanıyor). Yani bugünkü doğru hamle mükemmel sayıyı aramak değil, değeri cömertçe yüksek tutup AQE'nin aşağı doğru birleştirmesine izin vermek.

2. Skew join yönetimi

spark.sql.adaptive.skewJoin.enabled (varsayılan true) sort-merge join'lerde skew'lu partition'ları bölüyor, gerekirse de çoğaltıyor. Bir partition, medyan partition boyutunun skewedPartitionFactor (varsayılan 5.0) katından ve skewedPartitionThresholdInBytes (varsayılan 256MB) değerinden büyükse skew'lu sayılıyor. Eskiden bu, key'leri elle salt'lamak demekti. Artık çoğu zaman otomatik. Bu da join'lerdeki "Java heap space" hatalarının yaygın sebeplerinden birini ortadan kaldırıyor.

3. REBALANCE hint'i

Yalnızca daha az çıktı dosyası elde etmek için coalesce kullanmanın modern alternatifi bu. Hint referansına göre REBALANCE, her çıktı partition'ının makul bir boyutta olmasını, yani ne çok küçük ne çok büyük olmasını hedefliyor. Best-effort çalışıyor, skew'lu partition'ları bölüyor ve tam olarak bir sorgu sonucunu tabloya yazmak için tasarlanmış. AQE kapalıysa yok sayılıyor.

from pyspark.sql import SparkSession

spark = (SparkSession.builder
         .appName("daily-events")
         # AQE 3.2'den beri zaten açık; açıkça yazmak niyeti belgeliyor
         .config("spark.sql.adaptive.enabled", "true")
         # paylaşımlı cluster'da maksimum paralellik yerine 64MB hedefine uy
         .config("spark.sql.adaptive.coalescePartitions.parallelismFirst", "false")
         # yüksekten başla, AQE aşağı doğru birleştirsin
         .config("spark.sql.shuffle.partitions", "800")
         .getOrCreate())

events = spark.read.parquet("s3://bucket/raw/events/")

(events.filter("event_date = '2026-09-22'")
       .hint("rebalance", "country")      # /*+ REBALANCE(country) */ ile aynı
       .write.mode("overwrite")
       .partitionBy("country")
       .parquet("s3://bucket/curated/events/"))

Aynısı SQL ile:

INSERT OVERWRITE TABLE curated.events
SELECT /*+ REBALANCE(country) */ *
FROM raw.events
WHERE event_date = '2026-09-22';

Elle repartition'ın hâlâ doğru olduğu durumlar

AQE sihir değil. Yalnızca shuffle sınırlarında devreye giriyor ve aklından geçeni bilemiyor. Şu durumlarda açık çağrılar hâlâ doğru araç:

• Yazma için belirli bir düzen gerektiğinde. partitionBy("country") öncesinde repartition("country") yapmak, her çıktı klasörüne her task'tan ayrı bir dosya düşmesini engelliyor.
• Girdinin kendisi kötü bölünmüşse. Bir avuç dev gzip dosyası sana bir avuç task verir ve AQE bir scan'i düzeltemez. Okumanın hemen ardından bir repartition(n) işi yayıyor.
• Veriyi Spark dışı bir şeye verirken, örneğin bağlantı limiti olan bir JDBC hedefine. Upstream iş paralel kaldığı sürece (Spark UI'dan kontrol et) orada coalesce(n) gayet uygun.
• İteratif algoritmalarda, yani bir DataFrame'in defalarca kullanıldığı ve bir kez partition'layıp cache'lemenin kendini amorti ettiği durumlarda.

Bir incelik: açıkça verdiğin bir sayı söz olarak kabul ediliyor. AQE bir repartition(400)'ü senden habersiz birleştirmiyor. AQE'nin yardımını istiyorsan sayıyı verme ya da REBALANCE kullan.

Ara not: Spark 4.0 ve ANSI modu

Eski job'ları yükseltiyorsan şunu da not et. Migration guide, Spark 4.0'dan beri spark.sql.ansi.enabled ayarının varsayılan olarak açık olduğunu söylüyor. Eskiden sessizce NULL dönen CAST('abc' AS INT) gibi cast'ler ve eskiden taşıp başa saran integer overflow'lar artık hata fırlatıyor. Bu iyi bir şey, ama sessiz null'lara güvenen pipeline'ları kırıyor. Hatalı girdinin beklendiği yerlerde try_cast ve diğer try_* fonksiyonlarını kullan. ANSI modunu geri kapatmak bir çözüm değil, en fazla kısa süreli bir köprü olmalı.

from pyspark.sql import functions as F

clean = raw.withColumn("qty", F.expr("try_cast(qty_str AS INT)"))   # hata yerine NULL

Kontrol listesi

• AQE'nin açık olduğunu doğrula (spark.sql.adaptive.enabled), özellikle eski ya da vendor'ın ayarladığı config'lerde.
• Yoğun, paylaşımlı cluster'larda coalescePartitions.parallelismFirst=false yap.
• spark.sql.shuffle.partitions'ı bir başlangıç noktası olarak gör: yüksek tut, AQE birleştirsin.
• Pahalı bir işten sonra asla coalesce(1) yapma. repartition(1) kullan ya da tek dosya ihtiyacını sorgula.
• Çıktı dosya boyutları için sayıları elle ayarlamadan önce REBALANCE'ı dene.
• Eski "sihirli sayı" repartition'larını kaldır ve Spark UI'da karşılaştır. Çoğunun hiçbir işe yaramadığı ortaya çıkıyor.
• Spark 4.x'te cast'leri ve aritmetik işlemleri ANSI hatalarına karşı gözden geçir.

repartition() ve coalesce()'u anlamak hâlâ değerli, çünkü düğmelerin ne işe yaradığını onlar anlatıyor. AQE ile ise çoğu zaman o düğmeleri çeviren sen olmamalısın.