PySpark UDF’leri ve Pandas UDF’lerini Etkili Şekilde Nasıl Kullanırsınız

PySpark’ta Kullanıcı Tanımlı Fonksiyonlar (UDF’ler), Python geliştiricilerine, yerleşik Spark fonksiyonlarının tek başına başa çıkamadığı özgün görevleri ele almak için güçlü bir yol sunar. Python’da yetkin bir veri mühendisi, analist veya bilimciyseniz, UDF kavramlarını anlamak, karmaşık ve gerçek dünya veri zorluklarını etkili bir şekilde çözmenizi sağlar. 

Bu eğitim, PySpark UDF kavramlarını, pratik uygulamaları ve optimizasyon, test, hata ayıklama ile ileri kullanım kalıplarındaki en iyi uygulamaları adım adım açıklar. Sonunda, ölçekli olarak verimli UDF’leri güvenle yazabilecek, optimize edebilecek ve devreye alabileceksiniz.

PySpark’a yeniyseniz, önce PySpark’a Başlangıç öğreticimize göz atmanızı öneririm; burada ele aldıklarımız ileri düzey Spark kavramlarıdır.

PySpark UDF’leri nedir?

PySpark UDF’leri, Spark DataFrame’lerinde tutulan veriler üzerinde çalışmak üzere Spark’ın dağıtık yapısına entegre edilmiş özel Python fonksiyonlarıdır. Yerleşik Spark fonksiyonlarından farklı olarak, UDF’ler geliştiricilerin satır veya sütun düzeyinde karmaşık, özel mantık uygulamasına olanak tanır. 

Bizim PySpark El Kartımız Spark DataFrame’leri hakkında bilmeniz gereken her şeyi kapsar ve Spark UDF’lerini anlamanızı daha da kolaylaştırır. 

PySpark UDF’lerini ne zaman kullanmalısınız?

Şu durumlarda UDF kullanın:

• Mantığınız Spark’ın yerleşik fonksiyonlarıyla ifade edilemiyorsa.
• Dönüşümünüz, karmaşık Python-yerel işlemleri içeriyorsa (ör. regex işlemleri, özel NLP mantığı).
• Özellikle prototipleme sırasında veya küçük-orta boyutlu veri kümelerinde esneklik karşılığında performanstan feragat etmeyi kabul ediyorsanız.

Şu durumlarda UDF’lerden kaçının:

• Eşdeğer işlevsellik pyspark.sql.functions içinde varsa; Spark’ın yerel fonksiyonları daha hızlıdır, optimize edilmiştir ve yürütme motoruna itilebilir.
• Performansın kritik olduğu büyük veri kümeleriyle çalışıyorsanız. UDF’ler serileştirme yükü getirir ve Spark’ın yürütme planlarını optimize etme yeteneğini bozar.
• Mantığınızı SQL ifadeleri, Spark SQL yerleşikleri veya (vektörleştirilmiş işlemler için) Pandas UDF’leriyle ifade edebiliyorsanız.

Veri mühendisliğinde stratejik uygulamalar

PySpark UDF’lerinin başlıca kullanım alanları şunlardır:

• Gelişmiş metin ayrıştırma, veri çıkarımı veya dize işleme gibi karmaşık veri dönüşümleri.
• TensorFlow ve XGBoost gibi popüler makine öğrenmesi çerçeveleri dâhil üçüncü taraf Python kütüphaneleriyle entegrasyon.
• Eski sistemlerle köprü kurmak ve veri yapıları değişirken kesintisiz şema evrimini desteklemek.

UDF’ler, dağınık ve gerçek dünya veri mühendisliği görevlerini basitleştirerek ekiplerin çeşitli gereksinimleri esnek ve etkili bir şekilde ele almasını sağlar. 

Şimdi PySpark UDF’lerini nasıl uygulayabileceğinizi keşfedelim. 

PySpark UDF’lerinin uygulanması

Bu bölüm, PySpark UDF’lerinin nasıl tanımlanacağı ve uygulanacağına dair pratik bir özet sunar.

