Python skalarne funkcje definiowane przez użytkownika (UDF)

Python scalar UDF pozwala uruchamiać niestandardową logikę Python w zapytaniach SQL na Azure Databricks. Ta strona pokazuje, jak je rejestrować i wywoływać, korzystać z poświadczeń i sekretów usług oraz obsługiwać zastrzeżenia w kolejności oceny podwyrazów w SQL Spark.

Requirements

  • W środowisku Databricks Runtime 12.2 LTS i starszym funkcje zdefiniowane przez użytkownika języka Python i funkcje zdefiniowane przez użytkownika biblioteki Pandas nie są obsługiwane w obliczeniach na Unity Catalog, które korzystają z trybu standardowego dostępu.

  • Skalarne funkcje zdefiniowane przez użytkownika w Pythonie (UDF) i funkcje zdefiniowane przez użytkownika w Pandas (UDF) są obsługiwane w Databricks Runtime 13.3 LTS i nowszych wersjach dla wszystkich trybów dostępu.

  • Obsługa instancji ARM dla funkcji UDF języka Python w klastrach z włączonym Unity Catalog wymaga środowiska Databricks Runtime 15.2 lub nowszego.

W środowisku Databricks Runtime 14.0 i wcześniejszym funkcje zdefiniowane przez użytkownika w języku Python i funkcje zdefiniowane przez użytkownika w bibliotece Pandas nie są obsługiwane w klastrach Unity Catalog korzystających ze standardowego trybu dostępu. Skalarne funkcje zdefiniowane przez użytkownika w Pythonie i funkcje Pandas zdefiniowane przez użytkownika są obsługiwane dla wszystkich trybów dostępu w systemie Databricks Runtime 14.1 lub nowszym.

W Databricks Runtime 14.1 lub nowszym można zarejestrować skalarne funkcje UDF Pythona w Unity Catalog przy użyciu składni SQL. Zobacz funkcje definiowane przez użytkownika (UDF) w językach SQL i Python w Unity Catalog.

Rejestracja funkcji jako UDF

def squared(s):
  return s * s
spark.udf.register("squaredWithPython", squared)

Opcjonalnie można ustawić typ zwracany funkcji zdefiniowanej przez użytkownika. Domyślnym typem zwracania jest StringType.

from pyspark.sql.types import LongType
def squared_typed(s):
  return s * s
spark.udf.register("squaredWithPython", squared_typed, LongType())

Wywołaj UDF w Spark SQL

spark.range(1, 20).createOrReplaceTempView("test")

%sql select id, squaredWithPython(id) as id_squared from test

Używanie funkcji UDF z ramkami danych

from pyspark.sql.functions import udf
from pyspark.sql.types import LongType
squared_udf = udf(squared, LongType())
df = spark.table("test")
display(df.select("id", squared_udf("id").alias("id_squared")))

Alternatywnie można zadeklarować tę samą funkcję zdefiniowaną przez użytkownika przy użyciu składni adnotacji:

from pyspark.sql.functions import udf

@udf("long")
def squared_udf(s):
  return s * s
df = spark.table("test")
display(df.select("id", squared_udf("id").alias("id_squared")))

Warianty z funkcją UDF

Typ PySpark dla wariantu to VariantType , a wartości są typu VariantVal. Aby uzyskać informacje o wariantach, zobacz Zapytanie dotyczące danych wariantów.

from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import VariantType, VariantVal

# Return Variant
@udf(returnType = VariantType())
def toVariant(jsonString):
  return VariantVal.parseJson(jsonString)

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toVariant(col("json"))).display()

+---------------+
|toVariant(json)|
+---------------+
|        {"a":1}|
+---------------+

from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import StructField, StructType, VariantType, VariantVal

# Return Struct<Variant>
@udf(returnType = StructType([StructField("v", VariantType(), True)]))
def toStructVariant(jsonString):
  return {"v": VariantVal.parseJson(jsonString)}

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toStructVariant(col("json"))).display()

+---------------------+
|toStructVariant(json)|
+---------------------+
|        {"v":{"a":1}}|
+---------------------+

from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import ArrayType, VariantType, VariantVal

# Return Array<Variant>
@udf(returnType = ArrayType(VariantType()))
def toArrayVariant(jsonString):
  return [VariantVal.parseJson(jsonString)]

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toArrayVariant(col("json"))).display()

+--------------------+
|toArrayVariant(json)|
+--------------------+
|           [{"a":1}]|
+--------------------+

from pyspark.sql.functions import col, lit, udf
from pyspark.sql.types import MapType, StringType, VariantType, VariantVal

# Return Map<String, Variant>
@udf(returnType = MapType(StringType(), VariantType(), True))
def toMapVariant(jsonString):
  return {"v1": VariantVal.parseJson(jsonString), "v2": VariantVal.parseJson("[" + jsonString + "]")}

spark.range(1).select(lit('{"a" : 1}').alias("json")).select(toMapVariant(col("json"))).display()

+-----------------------------+
|           toMapVariant(json)|
+-----------------------------+
|{"v2":[{"a":1}],"v1":{"a":1}}|
+-----------------------------+

Pliki z UDF

