Come usare efficacemente le UDF in PySpark e le Pandas UDF

Le User-Defined Functions (UDF) in PySpark offrono agli sviluppatori Python un modo potente per gestire attività specifiche che le funzioni integrate di Spark non possono coprire. Se sei un data engineer, analyst o data scientist con esperienza in Python, comprendere i concetti di UDF ti permette di affrontare in modo efficace sfide complesse e reali sui dati. 

Questo tutorial ti guida attraverso i concetti delle UDF in PySpark, implementazioni pratiche e best practice per ottimizzazione, test, debug e pattern d’uso avanzati. Alla fine, saprai scrivere, ottimizzare e distribuire con sicurezza UDF efficienti su larga scala.

Se sei alle prime armi con PySpark, ti consiglio prima di consultare il nostro tutorial Introduzione a PySpark poiché qui trattiamo concetti avanzati di Spark.

Cosa sono le UDF in PySpark?

Le UDF in PySpark sono funzioni Python personalizzate integrate nel framework distribuito di Spark per operare sui dati archiviati negli Spark DataFrame. A differenza delle funzioni integrate, le UDF consentono di applicare logiche complesse e personalizzate a livello di riga o di colonna. 

La nostra PySpark Cheat Sheet copre tutto ciò che devi sapere sugli Spark DataFrame, rendendo ancora più facile comprendere le UDF in Spark. 

Quando dovresti usare le UDF in PySpark?

Usa le UDF quando:

• Hai bisogno di una logica che non può essere espressa con le funzioni integrate di Spark.
• La tua trasformazione coinvolge operazioni complesse tipiche di Python (ad esempio, manipolazioni regex, logiche NLP personalizzate).
• Sei disposto a scambiare prestazioni con flessibilità, soprattutto in fase di prototipazione o per dataset piccoli/medi.

Evita le UDF quando:

• Esiste un equivalente in pyspark.sql.functions: le funzioni native di Spark sono più veloci, ottimizzate e possono essere inviate al motore di esecuzione.
• Lavori con dataset grandi in cui le prestazioni sono critiche. Le UDF introducono overhead di serializzazione e impediscono a Spark di ottimizzare i piani di esecuzione.
• Puoi esprimere la tua logica con espressioni SQL, funzioni integrate di Spark SQL o Pandas UDF (per operazioni vettorializzate).

Applicazioni strategiche nel data engineering

Ecco i principali casi d’uso delle UDF in PySpark:

• Trasformazioni complesse dei dati, come parsing avanzato del testo, estrazione dati o manipolazione di stringhe.
• Integrazione con librerie Python di terze parti, inclusi framework di machine learning popolari come TensorFlow e XGBoost.
• Colmare sistemi legacy e supportare un’evoluzione fluida dello schema man mano che le strutture dei dati cambiano.

Le UDF semplificano i compiti disordinati e reali del data engineering, consentendo ai team di gestire requisiti diversi in modo flessibile ed efficace. 

Vediamo ora come puoi implementare le UDF in PySpark. 

Implementare le UDF in PySpark

Questa sezione illustra come definire e implementare concretamente le UDF in PySpark.

Metodi standard di dichiarazione delle UDF

Ci sono tre approcci comuni per dichiarare UDF in PySpark:

1. UDF basate su lambda: rapide da definire direttamente nelle query sui DataFrame; ideali per operazioni semplici.

Le UDF basate su lambda (Basic Python UDF) sono ideali per trasformazioni rapide e semplici. Evitale per job su larga scala dove le prestazioni sono importanti.

Ecco un esempio: 

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

2. Funzioni Python con decorator: annotate esplicitamente con @pyspark.sql.functions.udf, a supporto di riusabilità e leggibilità.

Per esempio:

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

3. UDF registrate in SQL: registrate direttamente nei contesti Spark SQL, così da poterle usare nelle query SQL.

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

Ogni metodo ha compromessi: le UDF lambda sono concise ma limitate, mentre le funzioni annotate privilegiano leggibilità, manutenibilità e best practice.

Le Pandas UDF consentono operazioni vettorializzate su batch Arrow. Spesso sono più veloci delle UDF normali e si integrano meglio con il motore di esecuzione di Spark.

Pandas UDF scalare (element-wise, tipo map)

Sono ideali per trasformazioni veloci, riga per riga, su grandi dataset. Per esempio: 

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

Pandas UDF Grouped Map

Sono ideali per logiche personalizzate per gruppo, simili a groupby().apply() in Pandas.  

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

Pandas UDF Aggregate

Esegue aggregazioni per gruppo, più veloce della grouped map. Per esempio: 

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

Pandas Iterator UDF

