Tutorial Filter PySpark: Teknik, Tips Performa, dan Contoh Penggunaan

Filtering adalah operasi mendasar di PySpark, penting untuk dengan cepat menyaring dataset besar guna mempersempit informasi yang relevan. Menggunakan filter secara efektif di PySpark dapat meningkatkan efisiensi alur kerja bagi data engineer tingkat menengah, data scientist, dan developer yang menangani pemrosesan big data. 

Artikel ini memperkenalkan teknik filtering PySpark yang fundamental dan lanjutan, menguraikan strategi optimasi untuk performa yang lebih baik, serta membahas contoh penggunaan praktis yang berlaku pada skenario dunia nyata.

Tutorial ini mengasumsikan Anda memiliki pengetahuan dasar PySpark, tetapi Anda dapat mempelajari apa itu PySpark dan bagaimana menggunakannya dalam tutorial Getting Started with PySpark kami. 

Apa itu Operasi Filter PySpark?

Seperti disebutkan dalam panduan Learn PySpark From Scratch in 2025 kami, PySpark banyak digunakan dalam data science karena kemudahan penggunaan dan efisiensinya. PySpark menyediakan cara sederhana untuk memfilter dataset secara efisien menggunakan fungsi bawaan seperti filter() dan where(). Fungsi-fungsi ini membantu para profesional data mengisolasi baris dari DataFrame yang memenuhi kondisi tertentu.  

Filtering sangat penting dalam prapemrosesan data, analisis, dan optimasi pipeline. Ini memungkinkan Anda mengecualikan catatan yang tidak relevan atau salah sedini mungkin, menghemat waktu, menurunkan biaya, dan meningkatkan analisis atau tugas pemodelan berikutnya.

Meskipun sederhana, penerapan filter di lingkungan terdistribusi dapat menimbulkan tantangan seperti bottleneck performa, kondisi filter yang tidak efisien atau kompleks, dan penanganan data hilang atau tidak lengkap yang kurang tepat.

Memahami Filter Spark penting untuk sukses dalam wawancara data engineering. 

Kuasai topik kunci dan pertanyaan yang sering ditanyakan dalam wawancara big data, dari konsep dasar seperti penyimpanan data dan komputasi terdistribusi hingga area lanjut seperti machine learning dan keamanan, melalui tutorial Top 30+ Big Data Interview Questions kami. 

Dasar-Dasar Filtering di PySpark

Filtering di PySpark beroperasi dalam kerangka model eksekusi terdistribusinya, yang dibangun di atas Apache Spark. Saat Anda menerapkan filter (misalnya menggunakan .filter() atau .where()), PySpark tidak langsung memindai data. Sebagai gantinya, PySpark membangun rencana kueri logis yang menguraikan transformasi yang ingin Anda lakukan.

Dua konsep inti sangat memengaruhi efisiensi filtering di PySpark:

• Evaluasi Malas (Lazy Evaluation) dan
• Catalyst Optimizer.  

Lazy evaluation

PySpark tidak langsung mengeksekusi perintah filter; alih-alih, PySpark membangun rencana kueri logis dan menunda aksi hingga diminta secara eksplisit untuk mengeksekusi. Model ini memberi PySpark fleksibilitas lebih besar dalam mengoptimalkan operasi di seluruh pipeline data.  

Catalyst optimizer

Mesin optimasi di Spark ini mengevaluasi rencana logis, termasuk filtering, dan secara otomatis mengidentifikasi cara tercepat dan paling efisien untuk mengeksekusinya.

Saat eksekusi dimulai, data yang terfilter diproses secara paralel di seluruh node pekerja menggunakan RDD atau DataFrame. Setiap partisi menerapkan filter secara mandiri, sehingga dapat diskalakan untuk dataset besar.

Anda dapat mempelajari lebih lanjut tentang RDD dalam tutorial PySpark Cheat Sheet: Spark in Python kami. 

Sintaks Inti Filter PySpark  

Memfilter data berdasarkan kondisi itu mudah. Anda dapat memberikan kondisi sebagai ekspresi kolom atau string bergaya SQL. 

1. Ekspresi kolom

Gunakan objek kolom DataFrame dan operator untuk mendefinisikan kondisi:

filtered_data = df.filter(df.age > 25)Didukung oleh

