Python skalární uživatelsky definované funkce (UDF)

Python skalární UDF umožňuje spouštět vlastní Python logiku uvnitř SQL dotazů na Azure Databricks. Tato stránka popisuje, jak je registrovat a volat, jak používat přihlašovací údaje služby a tajné klíče a jak řešit omezení pořadí vyhodnocování podvýrazů v Spark SQL.

Požadavky

  • Ve službě Databricks Runtime 12.2 LTS a starších verzích nejsou na výpočetních prostředcích katalogu Unity podporovány Python UDF ani Pandas UDF při použití standardního režimu přístupu.

  • Uživatelem definované skalární funkce v Pythonu a uživatelem definované funkce v Pandas jsou podporovány v prostředí Databricks Runtime 13.3 LTS a vyšším ve všech režimech přístupu.

  • Podpora instancí ARM pro uživatelské funkce definované v Pythonu na clusterech se zapnutým katalogem Unity vyžaduje Databricks Runtime 15.2 nebo vyšší.

V Databricks Runtime 14.0 a starších verzích nejsou u clusterů Unity Catalog, které používají standardní režim přístupu, podporované uživatelem definované funkce Pythonu a uživatelem definované funkce Pandas. Skalární uživatelem definované funkce v Pythonu a uživatelem definované funkce v Pandas jsou podporovány pro všechny režimy přístupu v Databricks Runtime 14.1 a novějších.

Ve Databricks Runtime 14.1 a novějších můžete registrovat skalární uživatelsky definované funkce Pythonu do Unity Catalog pomocí SQL syntaxe. Viz SQL a Python uživatelem definované funkce (UDF) v katalogu Unity.

Registrujte funkci jako UDF

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

Můžete volitelně nastavit návratový typ vaší uživatelsky definované funkce. Výchozí návratový typ je StringType.

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

Zavolejte UDF v Spark SQL

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

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

Použití UDF s DataFrames

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

Alternativně můžete deklarovat stejnou UDF pomocí poznámkové syntaxe:

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

Varianty s UDF

Typ PySpark pro variantu je VariantType a hodnoty jsou typu VariantVal. Informace o variantách najdete v tématu Dotazování na data variant.

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}}|
+-----------------------------+

Soubory s UDF

Important

Tato funkce je v beta verzi.

Typ PySpark pro soubor je .FileType Použijte jej jako parametr nebo návratový typ v UDF, buď jako typ nejvyšší úrovně, nebo jako vnořený typ. Informace o typu, jeho pravidlech vnoření a rozhraní FileRef API naleznete v části FileType.

Chcete-li v UDF přečíst obsah souboru, zavolejte file.as_local_file(), čímž získáte místní cestu, kterou můžete otevřít, nebo file.open() k načtení jeho bajtů jako datového proudu. Pro příklady v Python, Scala a SQL, včetně zpracování obrazu, detekce typů souborů a extrakce video snímků, viz Process files with UDF. Pro typovou referenci viz FILE typ.

Funkce UDF registrovaná v Unity Catalog (CREATE FUNCTION) může číst metadata objektu FILE, nikoli však jeho obsah, a nemůže vytvářet soubory. Pomocí funkce UDF s platností v rámci relace můžete načíst obsah souboru (open, as_local_file) nebo soubor vytvořit (from_bytes, from_local_file).

V těle UDF je každá hodnota FileRef objektem FILE:

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

Kontrola pořadí vyhodnocování a ověřování nulových hodnot

Spark SQL (včetně SQL a rozhraní DATAFrame and Dataset API) nezaručuje pořadí vyhodnocení dílčích výrazů. Zejména vstupy operátoru nebo funkce nejsou nutně vyhodnoceny zleva doprava nebo v jiném pevném pořadí. Například logické výrazy AND a OR nemají sémantiku "krácení výpočtu" zleva doprava.

Proto je nebezpečné spoléhat se na vedlejší účinky nebo pořadí vyhodnocení logických výrazů, a také na pořadí klauzulí WHERE a HAVING, protože během optimalizace a plánování dotazů může dojít k jejich přeuspořádání. Konkrétně platí, že pokud UDF spoléhá na krátké spojení v SQL pro kontrolu hodnoty null, neexistuje žádná záruka, že se kontrola hodnoty null provede před vyvoláním UDF. Příklad:

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

Tato WHERE klauzule nezaručuje, že se UDF (uživatelsky definovaná funkce) strlen bude volat poté, co budou vyfiltrovány hodnoty null.

Pokud chcete provést správnou kontrolu hodnoty null, doporučujeme provést některou z následujících akcí:

  • Zajistěte, aby funkce definovaná uživatelem byla schopná pracovat s hodnotami null, a proveďte kontrolu hodnot null uvnitř samotné funkce definované uživatelem.
  • Použijte výrazy IF nebo CASE WHEN k ověření hodnoty null a vyvolání funkce definované uživatelem v podmíněné větvi
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

Přístup k tajemstvím katalogu Unity