La Pandas Iterator UDF è ideale per dataset grandi che richiedono elaborazione a basso consumo di memoria (per batch). Per esempio: 

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

Gestione dei tipi e sicurezza rispetto ai null

Tipi e valori null sono sfide frequenti per le UDF in PySpark. PySpark impone un controllo rigoroso dei tipi, causando spesso conversioni implicite o problemi a runtime. Inoltre, Spark passa i valori null direttamente alle UDF, con potenziali crash se non gestiti esplicitamente.

Assicurati UDF robuste con queste strategie:

• Specifica esplicitamente i tipi di ritorno.
• Incorpora controlli sui null (ad esempio, istruzioni condizionali) nelle tue funzioni Python.
• Adotta pratiche di coding difensivo; semplici controlli sui null prevengono fastidiose eccezioni a runtime.

Ottimizzare le prestazioni delle UDF

Le prestazioni risultano spesso il tallone d’Achille delle UDF standard a causa del modello di esecuzione riga per riga. Sfruttare le UDF vettorializzate e gli strumenti di ottimizzazione di Spark migliorerà in modo significativo i tempi di esecuzione.

UDF vettorializzate con integrazione Pandas

Le Pandas UDF introducono un approccio vettorializzato alle UDF in PySpark passando batch di dati come Pandas Series alle funzioni Python. Questo design migliora sensibilmente le prestazioni riducendo l’overhead di serializzazione rispetto alle UDF standard basate su righe.

Supportate da Apache Arrow per il trasferimento dati zero-copy tra JVM e processi Python, le Pandas UDF consentono un’esecuzione efficiente su larga scala. Sono particolarmente efficaci per calcoli intensivi e manipolazioni complesse di stringhe su milioni di record. 

Approfondiamo la manipolazione dei dati con PySpark nel nostro corso Cleaning Data with PySpark. 

Inoltre, le Pandas UDF permettono un’integrazione fluida con l’ecosistema Python per la data science, sfruttando strumenti e workflow familiari.

Tecniche di ottimizzazione con Arrow

Il formato di memoria colonnare di Apache Arrow consente un trasferimento dati efficiente e senza copie tra la JVM di Spark e i processi Python. Abilitando Arrow (spark.sql.execution.arrow.pyspark.enabled=true) nelle configurazioni di Spark, i dati si muovono rapidamente tra JVM e ambiente Python, velocizzando sensibilmente l’esecuzione delle UDF.

Ottimizzazione del piano di esecuzione

Ottimizzare i job PySpark implica capire come influenzare il Catalyst optimizer di Spark. Strategie avanzate includono predicate pushdown, potatura delle colonne e uso di hint per i broadcast join per migliorare la pianificazione e l’efficienza di esecuzione delle query.

Per massimizzare le prestazioni, è importante ridurre al minimo la portata di esecuzione delle UDF e preferire le funzioni integrate di Spark SQL quando possibile. Un uso strategico della cache e una progettazione attenta del piano possono migliorare ulteriormente la velocità di esecuzione e l’utilizzo delle risorse.

L’ottimizzazione delle prestazioni è una delle principali domande che potresti incontrare in un colloquio su PySpark. Scopri come rispondere a questa e ad altre domande su Spark nel nostro post Top 36 domande e risposte su PySpark per il 2025. 

Pattern avanzati e anti-pattern

Comprendere usi corretti e scorretti aiuta a garantire implementazioni di UDF stabili ed efficienti.

Implementazioni di UDF stateful

Le UDF stateful e non deterministiche presentano sfide particolari in PySpark. Queste funzioni producono risultati che dipendono da stato esterno o condizioni variabili, come variabili d’ambiente, ora di sistema o contesto di sessione.

Sebbene le UDF non deterministiche a volte siano necessarie—per esempio per generare timestamp, tracciare sessioni utente o introdurre casualità—possono complicare debug, riproducibilità e ottimizzazione.

Implementare UDF stateful richiede pattern progettuali accurati: documentare chiaramente il comportamento, isolare gli effetti collaterali e aggiungere logging approfondito per facilitare la risoluzione dei problemi e garantire coerenza tra le esecuzioni.

Se usate con criterio, offrono capacità potenti, ma mantenere pipeline dati affidabili richiede una gestione disciplinata. Il nostro corso Big Data Fundamentals with PySpark approfondisce come gestire big data in PySpark. 

Anti-pattern comuni

Gli anti-pattern più comuni nell’uso delle UDF possono degradare significativamente le prestazioni di PySpark:

• Elaborazione riga per riga invece che per batch: applicare UDF a singole righe, invece di approcci vettorializzati come le Pandas UDF, porta a forti rallentamenti.
• Operazioni su DataFrame annidate dentro le UDF: incorporare query su DataFrame all’interno delle UDF causa calcolo eccessivo e ostacola la capacità di Spark di ottimizzare i piani.
• Registrazione ripetuta di UDF inline: definire e registrare UDF più volte nelle query aggiunge overhead inutile; è meglio dichiararle una volta e riutilizzarle tra i job.
• Abusare di logiche Python personalizzate per operazioni semplici: attività come filtri di base, aritmetica o trasformazioni lineari dovrebbero preferire le funzioni integrate e altamente ottimizzate di Spark rispetto a UDF personalizzate.

Evitare queste insidie garantisce migliori prestazioni, facilita l’ottimizzazione da parte di Catalyst e rende il codice PySpark più manutenibile.

Debug e test delle UDF in PySpark

Testare e fare debug delle UDF garantisce affidabilità e robustezza in produzione.

Pattern di gestione delle eccezioni

Implementare una gestione strutturata degli errori all’interno delle UDF è essenziale per costruire pipeline PySpark robuste e manutenibili. Usa blocchi try-except nelle UDF per gestire con grazia imprevisti a runtime, come valori null, mismatch di tipo o divisioni per zero.

Una gestione solida delle eccezioni stabilizza le pipeline contro dati inattesi e semplifica il debug grazie a messaggi di errore chiari e azionabili. Eccezioni catturate e loggate correttamente rendono il comportamento delle UDF più trasparente, accelerando la risoluzione dei problemi e migliorando l’affidabilità complessiva.

Framework per unit test

Usa la classe base di testing integrata in PySpark, pyspark.testing.utils.ReusedPySparkTestCase, insieme a framework come pytest, per scrivere unit test affidabili per le tue UDF. Strutturare test chiari e mirati garantisce correttezza, stabilità e manutenibilità della logica delle UDF nel tempo.

Le best practice per testare le UDF includono coprire casi tipici e limite, validare gli output rispetto a risultati noti e isolare il comportamento della UDF da dipendenze esterne. Test ben progettati proteggono dalle regressioni e semplificano lo sviluppo e il refactoring futuri.

Evoluzione e direzioni future

L’ecosistema PySpark continua a evolversi rapidamente, introducendo nuove funzionalità che migliorano ulteriormente le UDF.

Integrazione con Unity Catalog

Sviluppi recenti hanno integrato la registrazione delle UDF nel Unity Catalog, semplificando la gestione, la scoperta e la governance delle UDF su larga scala. Unity Catalog consente un controllo centralizzato sul ciclo di vita delle UDF, inclusi registrazione, versioning e controllo degli accessi, elementi critici in ambienti enterprise.

Questa integrazione migliora la governance, applica policy di sicurezza coerenti e aumenta la reperibilità tra i team, rendendo le UDF più facili da riutilizzare, auditare e gestire in ecosistemi dati ampi e complessi.

UDF accelerate su GPU

Framework come RAPIDS Accelerator consentono l’offloading su GPU delle UDF computazionalmente intensive in PySpark, offrendo miglioramenti prestazionali notevoli. Spostando su GPU operazioni pesanti, come analisi numerica, inferenza di deep learning e modellazione su larga scala, RAPIDS può ridurre i tempi di esecuzione da ore a minuti per workload idonei.

L’accelerazione GPU è particolarmente vantaggiosa in scenari con dataset massivi, computazioni vettorializzate complesse e pipeline di machine learning, ampliando drasticamente le prestazioni e la scalabilità di PySpark per i task moderni di data engineering. Il nostro corso Machine Learning with PySpark approfondisce questi concetti. 

Conclusione

Le UDF in PySpark sono uno strumento potente per estendere le capacità di Spark, consentendo ai team di affrontare elaborazioni dati complesse e personalizzate oltre le funzioni integrate. Se applicate correttamente, sbloccano flessibilità e innovazione nelle pipeline dati su larga scala. 

Tuttavia, ottimizzare le prestazioni delle UDF richiede attenzione, evitando insidie comuni come le operazioni riga per riga, gestendo le eccezioni con cura e sfruttando tecniche come la vettorializzazione con Pandas UDF e l’integrazione con Arrow.

Progressi emergenti, come l’accelerazione GPU tramite framework come RAPIDS, stanno ampliando ulteriormente ciò che è possibile con workflow basati su UDF. Che tu stia trasformando dati reali disordinati o integrando analitiche avanzate in sistemi di produzione, padroneggiare le best practice sulle UDF è essenziale per costruire pipeline dati veloci, efficienti e affidabili. 

Impara i dettagli concreti su cui i data scientist trascorrono il 70-80% del loro tempo, data wrangling e feature engineering, con il nostro corso Feature Engineering with PySpark.