Important

Ta funkcja jest dostępna w wersji beta.

Typ PySpark dla pliku to FileType. Użyj go jako parametru lub typu zwracanego w UDF, zarówno jako typu najwyższego poziomu, jak i zagnieżdżonego. Informacje o typie, jego regułach zagnieżdżania oraz interfejsie API FileRef znajdziesz w dokumencie FileType.

Aby odczytać zawartość pliku w UDF, wywołaj file.as_local_file() , aby uzyskać lokalną ścieżkę, którą możesz otworzyć, lub file.open() odczytać jego bajty jako strumień. Przykłady w Python, Scala i SQL, w tym przetwarzanie obrazów, wykrywanie typów plików i ekstrakcję klatek wideo, można znaleźć w artykule Pliki procesowe z UDF. Aby uzyskać informacje o typie, zobacz FILE typ.

UDF zarejestrowany w Unity Catalog (CREATE FUNCTION) może odczytać metadane elementu FILE, ale nie jego zawartość i nie może tworzyć plików. Użyj UDF ograniczonego do sesji do odczytu zawartości pliku (open, as_local_file) lub utworzenia go (from_bytes, from_local_file).

W ciele UDF każda FILE wartość jest obiektem FileRef :

from pyspark.sql.functions import col, udf
from pyspark.sql.types import FileRef, StringType

@udf(returnType=StringType())
def file_content_type(file: FileRef) -> str:
  return file.content_type

df = spark.table("documents")
display(df.select(col("file").uri, file_content_type(col("file"))))

Kolejność oceny i sprawdzanie wartości null

Spark SQL (w tym SQL, API DataFrame oraz Dataset) nie gwarantuje kolejności oceny podwyrażeń. W szczególności dane wejściowe operatora lub funkcji nie muszą być oceniane od lewej do prawej ani w innej stałej kolejności. Na przykład wyrażenia logiczne AND i OR nie mają semantyki od lewej do prawej „skrótowego przerywania”.

W związku z tym niebezpieczne jest poleganie na skutkach ubocznych lub kolejności obliczania wyrażeń logicznych oraz kolejności klauzul WHERE i HAVING, ponieważ takie wyrażenia i klauzule mogą zostać zmienione podczas optymalizacji zapytań i planowania. W szczególności, jeśli funkcja UDF wykorzystuje semantykę zwarciową w języku SQL w celu sprawdzania wartości null, nie ma gwarancji, że sprawdzanie wartości null nastąpi przed wywołaniem funkcji UDF. Na przykład:

spark.udf.register("strlen", lambda s: len(s), "int")
spark.sql("select s from test1 where s is not null and strlen(s) > 1") # no guarantee

Ta WHERE klauzula nie gwarantuje, że funkcja zdefiniowana przez użytkownika strlen zostanie wywołana po usunięciu wartości null.

Aby wykonać odpowiednie sprawdzanie wartości null, zalecamy wykonanie jednej z następujących czynności:

  • Upewnij się, że funkcja UDF obsługuje wartości null i sprawdza je wewnątrz siebie samej
  • Użyj wyrażeń IF lub CASE WHEN do sprawdzenia braku wartości i wywołania funkcji zdefiniowanej przez użytkownika w gałęzi warunkowej
spark.udf.register("strlen_nullsafe", lambda s: len(s) if not s is None else -1, "int")
spark.sql("select s from test1 where s is not null and strlen_nullsafe(s) > 1") # ok
spark.sql("select s from test1 where if(s is not null, strlen(s), null) > 1")   # ok

Uzyskaj dostęp do sekretów katalogu Unity

Aby uzyskać dostęp do sekretu Unity Catalog z funkcji UDF języka Python o zakresie sesji, zobacz Używanie sekretu w funkcji UDF języka Python o zakresie sesji. Aby uzyskać dostęp do zadeklarowanych wpisów tajnych w skalarnej lub wsadowej funkcji UDF języka Python w katalogu Unity Catalog, zobacz Używanie wpisów tajnych w funkcji UDF języka Python.

Dane uwierzytelniające usługi w funkcjach UDF języka Python

Skalarne funkcje UDF języka Python o zakresie sesji oraz skalarne funkcje UDF języka Python w usłudze Unity Catalog mogą używać poświadczeń usługi Unity Catalog do bezpiecznego uzyskiwania dostępu do zewnętrznych usług chmurowych. Jest to przydatne w przypadku integrowania operacji, takich jak tokenizacja w chmurze, szyfrowanie lub zarządzanie tajemnicami bezpośrednio w przekształceniach danych.

Wymagania różnią się w zależności od typu i poziomu obliczeniowego UDF. Zobacz Użyj poświadczenia serwisowego w Python UDF.

Aby utworzyć poświadczenia usługi, zobacz Tworzenie poświadczeń usługi.

Użyj poświadczenia usługi w skalarnym UDF języka Python o zakresie sesji

Aby uzyskać dostęp do poświadczeń usługi, użyj narzędzia databricks.service_credentials.getServiceCredentialsProvider() w logice UDF, aby zainicjować SDK w chmurze odpowiednimi poświadczeniami. Cały kod musi być enkapsulowany w treści funkcji UDF.