Standart UDF bildirim yöntemleri

PySpark’ta UDF bildirmek için yaygın üç yaklaşım vardır:

1. Lambda tabanlı UDF’ler: DataFrame sorguları içinde doğrudan hızlıca tanımlanır; basit işlemler için idealdir.

Lambda tabanlı UDF (Temel Python UDF), hızlı ve basit dönüşümler için en iyisidir. Performansın önemli olduğu büyük ölçekli işler için bunlardan kaçının.

İşte bir örnek: 

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

uppercase = udf(lambda x: x.upper() if x else None, StringType())
df = spark.createDataFrame([("Ada",), (None,)], ["name"])
df.withColumn("upper_name", uppercase("name")).show()Altyapı:

2. Dekoratörlü Python fonksiyonları: @pyspark.sql.functions.udf ile açıkça anotasyonlanır; yeniden kullanılabilirlik ve okunabilirlik sağlar.

Örneğin:

from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType

@F.udf(returnType=IntegerType())
def str_length(s):
    return len(s) if s else 0

df.withColumn("name_length", str_length("name")).show()Altyapı:

3. SQL’e kaydedilen UDF’ler: Doğrudan Spark SQL bağlamlarına kaydedilerek SQL sorguları içinde kullanılabilir.

from pyspark.sql.types import StringType

def reverse_string(s):
    return s[::-1] if s else ""

spark.udf.register("reverse_udf", reverse_string, StringType())

df.createOrReplaceTempView("people")
spark.sql("""SELECT name, reverse_udf(name) AS reversed FROM people""").show()Altyapı:

Her yöntemin ödünleri vardır: lambda UDF’ler kısa ve özdür ancak sınırlıdır; fonksiyon anotasyonları ise okunabilirlik, sürdürülebilirlik ve en iyi uygulamaları destekler.

Pandas UDF’ler, Arrow partileri üzerinde vektörleştirilmiş işlemlere izin verir. Genellikle normal UDF’lerden daha hızlıdır ve Spark’ın yürütme motoruyla daha iyi bütünleşir.

Skaler Pandas UDF (öğe bazlı, map gibi)

Büyük veri kümeleri üzerinde hızlı, satır bazlı dönüşümler için en uygundur. Örneğin: 

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import IntegerType
import pandas as pd

@pandas_udf(IntegerType())
def pandas_strlen(s: pd.Series) -> pd.Series:
    return s.str.len()

df.withColumn("name_len", pandas_strlen("name")).show()Altyapı:

Gruplanmış Harita (Grouped Map) Pandas UDF

Bunlar, Pandas’taki groupby().apply() ile benzer şekilde, grup başına özel mantık için en iyisidir.  

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, DoubleType
import pandas as pd

schema = StructType([
    StructField("group", StringType()),
    StructField("avg_val", DoubleType())
])

@pandas_udf(schema, F.PandasUDFType.GROUPED_MAP)
def group_avg(pdf: pd.DataFrame) -> pd.DataFrame:
    return pd.DataFrame({
        "group": [pdf["group"].iloc[0]],
        "avg_val": [pdf["value"].mean()]
    })

df.groupBy("group").apply(group_avg).show()Altyapı:

Pandas Toplama (Aggregate) UDF

Bu tür, gruplar üzerinde toplamalar yapar ve grouped map’ten daha hızlıdır. Örneğin: 

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DoubleType
import pandas as pd

@pandas_udf(DoubleType(), functionType="grouped_agg")
def mean_udf(v: pd.Series) -> float:
    return v.mean()

df.groupBy("category").agg(mean_udf("value").alias("mean_value")).show()Altyapı:

Pandas Iterator UDF

Pandas Iterator UDF, düşük bellek kullanımı gerektiren (toplu işleme) büyük veri kümeleri için en iyisidir. Örneğin: 

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import IntegerType
from typing import Iterator
import pandas as pd

@pandas_udf(IntegerType(), functionType="iterator")
def batch_sum(it: Iterator[pd.Series]) -> Iterator[pd.Series]:
    for batch in it:
        yield batch + 1

