Wie man PySpark UDFs und Pandas UDFs effektiv nutzt

Benutzerdefinierte Funktionen (User-Defined Functions, UDFs) in PySpark bieten Python-Entwicklern eine leistungsstarke Möglichkeit, einzigartige Aufgaben zu bewältigen, die die eingebauten Spark-Funktionen einfach nicht schaffen. Wenn du ein/e Dateningenieur/in, Analyst/in oder Wissenschaftler/in bist, der/die Python beherrscht, kannst du mit dem Verständnis der UDF-Konzepte komplexe, reale Datenherausforderungen effektiv angehen. 

Dieses Tutorial führt dich durch PySpark UDF-Konzepte, praktische Implementierungen und Best Practices für Optimierung, Testen, Debugging und fortgeschrittene Nutzungsmuster. Am Ende des Kurses wirst du in der Lage sein, effiziente UDFs zu schreiben, zu optimieren und in großem Umfang einzusetzen.

Wenn du neu in PySpark bist, empfehle ich dir, dir zuerst unser Erste Schritte mit PySpark-Tutorial da wir hier fortgeschrittene Spark-Konzepte behandeln.

Was sind PySpark UDFs?

PySpark UDFs sind benutzerdefinierte Python-Funktionen, die in das verteilte Framework von Spark integriert sind, um Daten zu bearbeiten, die in Spark DataFrames gespeichert sind. Im Gegensatz zu den integrierten Spark-Funktionen können Entwickler mit UDFs komplexe, benutzerdefinierte Logik auf Zeilen- oder Spaltenebene anwenden. 

Unser PySpark Spickzettel deckt alles ab, was du über Spark DataFrames wissen musst, und macht es dir noch einfacher, Spark UDFs zu verstehen.

Wann solltest du PySpark UDFs verwenden?

Verwenden Sie UDFs, wenn:

• Du brauchst Logik, die nicht mit den eingebauten Funktionen von Spark ausgedrückt werden kann.
• Deine Transformation beinhaltet komplexe Python-spezifische Operationen (z. B. Regex-Manipulationen, benutzerdefinierte NLP-Logik).
• Es ist in Ordnung, wenn du Leistung gegen Flexibilität eintauschst, vor allem beim Prototyping oder bei kleinen bis mittleren Datensätzen.

Vermeiden Sie UDFs, wenn:

• Eine vergleichbare Funktionalität gibt es in pyspark.sql.functions. Die nativen Funktionen von Spark sind schneller, optimiert und können in die Ausführungsengine verlagert werden.
• Du arbeitest mit großen Datenmengen, bei denen die Leistung entscheidend ist. UDFs verursachen Serialisierungs-Overhead und beeinträchtigen die Fähigkeit von Spark, Ausführungspläne zu optimieren.
• Du kannst deine Logik mit SQL-Ausdrücken, Spark SQL-Build-Ins oder Pandas UDFs (für vektorisierte Operationen) ausdrücken.

Strategische Anwendungen in der Datentechnik

Hier sind die wichtigsten Anwendungsfälle für PySpark UDFs:

• Komplexe Datentransformationen, wie z.B. fortgeschrittenes Text-Parsing, Datenextraktion oder String-Manipulation.
• Integration mit Python-Bibliotheken von Drittanbietern, einschließlich beliebter Frameworks für maschinelles Lernen wie TensorFlow und XGBoost.
• Überbrückung von Altsystemen und Unterstützung einer nahtlosen Schemaentwicklung, wenn sich Datenstrukturen ändern.

UDFs vereinfachen unübersichtliche, reale Data-Engineering-Aufgaben und versetzen Teams in die Lage, unterschiedliche Anforderungen flexibel und effektiv zu erfüllen. 

Jetzt wollen wir herausfinden, wie du PySpark UDFs implementieren kannst. 

PySpark UDFs implementieren

In diesem Abschnitt wird beschrieben, wie man PySpark UDFs praktisch definiert und implementiert.

Standard UDF Deklarationsmethoden

Es gibt drei gängige Methoden, um UDFs in PySpark zu deklarieren:

1. Lambda-basierte UDFs: Schnell und direkt in DataFrame-Abfragen zu definieren; am besten für einfache Operationen.

Lambda-basierte UDF (Basic Python UDF) eignen sich am besten für schnelle und einfache Transformationen. Vermeide sie für große Aufträge, bei denen es auf Leistung ankommt.

Hier ist ein Beispiel: 

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

