PySpark Filter Eğitimi: Teknikler, Performans İpuçları ve Kullanım Alanları

Filtreleme, PySpark’ta temel bir işlemdir ve büyük veri kümelerini hızlıca daraltıp ilgili bilgilere odaklanmak için gereklidir. PySpark’ta filtreleri etkin bir şekilde kullanmak, orta seviyedeki veri mühendisleri, veri bilimcileri ve büyük veri işleme ile uğraşan geliştiricilerin iş akışı verimliliğini artırabilir. 

Bu makale, temel ve ileri seviye PySpark filtreleme tekniklerini tanıtır, daha iyi performans için optimizasyon stratejilerini özetler ve gerçek hayata uygulanabilir pratik kullanım örneklerini kapsar.

Bu eğitim, temel düzeyde PySpark bilgisi varsayar; ancak PySpark’ın ne olduğunu ve nasıl kullanılabileceğini PySpark’a Başlangıç eğitimimizde keşfedebilirsiniz. 

PySpark Filter (Filtre) İşlemi Nedir?

2025 tarihli Sıfırdan PySpark Öğrenin rehberimizde de belirtildiği gibi, PySpark kullanım kolaylığı ve verimliliği nedeniyle veri biliminde yaygın bir araçtır. filter() ve where() gibi yerleşik fonksiyonlarla veri kümelerini verimli şekilde filtrelemek için basit yollar sunar. Bu fonksiyonlar, belirtilen koşulları sağlayan DataFrame satırlarını ayıklamaya yardımcı olur.  

Filtreleme; veri ön işleme, analiz ve veri hattı optimizasyonunda kritiktir. İlgisiz veya hatalı kayıtları erken aşamada hariç tutmanızı sağlayarak zaman kazandırır, maliyetleri düşürür ve sonraki analiz veya modelleme görevlerini iyileştirir.

Basitliğine rağmen, dağıtık ortamlarda filtre uygulamak; performans darboğazları, verimsiz veya karmaşık filtre koşulları ve eksik ya da tamamlanmamış verinin hatalı ele alınması gibi zorluklar yaratabilir.

Spark filtrelerini anlamak, bir veri mühendisliği mülakatında başarılı olmak için hayati önemdedir. 

Temel kavramlardan (veri depolama ve dağıtık hesaplama) makine öğrenimi ve güvenlik gibi ileri konulara kadar büyük veri mülakatlarında sorulan başlık ve soruları En İyi 30+ Büyük Veri Mülakat Sorusu eğitimimizle öğrenin. 

PySpark Filtrelemenin Temelleri

PySpark’ta filtreleme, Apache Spark üzerine kurulu dağıtık yürütme modelinin çerçevesinde çalışır. Bir filtre uyguladığınızda (ör. .filter() veya .where()), PySpark veriyi anında taramaz. Bunun yerine, gerçekleştirmek istediğiniz dönüşümleri ana hatlarıyla belirten mantıksal bir sorgu planı oluşturur.

PySpark’ta filtreleme verimliliğini önemli ölçüde etkileyen iki temel kavram vardır:

• Tembel Değerlendirme (Lazy Evaluation) ve
• Catalyst Enoptimizörü.  

Tembel değerlendirme

PySpark bir filtre komutunu hemen yürütmez; bunun yerine mantıksal bir sorgu planı oluşturur ve açıkça çalıştırılması söylenene kadar işlemi erteler. Bu model, PySpark’a veri hatları genelinde işlemleri optimize etmede daha fazla esneklik sağlar.  

Catalyst enoptimizörü

Spark’taki bu optimizasyon motoru, filtreleme dahil mantıksal planları değerlendirir ve bunları en hızlı, en verimli şekilde yürütmenin yolunu otomatik olarak belirler.

Yürütme başladığında, filtrelenen veriler RDD’ler veya DataFrame’ler kullanılarak işçi düğümler arasında paralel biçimde işlenir. Her bölüm (partition) filtreyi bağımsız uygular; bu da büyük veri kümeleri için ölçeklenebilirlik sağlar.

RDD’ler hakkında daha fazla bilgiyi PySpark Cheat Sheet: Spark in Python eğitimimizde keşfedebilirsiniz. 

