Tutorial sul filtro in PySpark: tecniche, consigli sulle performance e casi d’uso

Il filtraggio è un’operazione fondamentale in PySpark, essenziale per affinare rapidamente grandi dataset e concentrare le informazioni rilevanti. Usare i filtri in modo efficace in PySpark può migliorare l’efficienza del flusso di lavoro per data engineer intermedi, data scientist e sviluppatori alle prese con l’elaborazione di big data. 

Questo articolo introduce tecniche di filtraggio PySpark di base e avanzate, illustra strategie di ottimizzazione per ottenere prestazioni migliori e copre casi d’uso pratici applicabili a scenari reali.

Questo tutorial presuppone alcune conoscenze fondamentali di PySpark, ma puoi scoprire cos’è PySpark e come può essere usato nel nostro tutorial introduttivo su PySpark. 

Che cos’è l’operazione di filtro in PySpark?

Come menzionato nella nostra guida Impara PySpark da zero nel 2025, PySpark è uno strumento ampiamente utilizzato nella data science per facilità d’uso ed efficienza. Offre modi semplici per filtrare i dataset in modo efficiente con funzioni integrate come filter() e where(). Queste funzioni aiutano i professionisti dei dati a isolare le righe dei DataFrame che soddisfano condizioni specificate.  

Il filtraggio è fondamentale nel preprocessing, nell’analisi e nell’ottimizzazione delle pipeline. Ti permette di escludere in anticipo record irrilevanti o errati, risparmiando tempo, riducendo i costi e migliorando le analisi o le attività di modellazione successive.

Nonostante la sua semplicità, applicare filtri in ambienti distribuiti può creare sfide come colli di bottiglia prestazionali, condizioni di filtro inefficienti o complesse e una gestione non corretta dei dati mancanti o incompleti.

Comprendere i filtri in Spark è vitale per superare un colloquio da data engineer. 

Padroneggia gli argomenti chiave e le domande dei colloqui sui big data, dai concetti di base come l’archiviazione dei dati e il calcolo distribuito ad aree avanzate come machine learning e sicurezza, con il nostro tutorial Top 30+ domande per colloqui sui Big Data. 

Fondamenti del filtraggio in PySpark

Il filtraggio in PySpark opera all’interno del suo modello di esecuzione distribuito, basato su Apache Spark. Quando applichi un filtro (ad es., usando .filter() o .where()), PySpark non scansiona immediatamente i dati. Invece, costruisce un piano di query logico che delinea le trasformazioni che vuoi eseguire.

Due concetti chiave influenzano in modo significativo l’efficienza del filtraggio in PySpark:

• Valutazione pigra (Lazy Evaluation) e
• Catalyst Optimizer.  

Lazy evaluation

PySpark non esegue immediatamente un comando di filtro; al contrario, costruisce un piano di query logico e rimanda l’esecuzione finché non gli viene richiesto esplicitamente. Questo modello consente a PySpark maggiore flessibilità nell’ottimizzare le operazioni lungo le pipeline dei dati.  

Catalyst optimizer

Questo motore di ottimizzazione in Spark valuta i piani logici, incluso il filtraggio, e identifica automaticamente il modo più veloce ed efficiente per eseguirli.

Quando inizia l’esecuzione, i dati filtrati vengono elaborati in parallelo sui nodi worker usando RDD o DataFrame. Ogni partizione applica il filtro in modo indipendente, rendendo il processo scalabile per grandi dataset.

Puoi approfondire gli RDD nel nostro PySpark Cheat Sheet: Spark in Python. 

Sintassi base del filtro in PySpark  

Filtrare i dati in base a condizioni è semplice. Puoi fornire condizioni come espressioni di colonna o stringhe in stile SQL. 

1. Espressioni di colonna

Usa oggetti colonna del DataFrame e operatori per definire le condizioni:

filtered_data = df.filter(df.age > 25)Offerto da