2. Geschmückte Python-Funktionen: Explizit kommentiert mit @pyspark.sql.functions.udf, was die Wiederverwendbarkeit und Lesbarkeit unterstützt.

Zum Beispiel:

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

3. SQL-registrierte UDFs: Sie werden direkt in Spark SQL-Kontexten registriert, damit sie in SQL-Abfragen verwendet werden können.

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

Jede Methode ist mit Kompromissen verbunden: Lambda-UDFs sind knapp, aber begrenzt, während Funktionsannotationen die Lesbarkeit, Wartbarkeit und Best Practices verbessern.

Pandas UDFs ermöglichen vektorisierte Operationen auf Pfeilstapeln. Sie sind oft schneller als normale UDFs und lassen sich besser in die Ausführungsengine von Spark integrieren.

Skalare Pandas UDF (elementweise, wie map)

Diese eignen sich am besten für schnelle, zeilenweise Transformationen großer Datensätze. Zum Beispiel: 

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

Gruppierte Karte Pandas UDF

Diese sind am besten für benutzerdefinierte Logik pro Gruppe geeignet, ähnlich wie 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()

Pandas Aggregate UDF

Diese führt Aggregationen über Gruppen durch, schneller als eine gruppierte Karte. Zum Beispiel: 

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

Pandas Iterator UDF

Die Pandas Iterator UDF eignet sich am besten für große Datensätze, die eine speicherarme Verarbeitung (stapelweise) erfordern. Zum Beispiel: 

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

Typenhandhabung und Nullsicherheit

Typen und Nullwerte stellen für PySpark UDFs häufig eine Herausforderung dar. PySpark erzwingt eine strenge Typüberprüfung, was oft zu impliziten Typkonvertierungen oder Laufzeitproblemen führt. Außerdem übergibt Spark Nullwerte direkt an UDFs, was zu Abstürzen führen kann, wenn es nicht explizit behandelt wird.

Sorge mit diesen Strategien für robuste Stadtentwicklungsfonds:

• Gib explizit Rückgabetypen an.
• Integriere Nullprüfungen (z. B. bedingte Anweisungen) in deine Python-Funktionen.
• Führe defensive Kodierungspraktiken ein; einfache Nullprüfungen verhindern frustrierende Laufzeitausnahmen.

Optimierung der UDF-Leistung

Aufgrund ihres zeilenweisen Ausführungsmodells ist die Leistung oft die Achillesferse von Standard-UDFs. Durch die Nutzung vektorisierter UDFs und der Optimierungswerkzeuge von Spark lassen sich die Laufzeiten erheblich verbessern.

Vektorisierte UDFs mit Pandas-Integration

Pandas UDFs führen einen vektorisierten Ansatz für UDFs in PySpark ein, indem sie Datenpakete als Pandas Series an Python-Funktionen übergeben. Dieses Design verbessert die Leistung erheblich, indem es den Serialisierungs-Overhead im Vergleich zu zeilenbasierten Standard-UDFs reduziert.

Unterstützt von Apache Arrow für den kopierfreien Datentransfer zwischen der JVM und den Python-Prozessen, ermöglichen Pandas UDFs die effiziente Ausführung von Operationen in großem Umfang. Sie sind besonders effektiv für intensive Berechnungen und komplexe String-Manipulationen über Millionen von Datensätzen hinweg. 

Mehr Details zur Datenbearbeitung mit PySpark findest du in unserem Datenbereinigung mit PySpark.

Darüber hinaus ermöglichen die Pandas UDFs eine nahtlose Integration in das breitere Ökosystem der Python-Datenwissenschaft, indem sie bekannte Tools und Workflows nutzen.

Pfeil-Optimierungstechniken

Das spaltenförmige Speicherformat von Apache Arrow ermöglicht einen effizienten Datentransfer ohne Kopieren zwischen der Spark JVM und Python-Prozessen. Wenn du Arrow (spark.sql.execution.arrow.pyspark.enabled=true) in deinen Spark-Konfigurationen aktivierst, werden die Daten schnell zwischen der JVM und den Python-Umgebungen verschoben, was die UDF-Ausführung erheblich beschleunigt.

Optimierung des Ausführungsplans

Um PySpark-Jobs zu optimieren, musst du wissen, wie du den Catalyst-Optimierer von Spark beeinflussen kannst. Zu den fortgeschrittenen Strategien gehören Techniken wie Prädikat-Pushdown, Spaltenbeschneidung und die Verwendung von Broadcast Join Hints, um die Abfrageplanung und Ausführungseffizienz zu verbessern.

