Co to są funkcje zdefiniowane przez użytkownika (UDF)?

Funkcja zdefiniowana przez użytkownika (UDF) jest funkcją zdefiniowaną przez użytkownika, umożliwiając ponowne użycie logiki niestandardowej w środowisku użytkownika. Usługa Azure Databricks obsługuje wiele różnych typów funkcji zdefiniowanych przez użytkownika, aby umożliwić dystrybucję rozszerzalnej logiki. W tym artykule przedstawiono niektóre ogólne mocne strony i ograniczenia funkcji zdefiniowanych przez użytkownika.

Uwaga

Nie wszystkie formy funkcji zdefiniowanych przez użytkownika są dostępne we wszystkich środowiskach wykonywania w usłudze Azure Databricks. Jeśli pracujesz z wykazem aparatu Unity, zobacz Funkcje zdefiniowane przez użytkownika (UDF) w wykazie aparatu Unity.

Definiowanie logiki niestandardowej bez kar serializacji

Usługa Azure Databricks dziedziczy wiele zachowań funkcji zdefiniowanej przez użytkownika z platformy Apache Spark, w tym ograniczenia wydajności dotyczące wielu typów funkcji zdefiniowanych przez użytkownika. Zobacz Jakie funkcje zdefiniowane przez użytkownika są najbardziej wydajne?.

Możesz bezpiecznie modularyzować kod bez obaw o potencjalne kompromisy wydajności związane z funkcjami zdefiniowanymi przez użytkownika. W tym celu należy zdefiniować logikę jako serię wbudowanych metod platformy Spark przy użyciu ramek danych SQL lub Spark. Na przykład następujące funkcje SQL i Python łączą wbudowane metody platformy Spark w celu zdefiniowania konwersji jednostki jako funkcji wielokrotnego użytku:

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)

Aby uruchomić powyższe funkcje zdefiniowane przez użytkownika, możesz utworzyć przykładowe dane.

Które funkcje zdefiniowane przez użytkownika są najbardziej wydajne?

Funkcje zdefiniowane przez użytkownika mogą wprowadzać znaczne wąskie gardła przetwarzania do wykonywania kodu. Usługa Azure Databricks automatycznie używa wielu różnych optymalizatorów dla kodu napisanego przy użyciu składni apache Spark, SQL i Delta Lake. Po wprowadzeniu logiki niestandardowej przez funkcje zdefiniowane przez użytkownika te optymalizatory nie mają możliwości wydajnego planowania zadań wokół tej logiki niestandardowej. Ponadto logika wykonywana poza maszyną wirtualną JVM ma dodatkowe koszty związane z serializacji danych.

Uwaga

Usługa Azure Databricks optymalizuje wiele funkcji przy użyciu aplikacji Photon, jeśli używasz obliczeń z obsługą technologii Photon. Tylko funkcje, które łączą ze sobą polecenia Spark SQL ramki danych, można zoptymalizować za pomocą aplikacji Photon.

Niektóre funkcje zdefiniowane przez użytkownika są bardziej wydajne niż inne. Pod względem wydajności:

• Wbudowane funkcje będą najszybsze ze względu na optymalizatory usługi Azure Databricks.
• Kod wykonywany w środowisku JVM (Scala, Java, Hive UFs) będzie szybszy niż funkcje zdefiniowane przez użytkownika języka Python.
• Funkcje zdefiniowane przez użytkownika biblioteki Pandas używają funkcji Strzałka, aby zmniejszyć koszty serializacji skojarzone z funkcjami zdefiniowanymi przez użytkownika języka Python.
• Funkcje zdefiniowane przez użytkownika języka Python działają dobrze w przypadku logiki proceduralnej, ale należy unikać obciążeń produkcyjnych ETL w dużych zestawach danych.

Uwaga

W środowisku Databricks Runtime 12.2 LTS i nowszych funkcjach zdefiniowanych przez użytkownika języka Python nie jest obsługiwana w wykazie aparatu Unity w klastrach korzystających z trybu dostępu współdzielonego. Te funkcje zdefiniowane przez użytkownika są obsługiwane w środowisku Databricks Runtime 13.3 LTS i nowszym dla wszystkich trybów dostępu.

W środowisku Databricks Runtime 14.1 lub nowszym scala skalarne funkcje zdefiniowane przez użytkownika nie są obsługiwane w wykazie aparatu Unity w klastrach korzystających z trybu dostępu współdzielonego. Te funkcje zdefiniowane przez użytkownika są obsługiwane dla wszystkich trybów dostępu w środowisku Databricks Runtime 14.2 lub nowszym.

W środowisku Databricks Runtime 13.3 LTS i nowszym można zarejestrować scalarne funkcje zdefiniowane przez użytkownika języka Python w katalogu aparatu Unity przy użyciu składni JĘZYKA SQL. Zobacz Funkcje zdefiniowane przez użytkownika (UDF) w wykazie aparatu Unity.

Typ	Optymalizacja	Środowisko wykonywania
Funkcja zdefiniowanej przez użytkownika programu Hive	Nie.	JVM
Python UDF	Nie.	Python
Pandas UDF	Nie.	Python (strzałka)
Scala UDF	Nie.	JVM
Spark SQL	Tak	JVM (Photon)
Ramka danych platformy Spark	Tak	JVM (Photon)

Kiedy należy użyć funkcji zdefiniowanej przez użytkownika?

Główną zaletą funkcji zdefiniowanych przez użytkownika jest umożliwienie użytkownikom wyrażania logiki w znanych językach, co zmniejsza koszty człowieka związane z refaktoryzowaniem kodu. W przypadku zapytań ad hoc, czyszczenia danych ręcznych, eksploracyjnej analizy danych i większości operacji na małych lub średnich zestawach danych koszty obciążeń związanych z opóźnieniami skojarzonymi z funkcjami zdefiniowanymi przez użytkownika są mało prawdopodobne, aby przewyższać koszty związane z refaktoryzowaniem kodu.

W przypadku zadań ETL, operacji przesyłania strumieniowego, operacji na bardzo dużych zestawach danych lub innych obciążeniach, które są wykonywane regularnie lub stale, logika refaktoryzacji do korzystania z natywnych metod platformy Apache Spark szybko płaci dywidendy.

Przykładowe dane na przykład zdefiniowane przez użytkownika

Przykłady kodu w tym artykule używają funkcji zdefiniowanych przez użytkownika do konwertowania temperatur między Celcius i Farenheit. Jeśli chcesz wykonać te funkcje, możesz utworzyć przykładowy zestaw danych przy użyciu następującego kodu w języku Python:

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