Temel PySpark Filtre Sözdizimi  

Koşullara göre veri filtrelemek basittir. Koşulları sütun ifadeleriyle veya SQL benzeri dizelerle sağlayabilirsiniz. 

1. Sütun ifadeleri

Koşulları tanımlamak için DataFrame sütun nesnelerini ve operatörleri kullanın:

filtered_data = df.filter(df.age > 25)Altyapı:

2. SQL benzeri dizeler

Koşulu SQL tarzı bir dize olarak yazın:

filtered_data_2 = df.filter("age > 25")Altyapı:

Alternatif olarak, where() fonksiyonunu da aynı şekilde kullanabilirsiniz:

filtered_data_3 = df.where(df.age > 25)
filtered_data_4 = df.where("age > 25")Altyapı:

Bunların tümü geçerlidir ve aynı sonucu üretir. Hangisini seçeceğiniz genellikle kişisel tercihe veya bağlamınızda hangisinin daha okunaklı göründüğüne bağlıdır.

PySpark’ta SQL sözdizimini kullanma hakkında daha fazla bilgi için Python ile Spark SQL’e Giriş kursumuza göz atın. 

PySpark Filtreleme Teknikleri ve Uygulamaları

Şimdi çeşitli filtreleme yöntemlerini ve pratik kullanım durumlarını inceleyelim.

İlişkisel filtreler  

Verileri sayısal veya kategorik eşiklere göre izole etmek için büyüktür >, küçüktür <, eşittir == gibi ilişkisel operatörlerle filtreleyebilirsiniz:

greater_than_df = df.filter(df.salary > 50000)  
category_match_df = df.filter(df.country == 'France')Altyapı:

Mantıksal filtreler  

Birden fazla koşulu tek bir filtrede birleştirmek için VE &, VEYA | ve DEĞİL ~ gibi mantıksal operatörleri kullanarak karmaşık gereksinimler için özlü ifadeler oluşturabilirsiniz:

combined_df = df.filter((df.age > 25) & (df.city == 'New York'))Altyapı:

Gelişmiş Filtreleme Fonksiyonları  

PySpark, özel filtreleme durumları için gelişmiş fonksiyonlar içerir. Bazılarına bakalım. 

Üyelik Testleri: Önceden tanımlı bir kümeden değerleri filtrelemek için isin kullanın:

selected_countries_df = df.filter(df.country.isin('Japan', 'Brazil', 'India'))Altyapı:

Desen Eşleme: Metin desenlerine göre filtreleme için basit eşleşmelerde like, regex için rlike  kullanın:

name_filter_df = df.filter(df.name.like('A%'))  
regex_filter_df = df.filter(df.name.rlike('^[A-Z][a-z]+Altyapı:

Aralık Filtreleri: between gibi kullanışlı yöntemlerle filtreleme:

range_df = df.filter(df.age.between(25, 35))Altyapı:

Null Kontrolleri: Eksik değerleri açıkça null kontrolü yaparak sorunsuzca ele almak:

valid_email_df = df.filter(df.email.isNotNull())Altyapı:

Karmaşık Veri Tiplerini Filtreleme

Kuruluşlar sıklıkla iç içe (nested) veya yapısal veriler kullanır. PySpark filtreleme, iç içe alanlar ve dizileri (array) destekler. Yapısal veri (struct) için, iç içe alanları kolayca filtreleyin:

city_df = df.filter(df.address.city == 'Boston')Altyapı:

Dizi olarak depolanan veriler için array_contains gibi yerleşik fonksiyonlardan yararlanabilirsiniz:

from pyspark.sql.functions import array_contains  

skill_df = df.filter(array_contains(df.skills, 'Python'))Altyapı:

Verimli PySpark Filtreleme için Optimizasyon Teknikleri

Bu bölümde, PySpark’ta filtre işlemlerinin performansını artırmaya yönelik stratejileri keşfedeceksiniz.

Predicate pushdown  

Predicate pushdown, filtreleme koşullarını veri depolama katmanına olabildiğince yakına yerleştirmeyi ifade eder. Bu, PySpark’ın başlangıçta daha az veri okuması anlamına gelir ve performansı iyileştirir.

Örneğin: 

# Read from a Parquet file
df = spark.read.parquet("s3://your-bucket/sales_data/")

# Apply filter on a non-partition column (e.g., product_category)
filtered_df = df.filter(df.product_category == "Electronics")

# Trigger query plan inspection
filtered_df.explain(True)Altyapı:

Yukarıdaki örnekte, normal bir sütuna filtre uyguluyoruz. Normalde Spark tüm veri kümesini yükleyip bellekte filtrelerdi; fakat Parquet predicate pushdown’ı desteklediğinden Spark filtre koşulunu doğrudan Parquet okuyucusuna iletebilir.

product_category kontrolünü tüm satırları okuyup sonra yapmak yerine, Spark yalnızca product_category = 'Electronics' olan satırları okur. Bu da disk I/O’sunu ve belleği korur, işlemi hızlandırır.

explain, Spark’ın yürütme sırasında tam olarak ne yapmayı planladığını gösteren fiziksel sorgu planını yazdırır.

Predicate pushdown uygulanmışsa, şuna benzer bir çıktı görürsünüz:

PushedFilters: [EqualTo(product_category,Electronics)]Altyapı:

Bu, Spark’ın filtreyi veriyi yükledikten sonra uygulamak yerine dosya tarama seviyesine kadar ittiğini doğrular.

Predicate pushdown, Spark’ın araç setindeki en güçlü optimizasyonlardan biridir. Şunları sağlar:

• Diskten okunan veri miktarını azaltır.
• Sorgu yürütmesini hızlandırır.
• Bellek kullanımını ve ağ I/O’sunu düşürür.

Filtrelerinizi net yazarak ve veriyi Parquet gibi formatlarda depolayarak Spark’ın perde arkasında yürütmeyi optimize etmesine olanak tanırsınız.

PySpark’ta join yaparken de performans optimizasyonu kritiktir. PySpark Joins: Büyük Veri Join Performansını Optimize Etme eğitimimizle shuffle’ları azaltmayı, çarpıklığı (skew) ele almayı ve büyük veri hatları ile makine öğrenimi iş akışlarında performansı artırmayı öğrenin. 

Bölüm kırpma (partition pruning)  

Veri ilgili alanlara göre (örneğin tarih) bölümlendirildiğinde, PySpark gereksiz bölümleri yüklemekten kaçınabilir. Etkin bölüm kırpma, veri taramalarını ve yürütme süresini önemli ölçüde azaltır.

year ve month ile bölümlendirilmiş bir Parquet veri kümesinin s3://your-bucket/events_data/ altında tutulduğunu varsayalım. 

Her bölüm şu şekildedir:

events_data/year=2023/month=01/
events_data/year=2023/month=02/

...

events_data/year=2024/month=01/Altyapı:

PySpark’ta bölüm kırpmaya bir örnek şu şekilde görünür:

df = spark.read.parquet("s3://your-bucket/events_data/")
filtered_df = df.filter((df.year == 2023) & (df.month == 6))

# .explain(True) to verify pruning
filtered_df.explain(True)Altyapı:

year ve month bölüm sütunlarıdır ve Spark’ın Catalyst enoptimizörünün alakasız bölümleri kırpmasına olanak tanır. 

Önbelleğe alma ve kalıcılık (caching & persistence)  

Filtrelenmiş sonuçlar analiz iş akışında sıklıkla birden fazla kez kullanılır. Filtrelenmiş sonuçları geçici olarak bellekte saklamak, tekrar hesaplamayı önlemeye yardımcı olur; ancak kaynakları korumak için dikkatli kullanılmalıdır.

recent_users_df = df.filter(df.registration_date >= '2022-01-01')  

recent_users_df.cache()Altyapı:

Gelişmiş Filtreleme Stratejileri ve Kullanım Durumları  

Şimdi, gelişmiş filtreleme tekniklerinin uygulandığı karmaşık senaryoları inceleyelim.

Zamana dayalı filtreleme  

Zaman serisi verilerini işlerken, sıklıkla kesin zamanlamalı filtreleme gerekir. Belirli bir zaman aralığındaki verileri filtrelemek basittir:

time_df = df.filter((df.date >= '2023-08-01') & (df.date < '2023-09-01'))Altyapı:

Yukarıdaki ifade, 2023 Ağustos ayındaki tüm kayıtları seçer. Koşul hem DateType hem de TimestampType sütunlarıyla çalışır.

Hareketli metrikler hesaplama veya zaman içindeki eğilimleri belirleme gibi daha derin zaman serisi analizleri için, zaman filtrelerini pencere (window) fonksiyonlarıyla eşleştirebilirsiniz. Örneğin, önce bir tarih aralığını filtreleyip ardından hareketli ortalama uygulayabilir veya her gün içindeki olayları sıralayabilirsiniz:

from pyspark.sql.window import Window
from pyspark.sql.functions import avg
window_spec = Window.partitionBy("date").orderBy("timestamp").rowsBetween(-2, 0)
df_filtered = df.filter((df.date >= '2023-08-01') & (df.date < '2023-09-01'))
df_with_avg = df_filtered.withColumn("rolling_avg", avg("value").over(window_spec))Altyapı:

Bu örnek, zamana göre sıralanan değerler için gün başına 3 satırlık bir hareketli ortalama hesaplar. Zaman filtrelerini pencere fonksiyonlarıyla birleştirmek, zaman temelli analitiklerde size hassas kontrol sağlar.

Akış verilerinde filtreleme  

Gerçek zamanlı veri akışlarını filtrelemek benzersiz zorluklar içerir. Yığın (batch) işlemeye kıyasla veri sınırsızdır ve sürekli akar. Bu da filtrelerin anlık uygulanmasını ve doğru sonuçlar için geç gelen olayların dikkatle ele alınmasını gerektirir.

PySpark’ın Structured Streaming’i, akış verilerini filtrelemek ve zamanlama sorunlarını etkili biçimde yönetmek için sezgisel araçlar sunar. Statik bir DataFrame üzerinde olduğu gibi filtreler uygulayabilir ve sonuçları kesinleştirmeden önce sistemin geç gelen verileri ne kadar süre beklemesi gerektiğini su damgası (watermark) ile kontrol edebilirsiniz.

İşte bir örnek:

# Example: Real-Time Filtering with Watermarking

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StringType, TimestampType

# Set up Spark session
spark = SparkSession.builder.appName("RealTimeFiltering").getOrCreate()

# Define schema for streaming data
schema = StructType() \
    .add("user_id", StringType()) \
    .add("event_type", StringType()) \
    .add("event_time", TimestampType())

# Read from a Kafka stream (or any supported source)
streaming_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "user_events") \
    .load() \
    .selectExpr("CAST(value AS STRING) as json") \
    .selectExpr("from_json(json, 'user_id STRING, event_type STRING, event_time TIMESTAMP') as data") \
    .select("data.*")

# Apply real-time filter: only 'click' events
filtered_df = streaming_df.filter(col("event_type") == "click")

# Handle late data with watermarking (e.g., 10 minutes tolerance)
filtered_with_watermark = filtered_df \
    .withWatermark("event_time", "10 minutes")

# Write to console (for demo purposes)
query = filtered_with_watermark.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

query.awaitTermination()Altyapı:

Bu örnekte:

• Gerçek zamanlı filtreleme .filter(col("event_type") == "click") ile yapılır.
• Structured Streaming, gelen Kafka verisini kesintisiz işler.
• Su damgalama (withWatermark(...)), geç gelen olayların bir zaman penceresi içinde (bu örnekte 10 dakika) ele alınmasını sağlar.

Gerçek zamanlı filtreler ile su damgalamayı birleştirmek, ağ gecikmeleri veya düzensiz olaylar durumunda bile güvenilir akış uygulamaları oluşturmanıza olanak tanır. 

Makine öğrenimi için veri hazırlama  

Makine öğrenimi hatlarında, etkili filtreleme ön işleme sırasında kritiktir. Yalnızca gürültüyü gidermek için değil, model eğitimi için en ilgili özellikleri seçmek amacıyla da kullanılır. Filtreleme; aykırı değerleri elemek, eksik veya geçersiz değerlere sahip kayıtları düşürmek ve anlamlı sinyaller sunan öznitelikleri tutmak için yardımcı olur.

Veriyi temizleme ve temel özellikleri seçme şeklindeki bu ikili yaklaşım, PySpark ile Veri Temizleme kursumuzda da ele alındığı gibi, modellerinizin hem performansını hem güvenilirliğini ciddi ölçüde iyileştirebilir. Aksi halde, modeller alakasız girdiler tarafından saptırılabilir veya kötü veriler nedeniyle bozulabilir.

PySpark ile Büyük Veri kursumuz, PySpark’ta makine öğrenimine dair daha fazla ayrıntıya iner. 

En İyi Uygulamalar ve Yaygın Zorluklar  

Bu bölümde, PySpark ile veri filtrelerken karşılaşılan en iyi uygulamaları ve yaygın zorlukları inceleyeceğiz. 

Performans ayarlama kontrol listesi  

Filtreleme verimliliğini artırmak için bazı pratik öneriler:

1. İşleme sırasında mümkün olduğunca erken filtreleyin.  

2. Veri taşımayı (shuffle) azaltmak için seçilen sütun sayısını sınırlayın.  

3. Özel UDF’ler (kullanıcı tanımlı fonksiyonlar) yerine öncelikle PySpark’ın yerleşik filtre fonksiyonlarını kullanın.  

Yaygın tuzaklar

Şu tuzaklardan kaçının:

1. Optimizatör performansını zedeleyen karmaşık veya gereksiz yere iç içe filtre koşulları

Aşırı girift filtre ifadeleri, Spark’ın Catalyst enoptimizörünü sekteye uğratabilir. Örneğin, çok sayıda OR koşulunu zincirlemek veya derinlemesine iç içe mantıksal ifadeler mantıksal planı karmaşıklaştırarak optimal olmayan yürütmeye yol açabilir. Birden fazla eşitlik kontrolünün söz konusu olduğu durumlarda isin() kullanmayı ya da daha büyük veri kümeleri için daha verimli olan yayın (broadcast) join’lerinden yararlanmak amacıyla küçük bir DataFrame ile join yapmayı düşünün.

2. Aşırı önbelleğe alma yanlış kullanılırsa bellek sorunlarına yol açabilir

Önbelleğe alma, DataFrame’leri yeniden kullanmak için faydalıdır; ancak gelişigüzel kullanım küme belleğini tüketebilir. Yeniden kullanımını değerlendirmeden büyük veya çok sayıda DataFrame’i önbelleğe almak bellek taşmalarına neden olabilir. Yalnızca birden çok kez yeniden kullanılan DataFrame’leri önbelleğe almak ve ihtiyaç kalmadığında bunların kalıcılığını kaldırmak (unpersist) kritik önemdedir.

3. Çarpık veri dağılımlarını görmezden gelmek filtrelemeyi yavaşlatabilir

Bazı anahtarların orantısız derecede fazla veriye sahip olduğu veri çarpıklığı, performans darboğazlarına yol açabilir. Çarpık anahtarlar üzerinde yapılan join veya toplulaştırma gibi işlemler, görev yürütme sürelerinde dengesizliğe neden olabilir. Anahtarlara rastgele önekler ekleme (salting), özel bölümleme veya Spark’ın Uyarlanabilir Sorgu Yürütmesi (Adaptive Query Execution) gibi teknikler bu sorunları hafifletmeye yardımcı olabilir.

Sonuç  

Doğru filtreleme, PySpark ile verimli veri işlemeye kritik katkı sağlar. Analiz iş akışlarını sadeleştirir, hesaplama kaynaklarını korur ve sonraki analitik çıktılara netlik kazandırır. PySpark’ın basit ama güçlü filtreleme yöntemlerinden ve predicate pushdown, bölüm kırpma ve seçici önbelleğe alma gibi optimize stratejilerden yararlanmak, veri mühendisliği uygulamalarınızı önemli ölçüde güçlendirir. Bu teknikleri tutarlı biçimde uygulamak, sağlam ve sürdürülebilir veri hatları sağlar.

PySpark’ı daha fazla keşfetmek için derinlemesine Spark kurslarımızı inceleyin; bunlar arasında şunlar yer alır:

• Introduction to PySpark
• Foundations of PySpark 
• Big Data Fundamentals with PySpark
• Building Recommendation Engines with PySpark