Chcete-li získat přístup k tajnému klíči služby Unity Catalog z Python UDF s rozsahem relace, viz Použití tajného klíče v Python UDF s rozsahem relace. Pro přístup k deklarovaným tajemstvím ze skalárního nebo Batch Unity Catalog Python UDF viz Použít tajemství v Python UDF.

Přihlašovací údaje služeb v Python UDF

Relacně omezené skalární Python UDF a skalární Unity Catalog Python UDF mohou používat přihlašovací údaje Unity Catalog pro bezpečný přístup k externím cloudovým službám. To je užitečné pro integraci operací, jako je například cloudová tokenizace, šifrování nebo správa tajných kódů přímo do transformací dat.

Požadavky se liší podle typu UDF a výpočtové kapacity. Viz Použijte přihlašovací kartu služby v Python UDF.

Pokud chcete vytvořit přihlašovací údaje služby, přečtěte si téma Vytvoření přihlašovacích údajů služby.

Použijte služební přihlašovací údaje v relacistickém skalárním Python UDF

Pokud chcete získat přístup k přihlašovacím údajům služby, použijte databricks.service_credentials.getServiceCredentialsProvider() nástroj v logice UDF k inicializaci cloudových sad SDK s příslušnými přihlašovacími údaji. Veškerý kód musí být zapouzdřen v těle funkce 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

Oprávnění k přihlašovacím údajům služby

UDF s platností v rámci relace používají oprávnění volajícího. Viz Použít přihlašovací kartu v Python UDF pro požadovaná oprávnění.

Výchozí přihlašovací údaje na úrovni výpočtu pro relace omezené UDF

Při použití v skalárních Python UDF Databricks automaticky používá výchozí přihlašovací údaje služby z proměnné výpočetního prostředí. Toto chování umožňuje bezpečně odkazovat na externí služby bez explicitní správy aliasů přihlašovacích údajů v kódu UDF. Viz Zadání výchozích přihlašovacích údajů služby pro výpočetní prostředek

Výchozí podpora přihlašovacích údajů je dostupná jenom v clusterech standardního a vyhrazeného režimu přístupu. Není k dispozici v SQL warehousech.

Chcete-li použít poskytovatele azure-identity, musíte nainstalovat balíček DefaultAzureCredential. Pokud chcete balíček nainstalovat, přečtěte si téma Knihovny Pythonu s oborem poznámkového bloku nebo knihovny s vymezeným výpočetním oborem.

@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

Použijte přihlašovací údaje pro službu v skalárním Unity Catalog Python UDF

Specifikujte přihlašovací údaje služby v klauzuli CREDENTIALS definice UDF. Můžete označit jedno přihlašovací údaje DEFAULT , aby ho automaticky používaly opravené cloudové SDK. Na klasickém výpočtu tato funkce vyžaduje Databricks Runtime 18.1 nebo vyšší. U bezserverových výpočtů a v SQL skladech Pro a Serverless explicitně nastavte u UDF hodnotu environment_version na 6 nebo vyšší. Pro kompletní požadavky na výpočetní výkon, síťování a oprávnění viz Použijte přihlašovací údaje služby v Python UDF.

Získejte kontext provádění úkolu

Použijte PySpark API TaskContext k získání informací o kontextu, jako je identita uživatele, značky clusteru, ID úlohy Spark a další. Viz Získání kontextu úkolu v UDF.

Omezení

Následující omezení platí pro PySpark UDFs:

  • Omezení přístupu k souborům: Ve službě Databricks Runtime 14.2 a nižších nemohou PySpark UDFs na sdílených clusterech získat přístup ke složkám Gitu, souborům pracovního prostoru ani ke svazkům katalogu Unity.

  • proměnné všesměrového vysílání: UDF PySpark v clusterech standardního režimu přístupu a bezserverové výpočetní prostředky nepodporují proměnné všesměrového vysílání.

  • Omezení paměti na bezserverových platformách: Funkce definované uživatelem PySpark pro bezserverové platformy mají limit paměti 1 GB na PySpark UDF. Překročení tohoto limitu způsobí chybu typu UDF_PYSPARK_USER_CODE_ERROR. MEMORY_LIMIT_SERVERLESS.
  • Limit paměti ve standardním režimu přístupu: UDFy PySpark ve standardním režimu přístupu mají omezení paměti, které závisí na dostupné paměti zvoleného typu instance. Překročení dostupné paměti způsobí chybu typu UDF_PYSPARK_USER_CODE_ERROR. MEMORY_LIMIT.
  • Síťový přístup v bezserverových datových skladech SQL: Ve výchozím nastavení funkce UDF v Pythonu v bezserverových datových skladech SQL nemohou odesílat odchozí síťové požadavky a dotazy, které se pokusí o síťová volání, se zablokují na neurčito. Pokud chcete povolit odchozí síťový přístup, povolte ve svém pracovním prostoru na stránce Preview funkci Public Preview Povolit síťové připojení pro izolované úlohy v bezserverových skladech SQL. Jinak pro uživatelsky definované funkce, které vyžadují síťový přístup, použijte bezserverové výpočetní prostředí nebo klasické výpočetní prostředí.
  • Poslední aktualizace v