df.withColumn("incremented", batch_sum("id")).show()Altyapı:

Tür yönetimi ve null güvenliği

Türler ve null değerler, PySpark UDF’lerinde sık görülen zorluklardır. PySpark, katı tür denetimi uygular; bu da çoğu zaman örtük tür dönüşümlerine veya çalışma zamanı sorunlarına yol açar. Ayrıca Spark, null değerleri doğrudan UDF’lere iletir; açıkça ele alınmazsa olası çökmelere neden olur.

Dayanıklı UDF’ler için şu stratejileri uygulayın:

• Dönüş türlerini açıkça belirtin.
• Python fonksiyonlarınıza null kontrolleri (ör. koşullu ifadeler) ekleyin.
• Savunmacı kodlama uygulayın; basit null kontrolleri, can sıkıcı çalışma zamanı hatalarını önler.

UDF Performansını Optimize Etme

Satır satır yürütme modeli nedeniyle standart UDF’lerde performans sıklıkla zayıf halka olur. Vektörleştirilmiş UDF’lerden ve Spark’ın optimizasyon araçlarından yararlanmak, çalışma sürelerini önemli ölçüde iyileştirir.

Pandas entegrasyonuyla vektörleştirilmiş UDF’ler

Pandas UDF’ler, veri partilerini Python fonksiyonlarına Pandas Series olarak ileterek PySpark’ta UDF’lere vektörleştirilmiş bir yaklaşım getirir. Bu tasarım, standart satır bazlı UDF’lere kıyasla serileştirme yükünü azaltarak performansı önemli ölçüde artırır.

JVM ve Python süreçleri arasında sıfır kopyalı veri aktarımı için Apache Arrow tarafından desteklenen Pandas UDF’ler, ölçekli işlemlerin verimli şekilde yürütülmesini sağlar. Özellikle milyonlarca kayıt boyunca yoğun hesaplamalar ve karmaşık dize işlemleri için etkilidirler. 

PySpark ile veri manipülasyonu hakkında daha fazla ayrıntıyı PySpark ile Veri Temizleme kursumuzda ele alıyoruz. 

Ayrıca, Pandas UDF’ler, tanıdık araç ve iş akışlarından yararlanarak daha geniş Python veri bilimi ekosistemiyle sorunsuz entegrasyon sağlar.

Arrow optimizasyon teknikleri

Apache Arrow’un sütunsal bellek formatı, Spark JVM ile Python süreçleri arasında sıfır kopyalı, verimli veri aktarımına olanak tanır. Spark yapılandırmalarınızda Arrow’u (spark.sql.execution.arrow.pyspark.enabled=true) etkinleştirerek, veriler JVM ve Python ortamları arasında hızla taşınır ve UDF yürütmesi kayda değer ölçüde hızlanır.

Yürütme planı optimizasyonu

PySpark işlerini optimize etmek, Spark’ın Catalyst optimize edicisini nasıl etkileyebileceğinizi anlamayı içerir. Gelişmiş stratejiler arasında koşul itme (predicate pushdown), sütun budama ve sorgu planlaması ile yürütme verimliliğini artırmak için yayın (broadcast) birleştirme ipuçları kullanmak gibi teknikler yer alır.

Performansı en üst düzeye çıkarmak için, UDF yürütmesinin kapsamını en aza indirmek ve mümkün olduğunda yerleşik Spark SQL fonksiyonlarını tercih etmek önemlidir. Stratejik önbellekleme ve dikkatli plan kurgusu, iş yürütme hızını ve kaynak kullanımını daha da iyileştirebilir.

Performans optimizasyonu, bir PySpark mülakatında karşılaşabileceğiniz başlıca sorulardan biridir. Bu ve daha fazla Spark sorusuna nasıl yanıt vereceğinizi 2025 için En İyi 36 PySpark Mülakat Sorusu ve Cevabı blog yazımızda keşfedin. 

İleri Düzey Kalıplar ve Anti-Kalıplar