Um die Leistung zu maximieren, ist es wichtig, den Umfang der UDF-Ausführung zu minimieren und integrierte Spark SQL-Funktionen zu bevorzugen, wann immer dies möglich ist. Der strategische Einsatz von Caching und die sorgfältige Erstellung von Plänen können die Ausführungsgeschwindigkeit und die Ressourcennutzung weiter verbessern.

Leistungsoptimierung ist eine der wichtigsten Fragen, die dir in einem PySpark-Interview begegnen können. Wie du diese und weitere Spark-Fragen beantworten kannst, erfährst du in unseren Top 36 PySpark Interview Fragen und Antworten für 2025 Blog-Beitrag.

Fortgeschrittene Patterns und Anti-Patterns

Das Verständnis von richtigen und falschen Nutzungsmustern hilft, stabile und effiziente UDF-Einsätze zu gewährleisten.

Zustandsabhängige UDF-Implementierungen

Zustandsabhängige und nicht-deterministische UDFs stellen in PySpark besondere Herausforderungen dar. Diese Funktionen erzeugen Ergebnisse, die von externen Zuständen oder sich ändernden Bedingungen abhängen, wie z.B. Umgebungsvariablen, Systemzeit oder Sitzungskontext.

Auch wenn nicht-deterministische UDFs manchmal notwendig sind - z.B. um Zeitstempel zu erzeugen, Benutzersitzungen zu verfolgen oder Zufälligkeiten einzuführen - können sie die Fehlersuche, Reproduzierbarkeit und Optimierung erschweren.

Die Implementierung von zustandsbehafteten UDFs erfordert sorgfältige Entwurfsmuster: Das Verhalten muss klar dokumentiert werden, Seiteneffekte müssen isoliert werden und es muss eine gründliche Protokollierung erfolgen, um die Fehlersuche zu erleichtern und die Konsistenz zwischen den einzelnen Jobläufen zu gewährleisten.

Wenn sie gut durchdacht eingesetzt werden, können sie mächtige Fähigkeiten freisetzen, aber die Aufrechterhaltung zuverlässiger Datenpipelines erfordert ein diszipliniertes Management. Unser Big-Data-Grundlagen mit PySpark-Kurs geht genauer darauf ein, wie man Big Data in PySpark handhabt.

Häufige Anti-Muster

Gängige Anti-Patterns bei der Verwendung von UDFs können die Leistung von PySpark erheblich beeinträchtigen:

• Zeilenweise Verarbeitung anstelle von Stapelverarbeitung: Die Anwendung von UDFs auf einzelne Zeilen, anstatt vektorisierte Ansätze wie Pandas UDFs zu verwenden, führt zu einer erheblichen Verlangsamung der Ausführung.
• Verschachtelte DataFrame-Operationen innerhalb von UDFs: Die Einbettung von DataFrame-Abfragen in UDFs verursacht übermäßige Berechnungen und behindert die Fähigkeit von Spark, Ausführungspläne zu optimieren.
• Wiederholte Inline-UDF-Registrierung: Das mehrfache Definieren und Registrieren von UDFs innerhalb von Abfragen fügt unnötigen Overhead hinzu; es ist besser, UDFs einmal zu deklarieren und sie über Jobs hinweg wiederzuverwenden.
• Übermäßige Verwendung von benutzerdefinierter Python-Logik für einfache Operationen: Für Aufgaben wie einfache Filterung, Arithmetik oder einfache Transformationen sollten die hoch optimierten integrierten Funktionen von Spark den eigenen UDFs vorgezogen werden.

Die Vermeidung dieser Fallstricke sorgt für eine bessere Leistung, eine einfachere Optimierung durch Catalyst und einen besser wartbaren PySpark-Code.

Debuggen und Testen von PySpark UDFs

Das Testen und Debuggen von UDFs gewährleistet Zuverlässigkeit und Robustheit in Produktionsszenarien.

Muster für die Behandlung von Ausnahmen

Die Implementierung einer strukturierten Fehlererfassung in UDFs ist für den Aufbau stabiler und wartbarer PySpark-Pipelines unerlässlich. Verwende try-except-Blöcke innerhalb von UDFs, um Laufzeitüberraschungen wie Nullwerte, Typübereinstimmungen oder Fehler bei der Division durch Null elegant zu behandeln.