2. Stringhe in stile SQL

Scrivi la condizione come una stringa in stile SQL:

filtered_data_2 = df.filter("age > 25")Offerto da

In alternativa, puoi usare la funzione where() allo stesso modo:

filtered_data_3 = df.where(df.age > 25)
filtered_data_4 = df.where("age > 25")Offerto da

Tutte queste opzioni sono valide e producono lo stesso risultato. La scelta tra loro spesso dipende dalle preferenze personali o da quale stile risulta più leggibile nel tuo contesto.

Per saperne di più sull’uso della sintassi SQL in PySpark, consulta il nostro corso Introduzione a Spark SQL in Python. 

Tecniche di filtraggio in PySpark e applicazioni

Esploriamo ora vari metodi di filtraggio e i loro casi d’uso pratici.

Filtri relazionali  

Puoi filtrare usando operatori relazionali come maggiore di >, minore di <, uguale a == e altri per isolare dati in base a soglie numeriche o categoriali:

greater_than_df = df.filter(df.salary > 50000)  
category_match_df = df.filter(df.country == 'France')Offerto da

Filtri logici  

Per combinare più condizioni in un unico filtro, gli operatori logici come AND &, OR | e NOT ~ permettono di creare espressioni concise per requisiti complessi:

combined_df = df.filter((df.age > 25) & (df.city == 'New York'))Offerto da

Funzioni di filtraggio avanzate  

PySpark include funzioni avanzate per casi di filtraggio specializzati. Vediamone alcune. 

Test di appartenenza: per filtrare valori da un set predefinito usando isin:

selected_countries_df = df.filter(df.country.isin('Japan', 'Brazil', 'India'))Offerto da

Corrispondenza di pattern: filtraggio basato su pattern testuali usando like per corrispondenze semplici e rlike  per regex:

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

Filtri di intervallo: filtraggio con metodi pratici come between:

range_df = df.filter(df.age.between(25, 35))Offerto da

Verifiche di null: gestire i valori mancanti in modo corretto controllando esplicitamente i null:

valid_email_df = df.filter(df.email.isNotNull())Offerto da

Filtrare tipi di dati complessi

Le organizzazioni usano spesso dati annidati o strutturati. Il filtraggio in PySpark supporta campi annidati e array. Per i dati strutturati (struct), filtra facilmente i campi annidati:

city_df = df.filter(df.address.city == 'Boston')Offerto da

Per i dati archiviati come array, puoi utilizzare funzioni integrate come array_contains:

from pyspark.sql.functions import array_contains  

skill_df = df.filter(array_contains(df.skills, 'Python'))Offerto da

Tecniche di ottimizzazione per un filtraggio efficiente in PySpark

In questa sezione scoprirai le strategie per migliorare le prestazioni delle operazioni di filtro in PySpark.

Predicate pushdown  

Il predicate pushdown consiste nel posizionare le condizioni di filtraggio il più vicino possibile al livello di archiviazione dei dati. Ciò significa che PySpark legge meno dati inizialmente, migliorando le prestazioni.

Per esempio: 

# 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)Offerto da

Nell’esempio sopra, applichiamo un filtro su una colonna normale. Normalmente, Spark caricherebbe l’intero dataset e poi filtrerebbe in memoria, ma poiché Parquet supporta il predicate pushdown, Spark può inviare la condizione di filtro direttamente al lettore Parquet.

Invece di leggere tutte le righe e poi controllare product_category, Spark legge solo le righe in cui product_category = 'Electronics'. Questo riduce l’I/O su disco e la memoria, accelerando l’elaborazione.

explain stampa il piano di query fisico, mostrando esattamente cosa prevede di fare Spark durante l’esecuzione.

Se il predicate pushdown è applicato, vedrai qualcosa come:

PushedFilters: [EqualTo(product_category,Electronics)]Offerto da

Questo conferma che Spark ha spinto il filtro fino al livello di scansione dei file invece di applicarlo dopo il caricamento dei dati.