2. String bergaya SQL

Tulis kondisi sebagai string bergaya SQL:

filtered_data_2 = df.filter("age > 25")Didukung oleh

Sebagai alternatif, Anda dapat menggunakan fungsi where() dengan cara yang sama:

filtered_data_3 = df.where(df.age > 25)
filtered_data_4 = df.where("age > 25")Didukung oleh

Semua ini valid dan menghasilkan hasil yang sama. Pemilihannya sering bergantung pada preferensi pribadi atau gaya mana yang terasa lebih mudah dibaca dalam konteks Anda.

Untuk mempelajari lebih lanjut tentang penggunaan sintaks SQL di PySpark, lihat kursus Introduction to Spark SQL in Python kami. 

Teknik dan Aplikasi Filtering PySpark

Sekarang, mari jelajahi berbagai metode filtering dan contoh penggunaannya secara praktis.

Filter relasional  

Anda dapat memfilter menggunakan operator relasional seperti lebih besar dari >, lebih kecil dari <, sama dengan ==, dan lainnya untuk mengisolasi data berdasarkan ambang batas numerik atau kategorikal:

greater_than_df = df.filter(df.salary > 50000)  
category_match_df = df.filter(df.country == 'France')Didukung oleh

Filter logis  

Untuk menggabungkan beberapa kondisi dalam satu filter, operator logis seperti AND &, OR |, dan NOT ~ dapat membentuk ekspresi ringkas untuk kebutuhan yang kompleks:

combined_df = df.filter((df.age > 25) & (df.city == 'New York'))Didukung oleh

Fungsi Filtering Lanjutan  

PySpark menyertakan fungsi lanjutan untuk kasus filtering khusus. Mari kita telusuri beberapa di antaranya. 

Uji Keanggotaan: Memfilter nilai dari himpunan yang telah ditentukan menggunakan isin:

selected_countries_df = df.filter(df.country.isin('Japan', 'Brazil', 'India'))Didukung oleh

Pencocokan Pola: Filtering berdasarkan pola teks menggunakan like untuk kecocokan sederhana dan rlike  untuk regex:

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

Filter Rentang: Filtering dengan metode praktis seperti between:

range_df = df.filter(df.age.between(25, 35))Didukung oleh

Pemeriksaan Null: Menangani nilai hilang secara tepat dengan memeriksa null secara eksplisit:

valid_email_df = df.filter(df.email.isNotNull())Didukung oleh

Memfilter Tipe Data Kompleks

Organisasi sering menggunakan data bertingkat atau terstruktur. Filtering PySpark mengakomodasi field bersarang dan array. Untuk data terstruktur (struct), filter field bersarang dengan mudah:

city_df = df.filter(df.address.city == 'Boston')Didukung oleh

Untuk data yang disimpan sebagai array, Anda dapat memanfaatkan fungsi bawaan seperti array_contains:

from pyspark.sql.functions import array_contains  

skill_df = df.filter(array_contains(df.skills, 'Python'))Didukung oleh

Teknik Optimasi untuk Filtering PySpark yang Efisien

Di bagian ini, Anda akan mempelajari strategi untuk meningkatkan performa operasi filter di PySpark.

Predicate pushdown  

Predicate pushdown melibatkan penempatan kondisi filter sedekat mungkin dengan level penyimpanan data. Ini berarti PySpark membaca lebih sedikit data sejak awal, sehingga meningkatkan performa.

Sebagai contoh: 

# 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)Didukung oleh

Pada contoh di atas, kita menerapkan filter pada kolom biasa. Biasanya, Spark akan memuat seluruh dataset lalu memfilter di memori, tetapi karena Parquet mendukung predicate pushdown, Spark dapat mengirimkan kondisi filter langsung ke pembaca Parquet.

Alih-alih membaca semua baris lalu memeriksa product_category, Spark hanya membaca baris dengan product_category = 'Electronics'. Ini menghemat I/O disk dan memori, mempercepat pemrosesan.

explain mencetak rencana kueri fisik, menunjukkan secara tepat apa yang akan dilakukan Spark selama eksekusi.

Jika predicate pushdown diterapkan, Anda akan melihat sesuatu seperti:

PushedFilters: [EqualTo(product_category,Electronics)]Didukung oleh