@udf
def use_service_credential():
    from azure.mgmt.web import WebSiteManagementClient

    # Assuming there is a service credential named 'testcred' set up in Unity Catalog
    web_client = WebSiteManagementClient(subscription_id, credential = getServiceCredentialsProvider('testcred'))
    # Use web_client to perform operations

Uprawnienia poświadczeń usługi

Sesyjne UDF korzystają z uprawnień wywołującego. Zobacz Użyj poświadczenia usługi w Python UDF, aby uzyskać wymagane uprawnienia.

Domyślne poświadczenia na poziomie zasobów obliczeniowych dla funkcji UDF o zakresie sesji

W przypadku skalarnych UDF-ów Python, Databricks automatycznie używa domyślnego poświadczenia usługi ze zmiennej środowiska obliczeniowego. To zachowanie pozwala bezpiecznie odwoływać się do usług zewnętrznych bez jawnego zarządzania aliasami poświadczeń w kodzie UDF. Zobacz Określanie domyślnego poświadczenia usługi dla zasobu obliczeniowego

Domyślna obsługa poświadczeń jest dostępna tylko w klastrach w trybie dostępu standardowym i dedykowanym. Nie jest dostępny w magazynach SQL.

Musisz zainstalować pakiet azure-identity, aby używać dostawcy DefaultAzureCredential. Aby zainstalować pakiet, zobacz Biblioteki Pythona o zakresie notatnika lub Biblioteki o zakresie obliczeniowym.

@udf
def use_service_credential():
    from azure.identity import DefaultAzureCredential
    from azure.mgmt.web import WebSiteManagementClient

    # DefaultAzureCredential is automatically using the default service credential for the compute
    web_client_default = WebSiteManagementClient(DefaultAzureCredential(), subscription_id)

    # Use web_client to perform operations

Użyj poświadczenia usługi w skalarnym Unity Catalog Python UDF

Określ poświadczenie usługi w klauzuli CREDENTIALS definicji UDF. Możesz oznaczyć jedno poświadczenie jako DEFAULT, aby poprawione chmurowe pakiety SDK automatycznie z niego korzystały. W klasycznym trybie obliczeniowym ta funkcja wymaga Databricks Runtime 18.1 lub nowszego. W środowisku serverless compute oraz w hurtowniach SQL typu pro i serverless jawnie ustaw parametr environment_version funkcji UDF na 6 lub wyższą wartość. Pełne informacje o wymaganiach dotyczących mocy obliczeniowej, sieci i uprawnień znajdziesz w sekcji Korzystanie z poświadczenia usługi w funkcji UDF języka Python.

Pobierz kontekst wykonania zadania

Użyj API PySpark TaskContext, aby uzyskać informacje kontekstowe, takie jak tożsamość użytkownika, tagi klastra, identyfikator zadania Spark i więcej. Zobacz Uzyskiwanie kontekstu zadania w UDF.

Ograniczenia

Następujące ograniczenia dotyczą UDF-ów PySpark:

  • Ograniczenia dostępu do plików: W środowisku Databricks Runtime 14.2 i niżej funkcje zdefiniowane przez użytkownika PySpark w udostępnionych klastrach nie mogą uzyskiwać dostępu do folderów Git, plików obszarów roboczych ani woluminów Unity Catalog.

  • Zmienne rozgłaszane: Funkcje zdefiniowane przez użytkownika PySpark w klastrach w standardowym trybie dostępu i w obliczeniach bezserwerowych nie obsługują zmiennych rozgłaszanych.

  • Limit pamięci bezserwerowej: funkcje zdefiniowane przez użytkownika PySpark w obliczeniach bezserwerowych mają limit pamięci 1 GB na funkcję zdefiniowanej przez użytkownika PySpark. Przekroczenie tego limitu powoduje błąd typu UDF_PYSPARK_USER_CODE_ERROR. MEMORY_LIMIT_SERVERLESS.
  • Limit pamięci w standardowym trybie dostępu: UDF (funkcje zdefiniowane przez użytkownika) PySpark w standardowym trybie dostępu mają limit pamięci w zależności od dostępnej pamięci dla wybranego typu wystąpienia. Przekroczenie dostępnej pamięci powoduje błąd typu UDF_PYSPARK_USER_CODE_ERROR. MEMORY_LIMIT.
  • Dostęp do sieci w bezserwerowych hurtowniach SQL: Domyślnie funkcje UDF języka Python w bezserwerowych hurtowniach SQL nie mogą wykonywać wychodzących żądań sieciowych, a zapytania, które próbują wykonywać wywołania sieciowe, zawieszają się na czas nieokreślony. Aby włączyć dostęp do sieci wychodzącej, włącz funkcję Publiczna wersja zapoznawcza Włącz sieć dla izolowanych obciążeń w bezserwerowych magazynach SQL Warehouse na stronie Podglądy obszaru roboczego. W przeciwnym razie użyj bezserwerowych lub klasycznych zasobów obliczeniowych w przypadku funkcji zdefiniowanych przez użytkownika, które wymagają dostępu do sieci.
  • Ostatnia aktualizacja o