Doğru ve yanlış kullanım kalıplarını anlamak, istikrarlı ve verimli UDF dağıtımları sağlamaya yardımcı olur.

Durum tutan UDF uygulamaları

Durum tutan ve belirleyici olmayan UDF’ler, PySpark’ta özgün zorluklar sunar. Bu fonksiyonlar, ortam değişkenleri, sistem zamanı veya oturum bağlamı gibi harici duruma veya değişen koşullara bağlı sonuçlar üretir.

Belirleyici olmayan UDF’ler bazen gereklidir—örneğin zaman damgası üretmek, kullanıcı oturumlarını izlemek veya rastgelelik eklemek—ancak hata ayıklamayı, tekrarlanabilirliği ve optimizasyonu zorlaştırabilirler.

Durum tutan UDF’lerin uygulanması, net davranış dokümantasyonu, yan etkilerin yalıtılması ve sorun gidermeyi kolaylaştırmak ile iş çalıştırmaları arasında tutarlılığı sağlamak için kapsamlı günlükleme gibi dikkatli tasarım kalıpları gerektirir.

Düşünceli kullanıldıklarında güçlü yeteneklerin kilidini açabilirler; ancak güvenilir veri hatlarını sürdürmek disiplinli bir yönetim gerektirir. PySpark ile Büyük Veri Temelleri kursumuzda PySpark’ta büyük veriyi nasıl ele alacağınıza dair daha fazla ayrıntıya yer veriyoruz. 

Yaygın anti-kalıplar

UDF kullanımındaki yaygın anti-kalıplar, PySpark performansını önemli ölçüde düşürebilir:

• Toplu işleme yerine satır bazlı işleme: UDF’leri tek tek satırlara uygulamak yerine Pandas UDF’ler gibi vektörleştirilmiş yaklaşımları kullanmamak, büyük yavaşlamalara yol açar.
• UDF’lerin içinde iç içe DataFrame işlemleri: UDF’lerin içine DataFrame sorguları gömmek, aşırı hesaplamaya neden olur ve Spark’ın yürütme planlarını optimize etme kabiliyetini engeller.
• Satır içi UDF kaydını tekrarlamak: UDF’leri sorgular içinde defalarca tanımlayıp kaydetmek gereksiz yük getirir; UDF’leri bir kez tanımlayıp işleri genelinde yeniden kullanmak daha iyidir.
• Basit işlemler için aşırı özel Python mantığı kullanmak: Temel filtreleme, aritmetik veya basit dönüşümler gibi görevler için, özel UDF’ler yerine Spark’ın yüksek derecede optimize edilmiş yerleşik fonksiyonları tercih edilmelidir.

Bu tuzaklardan kaçınmak, daha iyi performans, Catalyst tarafından daha kolay optimizasyon ve daha sürdürülebilir PySpark kodu sağlar.

PySpark UDF’lerinde hata ayıklama ve test

UDF’leri test etmek ve hata ayıklamak, üretim senaryolarında güvenilirlik ve dayanıklılık sağlar.

İstisna yönetimi kalıpları

UDF’ler içinde yapılandırılmış hata yakalama uygulamak, dayanıklı ve sürdürülebilir PySpark hatları inşa etmek için esastır. Null değerler, tür uyumsuzlukları veya sıfıra bölme hataları gibi çalışma zamanı sürprizlerini zarifçe ele almak için UDF’lerde try-except blokları kullanın.

Sağlam istisna yönetimi, beklenmeyen verilere karşı hatları dengeler ve net, eyleme dönük hata mesajları sunarak hata ayıklamayı basitleştirir. Doğru şekilde yakalanıp günlüğe kaydedilen istisnalar, UDF davranışını daha şeffaf kılar; sorun çözümünü hızlandırır ve genel hat güvenilirliğini artırır.

Birim test çerçeveleri

UDF’leriniz için güvenilir birim testler yazmak üzere PySpark’ın yerleşik test taban sınıfı pyspark.testing.utils.ReusedPySparkTestCase’i, pytest gibi çerçevelerle birlikte kullanın. Net ve odaklı testler yapılandırmak, zaman içinde UDF mantığınızın doğruluğunu, kararlılığını ve sürdürülebilirliğini güvence altına alır.