Il predicate pushdown è una delle ottimizzazioni più potenti nella toolkit di Spark. Esso:

• Riduce la quantità di dati letti da disco.
• Accelera l’esecuzione delle query.
• Riduce l’uso di memoria e l’I/O di rete.

Scrivendo i filtri in modo chiaro e archiviando i dati in formati come Parquet, consenti a Spark di ottimizzare l’esecuzione dietro le quinte.

L’ottimizzazione delle prestazioni è cruciale anche quando si eseguono join in PySpark. Scopri come ottimizzare i join in PySpark, ridurre gli shuffle, gestire lo skew e migliorare le prestazioni nelle pipeline di big data e nei workflow di machine learning con il nostro tutorial sui join in PySpark. 

Partition pruning  

Quando i dati sono partizionati per campi rilevanti (ad esempio, la data), PySpark può evitare di caricare partizioni non necessarie. Un partition pruning efficace riduce significativamente le scansioni di dati e i tempi di esecuzione.

Supponiamo di avere un dataset Parquet partizionato per year e month, archiviato in s3://your-bucket/events_data/. 

Ogni partizione appare così:

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

...

events_data/year=2024/month=01/Offerto da

Un esempio di partition pruning in PySpark sarebbe:

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)Offerto da

year e month sono colonne di partizione, consentendo al Catalyst optimizer di Spark di potare tutte le partizioni irrilevanti. 

Caching e persistenza  

I risultati filtrati spesso vengono riutilizzati più volte in un flusso di analisi. Conservare temporaneamente i risultati filtrati in memoria tramite caching aiuta a evitare ricalcoli ripetuti, ma va usato con cautela per preservare le risorse.

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

recent_users_df.cache()Offerto da

Strategie di filtraggio avanzate e casi d’uso  

Esploriamo ora scenari complessi in cui si applicano tecniche di filtraggio avanzate.

Filtraggio basato sul tempo  

Per la gestione dei dati time series, è spesso richiesto un filtraggio temporale preciso. Filtrare i dati entro un intervallo di tempo specifico è semplice:

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

L’esempio sopra seleziona tutti i record di agosto 2023. La condizione funziona sia con colonne di tipo DateType che TimestampType.

Per un’analisi time series più approfondita, come il calcolo di metriche mobili o l’identificazione di trend nel tempo, puoi combinare filtri temporali con le window function. Per esempio, puoi prima filtrare un intervallo di date, poi applicare una media mobile o classificare gli eventi all’interno di ogni giorno:

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))Offerto da

Questo esempio calcola una media mobile su 3 righe al giorno per valori ordinati per timestamp. Combinare filtri temporali con le window function ti dà un controllo granulare sulle analisi basate sul tempo.

Filtraggio nei dati in streaming  

Filtrare flussi di dati in tempo reale presenta sfide uniche. A differenza del batch processing, i dati sono illimitati e arrivano continuamente. Ciò richiede che i filtri vengano applicati al volo, con una gestione accurata degli eventi in ritardo per garantire risultati corretti.

Structured Streaming di PySpark offre strumenti intuitivi per filtrare dati in streaming e gestire efficacemente i problemi di temporizzazione. Puoi applicare i filtri esattamente come faresti su un DataFrame statico, usando al contempo watermark per controllare per quanto tempo il sistema deve aspettare i dati in ritardo prima di finalizzare i risultati.

Ecco un esempio:

# 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()Offerto da

Nell’esempio:

• Il filtraggio in tempo reale è eseguito con .filter(col("event_type") == "click").
• Structured Streaming gestisce continuamente i dati Kafka in arrivo.
• Il watermarking (withWatermark(...)) garantisce che gli eventi in ritardo vengano gestiti entro una finestra temporale (10 minuti in questo caso).

Combinare filtri in tempo reale e watermarking ti consente di costruire applicazioni di streaming affidabili anche in presenza di ritardi di rete o eventi irregolari. 

