Batch Python användardefinierade funktioner (UDF: er) i Unity Catalog

Batch Unity Catalog Python UDF:er är nu allmänt tillgängliga. De arbetar med datamängder i stället för en rad i taget.

Krav

På klassisk beräkning kräver Batch Unity Catalog Python UDF:er Databricks Runtime 16.3 eller högre. De stöds också i serverless compute och i pro- och serverless-SQL-lager.

Ytterligare funktioner har egna beräknings- och versionskrav. Se Python UDF:s funktionskrav.

Skapa en batch-Python-UDF i Unity Catalog

Att skapa en Batch Unity Catalog Python UDF liknar att skapa en vanlig UDF för Unity Catalog med följande tillägg:

• PARAMETER STYLE PANDAS: Detta anger att UDF behandlar data i grupper med hjälp av Pandas-iteratorer.
• HANDLER 'handler_function': Detta specificerar den hanterarfunktion som bearbetar batcharna.

Följande exempel skapar en persistent Batch Python UDF i Unity-katalogen. Ersätt my_catalog och my_schema med din katalog och schema:

%sql
CREATE OR REPLACE FUNCTION my_catalog.my_schema.calculate_bmi_pandas(weight_kg DOUBLE, height_m DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
DETERMINISTIC
PARAMETER STYLE PANDAS
HANDLER 'handler_function'
AS $$
import pandas as pd
from typing import Iterator, Tuple

def handler_function(batch_iter: Iterator[Tuple[pd.Series, pd.Series]]) -> Iterator[pd.Series]:
  for weight_series, height_series in batch_iter:
    yield weight_series / (height_series ** 2)
$$;

Efter att ha registrerat UDF:en kan du anropa den med SQL eller Python.

SELECT person_id, my_catalog.my_schema.calculate_bmi_pandas(weight_kg, height_m) AS bmi
FROM (
  SELECT 1 AS person_id, CAST(70.0 AS DOUBLE) AS weight_kg, CAST(1.75 AS DOUBLE) AS height_m UNION ALL
  SELECT 2 AS person_id, CAST(80.0 AS DOUBLE) AS weight_kg, CAST(1.80 AS DOUBLE) AS height_m
);

Hanteringsfunktion för batch-UDF

Batch Unity Catalog Python UDF:er kräver en hanterarfunktion som bearbetar batcher och ger resultat. Du måste ange namnet på handlerfunktionen när du skapar UDF genom att använda klausulen HANDLER .

Hanteringsfunktionen gör följande:

1. Accepterar ett iteratorargument som itererar över en eller flera pandas.Series. Varje pandas.Series innehåller indatamängder för UDF.
2. Itererar över generatorn och bearbetar data.
3. Returnerar en generator-iterator.

Batch Unity Catalog Python UDF:er måste returnera samma antal rader som inmatningen. Hanterarfunktionen säkerställer detta genom att avge en pandas.Series med samma längd som inmatningsserien för varje batch.

Installera anpassade beroenden

Du kan utöka funktionerna i Batch Unity Catalog Python UDF:er utöver Databricks Runtime-miljön genom att definiera anpassade beroenden för externa bibliotek.

Se Utöka UDF:er med hjälp av anpassade beroenden.

Access Unity-katalogens hemligheter

Batch Unity Catalog Python UDF:er kan komma åt hemligheter som deklareras i klausulenSECRETS. Du måste uttryckligen ställa in environment_version till 6 eller högre. Direkt anrop av en UDF som använder denna klausul stöds inte i dedikerad åtkomstlägesberäkning. För undantaget för kolumnmasker, se Använd UDF:er med stöd för hemligheter i kolumnmasker på dedikerad beräkningsresurs.

Batch-UDF:er kan acceptera enskilda eller flera parametrar

Enkel parameter: När handlerfunktionen använder en enda indataparameter får den en iterator över a pandas.Series för varje batch.

%sql
CREATE OR REPLACE TEMPORARY FUNCTION one_parameter_udf(value INT)
RETURNS STRING
LANGUAGE PYTHON
DETERMINISTIC
PARAMETER STYLE PANDAS
HANDLER 'handler_func'
AS $$
import pandas as pd
from typing import Iterator
def handler_func(batch_iter: Iterator[pd.Series]) -> Iterator[pd.Series]:
  for value_batch in batch_iter:
    d = {"min": value_batch.min(), "max": value_batch.max()}
    yield pd.Series([str(d)] * len(value_batch))
$$;
SELECT one_parameter_udf(id), count(*) from range(0, 100000, 3, 8) GROUP BY ALL;

Flera parametrar: För flera indata-parametrar, tar funktionens hanterare emot en iterator som itererar över flera pandas.Series. Värdena i serien är i samma ordning som inmatningsparametrarna.

%sql
CREATE OR REPLACE TEMPORARY FUNCTION two_parameter_udf(p1 INT, p2 INT)
RETURNS INT
LANGUAGE PYTHON
DETERMINISTIC
PARAMETER STYLE PANDAS
HANDLER 'handler_function'
AS $$
import pandas as pd
from typing import Iterator, Tuple

def handler_function(batch_iter: Iterator[Tuple[pd.Series, pd.Series]]) -> Iterator[pd.Series]:
  for p1, p2 in batch_iter: # same order as arguments above
    yield p1 + p2
$$;
SELECT two_parameter_udf(id , id + 1) from range(0, 100000, 3, 8);

Optimera prestandan genom att separera kostsamma operationer

Du kan optimera beräkningsintensiva operationer genom att separera dessa operationer från hanterarfunktionen. Detta säkerställer att de bara körs en gång i stället för under varje iteration över batchar med data.

I följande exempel visas hur du ser till att en dyr beräkning endast utförs en gång:

%sql
CREATE OR REPLACE TEMPORARY FUNCTION expensive_computation_udf(value INT)
RETURNS INT
LANGUAGE PYTHON
DETERMINISTIC
PARAMETER STYLE PANDAS
HANDLER 'handler_func'
AS $$
def compute_value():
  # expensive computation...
  return 1

expensive_value = compute_value()
def handler_func(batch_iter):
  for batch in batch_iter:
    yield batch * expensive_value
$$;
SELECT expensive_computation_udf(id), count(*) from range(0, 100000, 3, 8) GROUP BY ALL

Miljöisolering

Batch Unity Catalog Python UDF:er med samma ägare och session kan dela en isoleringsmiljö som standard. Detta kan förbättra prestanda och minska minnesanvändningen genom att minska antalet separata miljöer som måste startas.

Strikt isolering

Lägg till STRICT ISOLATION egenskapssatsen för att säkerställa att en UDF alltid körs i en egen, helt isolerad miljö.

De flesta UDF:er behöver inte strikt isolering. UDF:er för standarddatabearbetning drar nytta av standardmiljön för delad isolering och körs snabbare med lägre minnesförbrukning.

Lägg till egenskapssatsen STRICT ISOLATION i UDF:er som:

• Kör indata som kod med hjälp av eval(), exec()eller liknande funktioner
• Skriva filer till det lokala filsystemet
• Ändra globala variabler eller systemtillstånd
• Ändra miljövariabler

I följande exempel visas en UDF som kör indata som kod och kräver strikt isolering:

CREATE OR REPLACE TEMPORARY FUNCTION eval_string(input STRING)
RETURNS STRING
LANGUAGE PYTHON
PARAMETER STYLE PANDAS
HANDLER 'handler_func'
STRICT ISOLATION
AS $$
import pandas as pd
from typing import Iterator

def handler_func(batch_iter: Iterator[pd.Series]) -> Iterator[pd.Series]:
  for code_series in batch_iter:
    def eval_func(code):
      try:
        return str(eval(code))
      except Exception as e:
        return f"Error: {e}"
    yield code_series.apply(eval_func)
$$;

Tjänstautentiseringsuppgifter i Batch Unity Catalog Python UDF:er

Batch Unity Catalog Python UDF kan använda Unity Catalog-tjänsteuppgifter för att komma åt externa molntjänster. Detta är särskilt användbart för att integrera molnfunktioner som säkerhetstokeniserare i arbetsflöden för databehandling.

Information om hur du skapar en tjänstautentiseringsuppgift finns i Skapa autentiseringsuppgifter för tjänsten.

Ange den tjänsteautentisering du vill använda i CREDENTIALS-klausulen i UDF-definitionen.

CREATE OR REPLACE TEMPORARY FUNCTION example_udf(data STRING)
RETURNS STRING
LANGUAGE PYTHON
PARAMETER STYLE PANDAS
HANDLER 'handler_function'
CREDENTIALS (
  `credential-name` DEFAULT,
  `complicated-credential-name` AS short_name,
  `simple-cred`,
  cred_no_quotes
)
AS $$
# Python code here
$$;

Tjänstautentiseringstillstånd

För krav på skapande och anroparbehörighet över beräkningstyper, se Använd en tjänstelegitimation i en Python UDF.

Förinställda uppgifter och aliaser

Du kan inkludera flera referenser i CREDENTIALS-satsen, men endast en kan markeras som DEFAULT. Du kan använda alias för icke-standardbehörigheter med hjälp av AS nyckelordet. Varje referens måste ha ett unikt alias.

Patchede moln-SDK:er hämtar automatiskt standardautentiseringsuppgifter. Standardautentiseringen har företräde framför alla standardvärden som anges i compute-enhetens Spark-konfiguration och kvarstår i Unity Catalog UDF-definitionen.

Du måste installera azure-identity paketen för att använda DefaultAzureCredential leverantören. Använd ENVIRONMENT klausulen för att installera externa bibliotek. För att lära dig mer om att installera externa bibliotek, se Utöka UDF:er med anpassade beroenden.

Exempel på tjänstautentiseringsuppgifter – Azure Blob Storage

I följande exempel används en tjänstautentiseringsuppgift för att få åtkomst till Azure Blob Storage från en Batch Unity Catalog Python UDF:

%sql
CREATE OR REPLACE FUNCTION main.test.read_azure_blob(blob_name STRING) RETURNS STRING LANGUAGE PYTHON
PARAMETER STYLE PANDAS
HANDLER 'batchhandler'
CREDENTIALS (
  `batch-udf-service-creds-example-cred` DEFAULT
)
ENVIRONMENT (
  dependencies = '["azure-identity", "azure-storage-blob"]', environment_version = 'None'
)
AS $$
import pandas as pd
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient


def batchhandler(it):
  # DefaultAzureCredential automatically uses the DEFAULT service credential
  credential = DefaultAzureCredential()
  blob_service = BlobServiceClient(
    account_url="https://your-storage-account.blob.core.windows.net",
    credential=credential
  )
  container = blob_service.get_container_client("your-container")

  for blob_names in it:
    results = []
    for name in blob_names:
      blob_client = container.get_blob_client(name)
      try:
        content = blob_client.download_blob().readall().decode("utf-8")
        results.append(content)
      except Exception as e:
        results.append(f"Error: {e}")
    yield pd.Series(results)
$$;

Ring UDF efter att den har registrerats:

SELECT main.test.read_azure_blob(blob_name)
FROM VALUES
('config/settings.json'),
('data/input.txt')
AS t(blob_name)

Hämta uppgiftskörningskontext

Använd TaskContext PySpark API för att få kontextinformation såsom användarens identitet, klustertaggar, spark-jobb-ID och mer. Se Hämta uppgiftskontext i en UDF.

Ange DETERMINISTIC om funktionen ger konsekventa resultat

Lägg till DETERMINISTIC i funktionsdefinitionen om den genererar samma utdata för samma indata. På så sätt kan frågeoptimeringar förbättra prestandan.

Som standard antas Batch Unity Catalog Python-UDF:er vara icke-deterministiska om de inte uttryckligen deklareras. Exempel på icke-deterministiska funktioner är: generera slumpmässiga värden, komma åt aktuella tider eller datum eller göra externa API-anrop.

Se CREATE FUNCTION (SQL, Python, Scala och Java)

Begränsningar

• Python-funktioner måste hantera NULL-värden oberoende, och alla typmappningar måste följa Azure Databricks SQL-språkets mappningar.
• Batch Unity Catalog Python UDF:er körs i en säker, isolerad miljö och har inte tillgång till ett delat filsystem eller interna tjänster.
• Flera UDF-anrop inom ett steg är serialiserade och mellanresultat är materialiserade och kan överföras till disk.

• Om du vill göra Batch Unity Catalog Python UDF-anrop på en serverlös anteckningsbok eller jobbberäkning måste du konfigurera serverlös utflödeskontroll