UDF’leri test etme en iyi uygulamaları arasında, hem tipik hem de uç durumları kapsamak, çıktıları bilinen sonuçlara karşı doğrulamak ve UDF davranışını dış bağımlılıklardan yalıtmak yer alır. İyi tasarlanmış testler sadece regresyonlara karşı koruma sağlamakla kalmaz, aynı zamanda gelecekteki geliştirme ve yeniden düzenleme çalışmalarını da basitleştirir.

Evrim ve Gelecek Yönelimleri

PySpark ekosistemi hızla evrilmeye devam ediyor ve UDF’leri daha da güçlendiren yeni yetenekler sunuyor.

Unity Catalog entegrasyonu

Son gelişmeler, UDF kaydını Unity Catalog ile entegre ederek UDF’lerin ölçekli olarak nasıl yönetildiğini, keşfedildiğini ve yönetişiminin sağlandığını sadeleştirdi. Unity Catalog, kurumsal ortamlarda kritik olan kayıt, versiyonlama ve erişim denetimi dâhil UDF yaşam döngüsü yönetimi üzerinde merkezi kontrol sağlar.

Bu entegrasyon, yönetişimi güçlendirir, tutarlı güvenlik ilkeleri uygular ve ekipler arası keşfedilebilirliği artırır; böylece UDF’lerin büyük ve karmaşık veri ekosistemleri içinde yeniden kullanımı, denetlenmesi ve yönetimi kolaylaşır.

GPU hızlandırmalı UDF’ler

RAPIDS Accelerator gibi çerçeveler, PySpark’ta hesaplama açısından yoğun UDF görevleri için GPU’ya aktarmayı mümkün kılarak dönüştürücü performans iyileştirmeleri sunar. Sayısal analiz, derin öğrenme çıkarımı ve büyük ölçekli veri modelleme gibi ağır işlemleri GPU’lara kaydırarak, uygun iş yüklerinde yürütme sürelerini saatlerden dakikalara indirebilir.

GPU hızlandırma, özellikle devasa veri kümeleri, karmaşık vektörleştirilmiş hesaplamalar ve makine öğrenmesi hatları içeren senaryolarda faydalıdır; PySpark’ın modern veri mühendisliği görevleri için performans ve ölçeklenebilirliğini çarpıcı biçimde genişletir. PySpark ile Makine Öğrenmesi kursumuz bu kavramları daha derinlemesine ele alır. 

Sonuç

PySpark UDF’leri, Spark’ın yeteneklerini genişletmek için güçlü bir araçtır; ekiplerin yerleşik fonksiyonların ötesine geçen karmaşık ve özelleştirilmiş veri işleme görevlerini ele almasına olanak tanır. Doğru uygulandıklarında, büyük ölçekli veri hatlarında esneklik ve yenilikçiliğin önünü açarlar. 

Bununla birlikte, UDF performansını optimize etmek dikkatli bir yaklaşım gerektirir; satır bazlı işlemler gibi yaygın tuzaklardan kaçınmak, istisnaları zarifçe yönetmek ve Arrow entegrasyonu ile Pandas UDF vektörleştirmesi gibi tekniklerden yararlanmak önemlidir.

RAPIDS gibi çerçeveler aracılığıyla GPU hızlandırma gibi ortaya çıkan gelişmeler, UDF odaklı iş akışlarında mümkün olanları daha da genişletiyor. İster dağınık gerçek dünya verilerini dönüştürüyor olun, ister üretim sistemlerine gelişmiş analitik gömüyor olun, UDF en iyi uygulamalarına hâkim olmak hızlı, verimli ve güvenilir veri hatları kurmak için esastır. 

Veri bilimcilerin zamanlarının %70-80’ini harcadıkları veri düzenleme ve özellik mühendisliğinin tüm ayrıntılarını PySpark ile Özellik Mühendisliği kursumuzdan öğrenin.