Was sind benutzerdefinierte Funktionen (UDFs, User Defined Functions)?

Eine benutzerdefinierte Funktion (User-Defined Function, UDF) ist eine vom Benutzer definierte Funktion, mit der benutzerdefinierte Logik in der Benutzerumgebung wiederverwendet werden kann. Azure Databricks unterstützt viele verschiedene Arten von UDFs, um die Verteilung erweiterbarer Logik zu ermöglichen. In diesem Artikel werden einige der allgemeinen Stärken und Einschränkungen von UDFs vorgestellt.

Definieren benutzerdefinierter Logik ohne Serialisierungsabzüge

Azure Databricks erbt einen Großteil des UDF-Verhaltens von Apache Spark, einschließlich der Effizienzbeschränkungen bei vielen Arten von UDFs. Weitere Informationen finden Sie unter Welche UDFs sind am effizientesten?

Sie können Ihren Code gefahrlos modularisieren, ohne sich über mögliche Effizienzverluste im Zusammenhang mit UDFs Gedanken zu machen. Hierzu müssen Sie Ihre Logik als eine Reihe von integrierten Spark-Methoden mit SQL oder Spark DataFrames definieren. Die folgenden SQL- und Python-Funktionen kombinieren beispielsweise integrierte Spark-Methoden, um eine Einheitenkonvertierung als wiederverwendbare Funktion zu definieren:

SQL

CREATE FUNCTION convert_f_to_c(unit STRING, temp DOUBLE)
RETURNS DOUBLE
RETURN CASE
  WHEN unit = "F" THEN (temp - 32) * (5/9)
  ELSE temp
END;

SELECT convert_f_to_c(unit, temp) AS c_temp
FROM tv_temp;

Python

def convertFtoC(unitCol, tempCol):
  from pyspark.sql.functions import when
  return when(unitCol == "F", (tempCol - 32) * (5/9)).otherwise(tempCol)

from pyspark.sql.functions import col

df_query = df.select(convertFtoC(col("unit"), col("temp"))).toDF("c_temp")
display(df_query)

Um die oben genannten UDFs auszuführen, können Sie Beispieldaten erstellen.

Welche UDFs sind am effizientesten?

UDFs können zu erheblichen Verarbeitungsengpässen bei der Codeausführung führen. Azure Databricks verwendet eine Reihe verschiedener Optimierer automatisch für Code, der mit der enthaltenen Apache Spark-, SQL- und Delta Lake-Syntax geschrieben wird. Wenn benutzerdefinierte Logik durch UDFs eingeführt wird, sind diese Optimierer nicht in der Lage, Aufgaben für diese benutzerdefinierte Logik effizient zu planen. Darüber hinaus verursacht Logik, die außerhalb der JVM ausgeführt wird, zusätzliche Kosten für die Datenserialisierung.

Einige UDFs sind effizienter als andere. In Bezug auf die Leistung:

• Integrierte Funktionen sind aufgrund von Azure Databricks-Optimierern am schnellsten.
• Code, der in der JVM ausgeführt wird (Scala-, Java-, Hive-UDFs), ist schneller als Python-UDFs.
• Pandas-UDFs verwenden Arrow, um die mit Python-UDFs verbundenen Serialisierungskosten zu reduzieren.
• Python-UDFs eignen sich gut für prozedurale Logik, sollten jedoch für ETL-Produktionsworkloads mit großen Datasets vermieden werden.

Wann sollten Sie eine UDF verwenden?

Ein Hauptvorteil von UDFs besteht darin, dass sie Benutzern gestatten, Logik in vertrauten Sprachen auszudrücken, wodurch die mit dem Refactoring von Code verbundenen Personalkosten reduziert werden. Bei Ad-hoc-Abfragen, manuellen Datenbereinigungen, explorativen Datenanalysen und den meisten Vorgängen mit kleinen oder mittleren Datasets werden die mit UDFs verbundenen Latenzkosten Kosten, die mit dem Refactoring von Code einhergehen, wahrscheinlich nicht überwiegen.

Bei ETL-Aufträgen, Streamingvorgängen, Vorgängen mit sehr großen Datasets oder anderen Workloads, die regelmäßig oder fortwährend ausgeführt werden, zahlt sich das Refactoring von Logik für die Verwendung nativer Apache Spark-Methoden schnell aus.

Beispieldaten für Beispiel-UDFs

Die Codebeispiele in diesem Artikel verwenden UDFs, um Temperaturen zwischen Celcius und Farenheit zu konvertieren. Wenn Sie diese Funktionen ausführen möchten, können Sie ein Beispieldataset mit dem folgenden Python-Code erstellen:

import numpy as np
import pandas as pd

Fdf = pd.DataFrame(np.random.normal(55, 25, 10000000), columns=["temp"])
Fdf["unit"] = "F"

Cdf = pd.DataFrame(np.random.normal(10, 10, 10000000), columns=["temp"])
Cdf["unit"] = "C"

df = spark.createDataFrame(pd.concat([Fdf, Cdf]).sample(frac=1))

df.cache().count()
df.createOrReplaceTempView("tv_temp")