Batch Python felhasználó által definiált függvények (UDF-ek) a Unity Catalogban

A Batch Unity Catalog Python UDF-ek általában elérhetők. Adatkötegeken működnek, nem pedig egyszerre egy-egy soron.

Követelmények

Klasszikus számítástechnikán a Batch Unity Catalog Python UDF-ekhez Databricks Runtime 16.3 vagy annál magasabb verzió szükséges. Ezek a szerver nélküli számítási környezetben, valamint a pro és a szerver nélküli SQL-adattárházakban is támogatottak.

További képességeknek saját számítási és verziós követelményei vannak. Lásd a Python UDF funkciókövetelményeket.

Hozzon létre egy batch Unity Catalog Python UDF

Egy Batch Unity Catalog Python UDF létrehozása hasonló egy normál Unity Catalog UDF létrehozásához, az alábbi kiegészítésekkel:

• PARAMETER STYLE PANDAS: Ez azt határozza meg, hogy a UDF a pandas iterátorokkal dolgozza fel a kötegekben lévő adatokat.
• HANDLER 'handler_function': Ez megadja a kezelő függvényt, amely a köteteket feldolgozza.

A következő példa egy tartós Batch Python UDF-et hoz létre a Unity Catalogban. Cseréld my_catalog le a katalógusoddal my_schema és a sémáddal:

%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)
$$;

A UDF regisztrálása után SQL-t vagy Python-t használva hívhatja meg.

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

Batch UDF kezelőfüggvény

A Batch Unity Catalog Python UDF-ekhez egy kezelőfüggvény szükséges, amely feldolgozza a csomagokat és eredményeket ad vissza. Az UDF létrehozásakor a HANDLER záradék használatával meg kell adnia a kezelőfüggvény nevét.

A kezelő függvény a következőket hajtja végre:

1. Elfogad egy iterátorargumentumot, amely egy vagy több pandas.Series elemen iterál. Minden pandas.Series tartalmazza a UDF bemeneti paramétereit.
2. Végigiterál a generátoron és feldolgozza az adatokat.
3. Visszaad egy generátor iterátort.

A Batch Unity Catalog Python UDF-eknek ugyanannyi sort kell visszaadniuk, mint amennyi a bemenet. A kezelőfüggvény ezt úgy biztosítja, hogy pandas.Series-t ad vissza, amely ugyanakkora hosszúságú, mint a bemeneti sorozatok minden egyes köteg esetében.

Egyéni függőségek telepítése

A Batch Unity Catalog Python UDF-k funkcionalitását kibővítheti a Databricks Runtime környezetén kívül is azzal, hogy egyedi függőségeket definiál az külső könyvtárakhoz.

Lásd: UDF-ek kiterjesztése egyéni függőségekkel.

Hozzáférés Unity katalógus titkaihoz

Batch Unity Catalog Python UDF-ek hozzáférhetnek a záradékban bejelentett SECRETS titkokhoz. A(z) environment_version értékét kifejezetten 6 vagy annál magasabb értékre kell állítani. Egy ilyen záradékot használó UDF közvetlen meghívása nem támogatott dedikált hozzáférési módú számítási rendszeren. Az oszlopmaszk kivétel esetén lásd: Titkos engedélyezett UDF-eket használunk oszlopmaszkokban dedikált számításon.

A Batch UDF-k elfogadhatnak egy vagy több paramétert

Egyetlen paraméter: Ha a kezelő függvény egyetlen bemeneti paramétert használ, minden tételhez egy iterátort kap az a pandas.Series felett.

%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;

Több paraméter: Több bemeneti paraméter esetén a kezelő függvény egy iterátort kap, amely többször átmegy a pandas.Series-en. Az adatsor értékei ugyanabban a sorrendben vannak, mint a bemeneti paraméterek.

%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);

Optimalizálja a teljesítményt a költséges műveletek szétválasztásával

A számításigényes műveletek optimalizálását azáltal érheti el, hogy elkülöníti ezeket a műveleteket a kezelő függvénytől. Ez biztosítja, hogy a végrehajtásuk csak egyszer történik, nem pedig minden adatkötegen keresztüli iteráció során.

A következő példa bemutatja, hogyan lehet biztosítani, hogy egy drága számítás csak egyszer történjen meg.

%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

Környezet elkülönítése

Megjegyzés:

A megosztott elkülönítési környezetekhez a Databricks Runtime 17.1-s vagy újabb verziója szükséges. A korábbi verziókban az összes Batch Unity Catalog Python UDF szigorú elkülönítési módban fut.

Az azonos tulajdonossal és munkamenettel rendelkező Batch Unity Catalog Python UDF-jei alapértelmezés szerint megoszthatnak egy elkülönítési környezetet. Ez javíthatja a teljesítményt és csökkentheti a memóriahasználatot az elindítandó különálló környezetek számának csökkentésével.

Szigorú elkülönítés

Annak érdekében, hogy az UDF mindig a saját, teljesen elkülönített környezetben fusson, adja hozzá a STRICT ISOLATION jellemző záradékot.

A legtöbb UDF-nek nincs szüksége szigorú elkülönítésre. A standard adatfeldolgozási UDF-ek kihasználják az alapértelmezett megosztott elkülönítési környezetet, és gyorsabban futnak alacsonyabb memóriahasználat mellett.