Ini menegaskan bahwa Spark mendorong filter ke level pemindaian file alih-alih menerapkannya setelah memuat data.

Predicate pushdown adalah salah satu optimasi paling kuat di Spark. Ini:

• Mengurangi jumlah data yang dibaca dari disk.
• Mempercepat eksekusi kueri.
• Menurunkan penggunaan memori dan I/O jaringan.

Dengan menulis filter Anda secara jelas dan menyimpan data dalam format seperti Parquet, Anda memungkinkan Spark mengoptimalkan eksekusi di balik layar.

Optimasi performa juga penting saat melakukan join di PySpark. Pelajari cara mengoptimalkan join PySpark, mengurangi shuffle, menangani skew, dan meningkatkan performa di seluruh pipeline big data dan alur kerja machine learning melalui tutorial PySpark Joins: Optimize Big Data Join Performance kami. 

Pemangkasan partisi (partition pruning)  

Saat data dipartisi berdasarkan field yang relevan (misalnya, tanggal), PySpark dapat menghindari pemuatan partisi yang tidak diperlukan. Pemangkasan partisi yang efektif secara signifikan mengurangi pemindaian data dan waktu eksekusi.

Misalkan kita memiliki dataset Parquet yang dipartisi berdasarkan year dan month, disimpan di s3://your-bucket/events_data/. 

Setiap partisi terlihat seperti:

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

...

events_data/year=2024/month=01/Didukung oleh

Contoh pemangkasan partisi di PySpark akan tampak seperti ini:

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)Didukung oleh

year dan month adalah kolom partisi, memungkinkan Catalyst optimizer Spark untuk memangkas semua partisi yang tidak relevan. 

Caching dan persistensi  

Hasil terfilter sering digunakan berulang kali dalam alur analisis. Menyimpan sementara hasil terfilter di memori melalui caching membantu menghindari komputasi berulang, tetapi harus digunakan dengan hati-hati untuk menghemat sumber daya.

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

recent_users_df.cache()Didukung oleh

Strategi Filtering Lanjutan dan Contoh Penggunaan  

Sekarang, mari jelajahi skenario kompleks tempat teknik filtering lanjutan diterapkan.

Filtering berbasis waktu  

Untuk menangani data deret waktu, filtering temporal yang presisi sering diperlukan. Memfilter data dalam rentang waktu tertentu itu mudah:

time_df = df.filter((df.date >= '2023-08-01') & (df.date < '2023-09-01'))Didukung oleh

Contoh di atas memilih semua catatan dari Agustus 2023. Kondisi ini berfungsi baik untuk kolom DateType maupun TimestampType.

Untuk analisis deret waktu yang lebih mendalam, seperti menghitung metrik rolling atau mengidentifikasi tren dari waktu ke waktu, Anda dapat memasangkan filter temporal dengan window function. Misalnya, Anda bisa terlebih dahulu memfilter rentang tanggal, lalu menerapkan rata-rata bergerak atau memberi peringkat kejadian dalam setiap hari:

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))Didukung oleh

Contoh ini menghitung rata-rata bergerak 3 baris per hari untuk nilai yang diurutkan berdasarkan timestamp. Mengombinasikan filter temporal dengan window function memberi Anda kontrol yang lebih halus atas analitik berbasis waktu.

Filtering pada data streaming  

Filtering aliran data real-time memiliki tantangan tersendiri. Berbeda dengan pemrosesan batch, datanya tak terbatas dan tiba secara terus-menerus. Ini membutuhkan penerapan filter secara langsung, dengan penanganan yang cermat untuk kejadian yang terlambat tiba agar hasil tetap akurat.

Structured Streaming milik PySpark menyediakan alat intuitif untuk memfilter data streaming dan mengelola masalah penjadwalan dengan efektif. Anda dapat menerapkan filter seperti pada DataFrame statis sambil menggunakan watermark untuk mengontrol berapa lama sistem harus menunggu data terlambat sebelum memfinalisasi hasil.

Berikut contohnya:

# 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()Didukung oleh

Dalam contoh tersebut:

• Filtering real-time dilakukan dengan .filter(col("event_type") == "click").
• Structured Streaming menangani data Kafka yang masuk secara kontinu.
• Watermarking (withWatermark(...)) memastikan kejadian yang terlambat tiba ditangani dalam jendela waktu (10 menit pada kasus ini).