Eine robuste Ausnahmebehandlung stabilisiert Pipelines gegen unerwartete Daten und vereinfacht die Fehlersuche, indem sie eindeutige, umsetzbare Fehlermeldungen anzeigt. Ordnungsgemäß erfasste und protokollierte Ausnahmen machen das Verhalten von UDFs transparenter, beschleunigen die Problemlösung und verbessern die Zuverlässigkeit der Pipeline insgesamt.

Unit-Testing-Frameworks

Nutze die in PySpark integrierte Test-Basisklasse pyspark.testing.utils.ReusedPySparkTestCase zusammen mit Frameworks wie pytest, um zuverlässige Unit-Tests für deine UDFs zu schreiben. Durch die Strukturierung klarer und zielgerichteter Tests wird die Korrektheit, Stabilität und Wartbarkeit deiner UDF-Logik im Laufe der Zeit sichergestellt.

Zu den bewährten Praktiken für das Testen von UDFs gehören das Abdecken von typischen und seltenen Fällen, die Validierung der Ausgaben anhand bekannter Ergebnisse und das Isolieren des UDF-Verhaltens von externen Abhängigkeiten. Gut durchdachte Tests schützen nicht nur vor Regressionen, sondern vereinfachen auch zukünftige Entwicklungs- und Refactoring-Aufgaben.

Entwicklung und zukünftige Richtungen

Das PySpark-Ökosystem entwickelt sich rasant weiter und führt neue Funktionen ein, die die UDFs noch weiter verbessern.

Integration des Unity-Katalogs

Jüngste Entwicklungen haben die UDF-Registrierung in den Unity-Katalogintegriert, wodurch die Verwaltung, Erkennung und Steuerung von UDFs im großen Maßstab vereinfacht wird. Unity Catalog ermöglicht die zentrale Kontrolle über das UDF-Lebenszyklusmanagement, einschließlich Registrierung, Versionskontrolle und Zugriffskontrolle, die für Unternehmensumgebungen entscheidend sind.

Diese Integration verbessert die Governance, setzt konsistente Sicherheitsrichtlinien durch und verbessert die Auffindbarkeit in verschiedenen Teams, so dass UDFs in großen, komplexen Datenökosystemen leichter wiederverwendet, geprüft und verwaltet werden können.

GPU-beschleunigte UDFs

Frameworks wie RAPIDS Accelerator ermöglichen die Auslagerung von rechenintensiven UDF-Aufgaben in PySpark auf die GPU, was zu erheblichen Leistungssteigerungen führt. Indem schwere Operationen wie numerische Analysen, Deep Learning Inferenzen und umfangreiche Datenmodellierung auf GPUs verlagert werden, kann RAPIDS die Ausführungszeiten bei geeigneten Workloads von Stunden auf Minuten reduzieren.

Die GPU-Beschleunigung ist besonders vorteilhaft für Szenarien mit massiven Datensätzen, komplexen vektorisierten Berechnungen und Pipelines für maschinelles Lernen, die die Leistung und Skalierbarkeit von PySpark für moderne Data-Engineering-Aufgaben drastisch erhöhen. Unser Kurs Maschinelles Lernen mit PySpark taucht tiefer in diese Konzepte ein.

Fazit

PySpark UDFs sind ein leistungsstarkes Werkzeug, um die Fähigkeiten von Spark zu erweitern und Teams in die Lage zu versetzen, komplexe, angepasste Datenverarbeitungsaufgaben zu lösen, die über die eingebauten Funktionen hinausgehen. Wenn sie richtig eingesetzt werden, ermöglichen sie Flexibilität und Innovation in großen Datenpipelines. 

Um die Leistung von UDFs zu optimieren, ist es jedoch wichtig, häufige Fallstricke wie zeilenweise Operationen zu vermeiden, Ausnahmen elegant zu verwalten und Techniken wie die UDF-Vektorisierung von Pandas mit Arrow-Integration zu nutzen.

Neue Entwicklungen, wie die GPU-Beschleunigung durch Frameworks wie RAPIDS, erweitern die Möglichkeiten von UDF-gesteuerten Workflows weiter. Ganz gleich, ob du unordentliche Daten aus der realen Welt umwandelst oder fortschrittliche Analysen in Produktionssysteme einbettest, die Beherrschung von UDF-Best Practices ist für den Aufbau schneller, effizienter und zuverlässiger Datenpipelines unerlässlich. 

In unserem Kurs "Feature Engineering mit PySpark" lernst du die Details kennen, mit denen Data Scientists 70-80% ihrer Zeit verbringen: Data Wrangling und Feature Engineering. Feature Engineering mit PySpark-Kurs.