Adja hozzá a STRICT ISOLATION jellemző záradékot az olyan UDF-ekhez, amelyek:

• Bemenet futtatása kódként eval(), exec()vagy hasonló függvényekkel
• Fájlok írása a helyi fájlrendszerbe
• Globális változók vagy rendszerállapot módosítása
• Környezeti változók módosítása

Az alábbi példa egy olyan UDF-t mutat be, amely kódként hajtja végre a bemenetet, és szigorú elkülönítést igényel:

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)
$$;

Szolgáltatás hitelesítő adatai a Batch Unity Catalog Python UDF-jeiben

A Batch Unity Catalog Python UDF-ek Unity Catalog szolgáltatási hitelesítő adatait használhatják külső felhőszolgáltatásokhoz való hozzáféréshez. Ez különösen hasznos a felhőbeli függvények, például a biztonsági tokenizerek adatfeldolgozási munkafolyamatokba való integrálásához.

Megjegyzés:

A szolgáltatás hitelesítő adataihoz tartozó UDF-specifikus API:
Az UDF-ekben használja databricks.service_credentials.getServiceCredentialsProvider() a szolgáltatás hitelesítő adatainak eléréséhez.

Ez eltér a dbutils.credentials.getServiceCredentialsProvider() jegyzetfüzetekben használt függvénytől, amely nem érhető el UDF-végrehajtási környezetekben.

Szolgáltatás hitelesítő adatainak létrehozásáról a Szolgáltatás hitelesítő adatainak létrehozása című témakörben olvashat.

Adja meg az UDF-definíció záradékában CREDENTIALS használni kívánt szolgáltatás-hitelesítő adatokat:

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
$$;

Szolgáltatási hitelesítő adatok jogosultságai

A számítási típusok különböző létrehozási és hívói engedélyigényeiről lásd: Használj szolgáltatási hitelminősítést a Python UDF-ben.

Alapértelmezett hitelesítési adatok és álnevek

Több hitelesítési adatot is beilleszthet a CREDENTIALS záradékba, de csak egyet jelölhet meg DEFAULT. A nem alapértelmezett hitelesítő adatokat álnévvel láthatja el a AS kulcsszó használatával. Minden hitelesítőnek egyedi aliaszal kell rendelkeznie.

A javított felhő SDK-k automatikusan felveszik az alapértelmezett hitelesítő adatokat. Az alapértelmezett hitelesítő adatok elsőbbséget élveznek a számítástechnika Spark konfigurációjában megadott bármely alapértelmezett beállítással szemben, és megmaradnak a Unity Catalog UDF definíciójában.

Ahhoz, hogy használni tudd a azure-identity szolgáltatót, telepítened kell a DefaultAzureCredential csomagot. Használja a ENVIRONMENT záradékot külső könyvtárak telepítéséhez. További információkat az külső könyvtárak telepítéséhez az Egyedi függőségek használatával kiterjesztett UDF-ek oldalon találhat.

Szolgáltatás hitelesítő adatainak példája – Azure Blob Storage

Az alábbi példa szolgáltatás-hitelesítő adatok használatával fér hozzá az Azure Blob Storage-hoz egy Batch Unity Catalog Python UDF-ből:

%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)
$$;

Hívja meg az UDF-et, miután regisztrálták.

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

Szerezze meg a feladat végrehajtási kontextust

Használja a TaskContext PySpark API-t a kontextus információk lekéréséhez, például a felhasználóazonosság, a klasztercímkék, a Spark feladat azonosító és további adatok megszerzéséhez. Lásd: Feladat kontextus lekérése egy UDF-ben.

Állítsa be DETERMINISTIC, ha a függvény következetes eredményeket hoz létre.

Adja hozzá DETERMINISTIC a függvénydefinícióhoz, ha ugyanazokat a kimeneteket állítja elő ugyanazokhoz a bemenetekhez. Ez lehetővé teszi a lekérdezésoptimalizálást a teljesítmény javításához.

Alapértelmezés szerint a Batch Unity Catalog Python UDTF-jei nem determinisztikusnak minősülnek, kivéve, ha explicit módon deklarálják. A nem determinisztikus függvények például a következők: véletlenszerű értékek generálása, az aktuális időpontok vagy dátumok elérése, vagy külső API-hívások indítása.

Lásd CREATE FUNCTION : (SQL, Python, Scala és Java)

Korlátozások

• A Python függvényeknek önállóan kell kezelniük a NULL értékeket, és az összes típusleképezésnek követnie kell az Azure Databricks SQL nyelvi leképezéseit.
• Az adagolt Unity Catalog Python UDF-ek biztonságos, elszigetelt környezetben futnak, és nem férnek hozzá megosztott fájlrendszerhez vagy belső szolgáltatásokhoz.
• Egy szakaszon belüli többszörös UDF-hívások sorosítottak, és a közbenső eredmények anyaggá alakítódnak, és esetenként a lemezre kerülhetnek.

• A Batch Unity Catalog Python UDF-hívások kiszolgáló nélküli notebookon vagy feladatszámításon való indításához konfigurálnia kell a kiszolgáló nélküli kimenő forgalom vezérlését