Preparazione dei dati per il machine learning  

Nelle pipeline di machine learning, un filtraggio efficace è cruciale durante il preprocessing. Non serve solo a rimuovere il rumore, ma anche a selezionare le feature più rilevanti per l’addestramento del modello. Il filtraggio aiuta a eliminare outlier, scartare record con valori mancanti o non validi e mantenere solo gli attributi che apportano segnali significativi.

Questo approccio in due fasi, pulizia dei dati e selezione delle feature chiave, può migliorare drasticamente sia le prestazioni sia l’affidabilità dei tuoi modelli, come trattato nel nostro corso Cleaning Data with PySpark. Senza di esso, i modelli rischiano di essere distorti da input irrilevanti o dati di bassa qualità.

Il nostro corso Big Data with PySpark entra più nel dettaglio del machine learning in PySpark. 

Best practice e sfide comuni  

In questa sezione, esploreremo best practice e sfide comuni riscontrate durante il filtraggio dei dati con PySpark. 

Checklist per l’ottimizzazione delle prestazioni  

Ecco alcune raccomandazioni pratiche per migliorare l’efficienza del filtraggio:

1. Filtra il prima possibile durante l’elaborazione.  

2. Limita il numero di colonne selezionate per ridurre lo shuffling dei dati.  

3. Dai priorità alle funzioni di filtro integrate di PySpark invece delle UDF (user-defined functions) personalizzate.  

Errori comuni

Evita questi errori frequenti:

1. Condizioni di filtro complesse o inutilmente annidate che danneggiano le prestazioni dell’optimizer

Espressioni di filtro eccessivamente intricate possono ostacolare il Catalyst optimizer di Spark. Ad esempio, concatenare numerose condizioni OR o annidare profondamente espressioni logiche può complicare il piano logico, portando a un’esecuzione subottimale. In scenari con più verifiche di uguaglianza, valuta l’uso di isin() o l’esecuzione di un join con un DataFrame più piccolo per sfruttare i broadcast join, più efficienti per grandi dataset.

2. Caching eccessivo può causare problemi di memoria se usato in modo improprio

Sebbene il caching sia utile per riutilizzare i DataFrame, un uso indiscriminato può esaurire la memoria del cluster. Mettere in cache DataFrame grandi o numerosi senza valutarne il riutilizzo può portare a errori di memoria insufficiente. È fondamentale mettere in cache solo i DataFrame riutilizzati più volte e rimuoverli (unpersist) quando non servono più.

3. Ignorare distribuzioni di dati sbilanciate può ridurre la velocità del filtraggio

Lo skew dei dati, in cui alcune chiavi hanno quantità di dati sproporzionate, può causare colli di bottiglia. Operazioni come join o aggregazioni su chiavi sbilanciate possono portare a tempi di esecuzione dei task non uniformi. Tecniche come il salting (aggiungere prefissi casuali alle chiavi), il partitioning personalizzato o l’uso dell’Adaptive Query Execution di Spark possono aiutare a mitigare questi problemi.

Conclusione  

Un filtraggio corretto svolge un ruolo cruciale nell’elaborazione efficiente dei dati con PySpark. Snellisce i workflow analitici, risparmia risorse computazionali e porta chiarezza ai risultati analitici successivi. Sfruttare metodi di filtraggio semplici ma potenti di PySpark, insieme a strategie ottimizzate come predicate pushdown, partition pruning e caching selettivo, migliora significativamente le tue pratiche di data engineering. Applicare queste tecniche in modo coerente garantisce pipeline di dati robuste e manutenibili.

Per esplorare ulteriormente PySpark, dai un’occhiata ai nostri corsi approfonditi su Spark, tra cui:

• Introduzione a PySpark
• Fondamenti di PySpark 
• Fondamenti di Big Data con PySpark
• Creare motori di raccomandazione con PySpark