Menggabungkan filter real-time dan watermarking memungkinkan Anda membangun aplikasi streaming yang andal bahkan saat terjadi keterlambatan jaringan atau kejadian yang tidak teratur. 

Persiapan data machine learning  

Dalam pipeline machine learning, filtering yang efektif sangat penting selama prapemrosesan. Filtering tidak hanya digunakan untuk menghapus noise tetapi juga untuk memilih fitur yang paling relevan untuk pelatihan model. Filtering membantu menghilangkan outlier, menghapus catatan dengan nilai yang hilang atau tidak valid, dan mempertahankan hanya atribut yang memberikan sinyal bermakna.

Pendekatan dua langkah ini—membersihkan data dan memilih fitur kunci—dapat secara drastis meningkatkan performa dan keandalan model Anda, sebagaimana dibahas dalam kursus Cleaning Data with PySpark kami. Tanpanya, model berisiko bias oleh masukan yang tidak relevan atau terdistorsi oleh data buruk.

Kursus Big Data with PySpark kami membahas lebih rinci tentang machine learning di PySpark. 

Praktik Terbaik dan Tantangan Umum  

Di bagian ini, kita akan membahas praktik terbaik dan tantangan umum yang ditemui saat memfilter data dengan PySpark. 

Daftar periksa penalaan performa  

Berikut beberapa rekomendasi praktis untuk meningkatkan efisiensi filtering:

1. Lakukan filter sedini mungkin selama pemrosesan.  

2. Batasi jumlah kolom yang dipilih untuk mengurangi data shuffling.  

3. Prioritaskan penggunaan fungsi filter bawaan PySpark alih-alih UDF (user-defined function) kustom.  

Kesalahan umum

Hindari kesalahan umum berikut:

1. Kondisi filter yang kompleks atau bersarang secara tidak perlu, merugikan performa optimizer

Ekspresi filter yang terlalu rumit dapat menghambat Catalyst optimizer Spark. Misalnya, merangkai banyak kondisi OR atau pernyataan logis yang sangat bersarang dapat mempersulit rencana logis, sehingga menghasilkan eksekusi yang suboptimal. Dalam skenario dengan banyak pengecekan kesetaraan, pertimbangkan menggunakan isin() atau melakukan join dengan DataFrame yang lebih kecil untuk memanfaatkan broadcast join, yang lebih efisien untuk dataset besar.

2. Caching berlebihan dapat menyebabkan masalah memori jika disalahgunakan

Meskipun caching bermanfaat untuk menggunakan kembali DataFrame, penggunaan tanpa pandang bulu dapat menguras memori klaster. Melakukan cache pada DataFrame yang besar atau banyak tanpa menilai pemakaiannya kembali dapat menyebabkan error kehabisan memori. Sangat penting untuk hanya melakukan cache pada DataFrame yang digunakan berulang kali dan melakukan unpersist saat tidak lagi diperlukan.

3. Mengabaikan distribusi data yang skew dapat memperlambat filtering

Skew data, yaitu saat kunci tertentu memiliki jumlah data yang tidak proporsional besar, dapat menyebabkan bottleneck performa. Operasi seperti join atau agregasi pada kunci skew dapat menyebabkan waktu eksekusi tugas yang tidak merata. Teknik seperti salting (menambahkan prefiks acak pada kunci), partisi kustom, atau menggunakan Adaptive Query Execution Spark dapat membantu mengurangi masalah ini.

Kesimpulan  

Filtering yang tepat memainkan peran krusial dalam pemrosesan data yang efisien dengan PySpark. Filtering merampingkan alur analisis, menghemat sumber daya komputasi, dan memberikan kejelasan pada hasil analitis berikutnya. Memanfaatkan metode filtering PySpark yang sederhana namun kuat, bersama strategi yang dioptimalkan seperti predicate pushdown, pemangkasan partisi, dan caching selektif, secara signifikan meningkatkan praktik data engineering Anda. Penerapan teknik ini secara konsisten memastikan pipeline data yang andal dan mudah dipelihara.

Untuk mengeksplorasi PySpark lebih jauh, lihat kursus Spark mendalam kami, termasuk:

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