Spark’ın klasik katalist optimizasyonu algoritması bir sorguyu çalıştırmadan önce o sorgunun planını kurar. İstatistikler güncel ve doğruysa bu yaklaşım oldukça etkilidir; değilse baştan verimsiz bir plan oluşturulmuş olur ve Spark işinin sonuna kadar bu verimsiz planla devam edilir. Üretimde bu çok klasik bir senaryodur. Bölüm eleme(partition pruning) beklenenden farklı çalışmış, bir join’in bir tarafı çalışma zamanında düşünüldüğünden çok daha küçük çıkmış ve “Spark işi neden 4 saat sürüyor?” sorusuna yanıt aramak için log incelenmeye başlanmıştır.
Uyarlanabilir Sorgu Uygulaması (AQE), Spark 3.0 ile gelen ve 3.2’den itibaren varsayılan olarak açık olan mekanizmadır; aşamalar tamamlandıkça gerçek çalışma zamanı istatistiklerini okuyarak kalan planı yeniden değerlendirir. Bu yazıda AQE’nin nasıl çalıştığını, üç ana optimizasyonu, özellikle çarpık verilerin (skew join) bakımı ve sınırlarını ele alınmaktadır.
Uyarlanabilir Sorgu Uygulaması (AQE) Nasıl Çalışır?
Spark bir query’i stage’lere böler ve stage sınırlarında shuffle yapar. Klasik yürütme(Execution) sürecinde plan baştan kurulur ve sonuna kadar uygulanır. AQE’nin yaptığı işlem şudur: bir eşleme karıştırma aşaması tamamlandığında, yazılan eşleme çıktılarının gerçek istatistikleri (bölüm başına byte, satır sayısı vb.) toplanır. AQE bu istatistiklere bakarak kalan planı revize eder, bölümleri birleştirir, çarpık olanları parçalar, hatta birleştirme stratejisini değiştirebilir.
Spark UI’ın SQL tab’ında AQE’nin işlettiği planı AdaptiveSparkPlan etiketiyle görmek mümkündür. Planın başta isFinalPlan=false olarak görünüp, runtime’da AQE düzeltmeleri uygulandıkça true’ya döndüğü gözlemlenir. Bir iş çalışırken planın çalışma sırasında değiştiğini görebilmek, AQE’yi öğrenmenin en atik yollarından biridir.
AQE’nin üç ana optimizasyonu bulunmaktadır: Karıştırma bölümlerini birleştirme, Çarpık verilerin bakımı ve Dinamik birleştirme stratejisi değişimi.
Karıştırma Bölümlerini Birleştirme: 200 Sayısına Veda
Spark’ın klasik problemlerinden biri, spark.sql.shuffle.partitions parametresinin varsayılan 200 değeridir. Bu sayı hiçbir iş için doğru olmaz: 1 GB’lık bir karıştırma için 200 bölüm aşırıdır (her biri ~5 MB, görev başlatma ek yükü baskın hale gelir); 500 GB’lık bir karıştırma için ise yetersizdir (her bölüm ~2,5 GB, yürütücü belleğinin tükenmesi (OOM) riski ortaya çıkar).
AQE’nin birleştirme mekanizması, karıştırma yazıldıktan sonra küçük bölümleri birleştirerek hedef boyuta yaklaştırır.
İlgili yapılandırmalar:
· spark.sql.adaptive.coalescePartitions.enabled (varsayılan true)
· spark.sql.adaptive.advisoryPartitionSizeInBytes (varsayılan 64 MB)
Somut bir örnek: 50 GB’lık bir karıştırma çıktısı 200 bölüm ile yazılmış, her biri ~256 MB civarında olsun. 64 MB’lık önerilen boyutla AQE bu bölümleri birleştirmez, zaten hedefin üstündedirler. Ancak bir filtre sonrası karıştırma çıktısı 5 GB’a düşmüşse, geriye 200 küçük bölüm kalmıştır ve AQE bunları ~80’e indirir. Görev sayısı düşer, zamanlama ek yükü azalır, iş hızlanır. Üstelik bunu yapmak için spark.sql.shuffle.partitions’ı her iş için elle ayarlamak gerekmez; yıllardır kod incelemelerinde dönüp duran bir başlık çoğu durumda gündemden düşer.
Çarpık Birleştirmelerin Yönetimi (Skew Join Handling)
Veri çarpıklığı (skew), dağıtık birleştirmelerin başlıca baş belalarından biridir. Bir birleştirme anahtarının (join key) dağılımında bazı değerler aşırı yoğun olduğunda — null user_id’ler, “unknown” değerler, popüler bir kiracının (tenant) ID’si — o anahtara denk gelen bölüme olağanüstü miktarda veri yığılır. Diğer görevler 30 saniyede biterken, o tek görev 2 saat çalışır ve iş onun bitmesini bekler. Spark arayüzünde (UI) “Tasks: 199 succeeded, 1 running” senaryosuyla karşılaşırsınız.
Klasik çözüm tuzlama (salting) yöntemiydi. Tuzlama çarpık anahtarlara rastgele bir değer ekleyip onları yapay olarak bölümlere dağıtmaya, birleştirme sonrasında ise geri toplamaya denir. Tuzlama etkili bir yöntemdir; ancak kodu karmaşıklaştırır ve özellikle zaman içinde çarpıklığa neden olan anahtarlar değişiyorsa bakımını zorlaştırır.
AQE’nin çarpık birleştirme yönetimi bunu çok daha temiz bir şekilde ele alır.
İşleyişi şu şekilde:
1. Karıştırma (Shuffle) map aşaması bittikten sonra AQE her bölümün boyutunu okur.
2. Ortanca (Median) bölüm boyutuna bakar. Eğer bir bölüm, ortancanın skewedPartitionFactor katından büyükse ve aynı zamanda skewedPartitionThresholdInBytes mutlak eşiğini geçiyorsa, “çarpık” (skewed) olarak işaretlenir.
3. Çarpık bölüm, advisoryPartitionSizeInBytes hedefine göre birden fazla küçük alt bölüme (subpartition) bölünür.
4. Birleştirmenin karşı tarafında bu anahtarlara denk gelen bölüm çoğaltılır (replicate), böylece her alt bölüm ihtiyaç duyduğu veriyi bulur.
Sonuçta tek bir dev görev yerine birkaç orta boy görev çalışır, iş dengeli biçimde tamamlanır.
İlgili yapılandırmalar:
· spark.sql.adaptive.skewJoin.enabled (varsayılan true)
· spark.sql.adaptive.skewJoin.skewedPartitionFactor (varsayılan 5)
· spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (varsayılan 256 MB)
Pratik bir not: Bir bölümün çarpık sayılması için iki koşulun da sağlanması gerekir. 10 MB’lık bir bölüm ortancanın 100 katı olsa bile AQE onu işlemez — çünkü zaten küçüktür, bölmenin bir anlamı yoktur. Bu ayrıntı, planı arayüzde (UI) inceleyenleri çok sık yanıltır; “ama çarpık bölüm yönetimi (skew) açık, neden çalışmıyor?” sorusunun cevabı genellikle bölümün eşik değerinin altında kalmasıdır.
Bir başka önemli sınır: AQE’nin çarpıklık yönetimi, dış birleştirmelerin (outer join) tüm türlerinde çalışmaz. Sürüme göre değişmekle birlikte, tam dış birleştirme (full outer join) ve bazı durumlarda sol/sağ dış birleştirmenin (left/right outer) çarpık tarafında olmayan kanadı bu mekanizmadan yararlanamaz. Planı incelerken bunu doğrulamak, beklediğiniz iyileşmeyi göremediğinizde ilk kontrol etmeniz gereken noktalardan biri olmalıdır.
Dinamik Birleştirme Stratejisi Değişimi
Daha az konuşulan ancak oldukça değerli bir yöntem. Catalyst, planı kurarken iki tablonun istatistiklerine bakarak birleştirme stratejisini seçer: küçük taraf yayınlanabiliyorsa (broadcast) yayın hash birleştirmesi (broadcast hash join), aksi halde sıralı birleştirme (sort-merge join) yapılır. Ama çalışma zamanında (runtime), bir kolda yapılan filtreleme veya toplama (aggregate) sonrasında bir tablonun beklenenden çok küçük çıktığını varsayalım — artık yayınlanabilir hale gelmiştir ama plan çoktan sıralı birleştirmeye karar vermiştir.
AQE bu durumu yakalar: karıştırma sonrasında gelen tarafın gerçek boyutu yayın eşiğinin altındaysa, birleştirmeyi dinamik olarak yayın hash birleştirmesine çevirir. Özellikle boyut tablosu (dimension table) birleştirmelerinde, filtre itmenin (filter push-down) çok etkili olduğu sorgularda ve istatistiklerin güncelliğini yitirdiği durumlarda önemli performans kazanımları sağlayabilir.
İlgili yapılandırma:
spark.sql.adaptive.autoBroadcastJoinThreshold.
Otomatik yayınlama yerine, çalıştıracağınız sorguda (query) tabloların boyutuna bakarak bu tabloları doğrudan yayınlayıp yayınlamamayı da seçebilirsiniz. Kümenizdeki ve yürütücülerinizdeki (executor) kaynaklara göre, 1 milyon satırlık bir tablo sizin için yayınlanabilir olabilir.
AQE Ne Çözmez?
AQE güçlü bir mekanizmadır ama sihirli bir kutu değildir; her problemi çözmez. Birkaç sınırını bilmek gerekir:
· Akışta (streaming) çalışmaz. AQE yalnızca toplu (batch) sorgular için geçerlidir. Structured Streaming işlerinizde çarpıklık problemi varsa hâlâ tuzlamaya (salting) veya özel bölümlemeye (custom partitioning) ihtiyaç duyulabilir.
· Çok küçük işlerde ek yük getirebilir. Saniyenin altında biten sorgularda plan revizyonu ek yükü göreli olarak büyür; interaktif not defteri (notebook) senaryolarında etkisini test etmekte fayda vardır.
· Önbelleğe alınmış planlarda davranış farklılaşır. Önbelleğe alınmış (cache) bir DataFrame üzerindeki plan AQE tarafından yeniden değerlendirilmez; önbelleğe alma anındaki plan kullanılır.
· AQE yanlış kodu düzeltmez. Kartezyen patlaması (cartesian explosion), gereksiz geniş dönüşümler (wide transformation), kötü yazılmış UDF’ler AQE ile düzelmez. AQE, karıştırma sonrasında gelen veriyle akıllıca uğraşır; o veriye gerek bile olmayan işlemler için yapabileceği bir şey yoktur.
Sonuç
AQE, bir önceki nesilde günler süren çarpıklık ve bölüm boyutlandırma (partition-sizing) tartışmalarını önemli ölçüde azaltıyor. Ama “açtık, tamam” demek yeterli değildir; neyin neden değiştiğini anlamak için Spark arayüzünde (UI) AQE planını okumayı, katsayı (factor) ile eşik (threshold) ikilisi gibi koşulları bilmeyi gerektirir. Önce varsayılan (default) ayarlarla başlamak, ardından işlerinizi gözlemleyerek doğru ayarları yapmak daha sağlıklı bir yaklaşımdır. Bilinçli kullanıldığında, çoğu zaman tek satırlık bir yapılandırma değişikliğiyle gelen en yüksek getirili iyileştirmelerden biridir.
Yazımıza Medium üzerinden de ulaşabilirsiniz.
Yazar: İbrahim Ekmen, Data Engineer
Fikirlerini paylaş, destek al.
Aşağıdaki iletişim bilgilerinden bize ulaşabilir ve iletişim formumuzu doldurabilirsiniz.
Seba Office Boulevard / İstanbul
Ayazağa Mah. Mimar Sinan Sok. A,
No: 21A İç Kapı No:1 Sarıyer, İstanbul
Levent Office / İstanbul
Yeşilce Mh. Göktürk Cad. Çeşni Sok. No:4 Kat:2 Kağıthane
P: 444 0 885
Ankara
Beştepe, Dumlupınar Blv. No: 10B Yenimahalle / Ankara
P: +90 312 419 43 43
Yardım
+90 212 220 70 80
Email:
info@treomind.com