Felhasználó által definiált függvények a Pythonhoz készült Databricks Connectben

A Databricks Connect for Python támogatja a felhasználó által definiált függvényeket (UDF). Az UDF-eket tartalmazó DataFrame-művelet végrehajtásakor az UDF-eket a Databricks Connect szerializálja, és a kérés részeként elküldi a kiszolgálónak.

A Databricks Connect for Scalához készült UDF-ekről további információt a Databricks Connect for Scala felhasználó által definiált funkcióiban talál.

UDF definiálása

Ha UDF-t szeretne létrehozni a Pythonhoz készült Databricks Connectben, használja az alábbi támogatott függvények egyikét:

• PySpark felhasználó által definiált függvények
  • pyspark.sql.udf
  • pyspark.sql.pandas_udf
  • pyspark.sql.udtf
  • pyspark.sql.DataFrame.mapInPandas
  • pyspark.sql.DataFrame.mapInArrow
  • pyspark.sql.GroupedData.applyInPandas
  • pyspark.sql.GroupedData.applyInArrow (egy PySpark függvény neve)
  • pyspark.sql.PandasCogroupedOps.applyInPandas
  • pyspark.sql.PandasCogroupedOps.applyInArrow
• PySpark streamelési függvények
  • pyspark.sql.streaming.DataStreamWriter.foreach
  • pyspark.sql.streaming.DataStreamWriter.foreachBatch
  • pyspark.sql.streaming.StatefulProcessor

Az alábbi Python például beállít egy egyszerű UDF-t, amely négyzetre állítja az oszlop értékeit.

from pyspark.sql.functions import col, udf
from pyspark.sql.types import IntegerType
from databricks.connect import DatabricksSession

@udf(returnType=IntegerType())
def double(x):
    return x * x

spark = DatabricksSession.builder.getOrCreate()

df = spark.range(1, 2)
df = df.withColumn("doubled", double(col("id")))

df.show()

UDF-függőségek kezelése

A Databricks Connect támogatja az UDF-ekhez szükséges Python-függőségek megadását. Ezek a függőségek a Databricks-számításra vannak telepítve a UDF Python-környezetének részeként.

Ez a funkció lehetővé teszi a felhasználók számára, hogy az alapkörnyezetben biztosított csomagokon kívül olyan függőségeket is meghatározjanak, amelyekre az UDF-nek szüksége van. A csomag másik verziójának telepítésére is használható az alapkörnyezetben megadotttól.

A függőségi elemek a következő forrásokból telepíthetők:

• PyPI-csomagok
  • A PyPI-csomagok a PEP 508 szerint adhatók meg, dicepéldául . pyjokes<1simplejson==3.19.*
• Unity Catalog-kötetekben tárolt csomagok
  • A beépített disztribúciók (.whl) és a forráseloszlások (.tar.gz) egyaránt támogatottak.
  • A Unity Catalog-kötetek csomagjai dbfs:<path> megadhatóak például dbfs:/Volumes/users/someone@example.com/wheels/my_private_dep-3.20.2-py3-none-any.whl, vagy dbfs:/Volumes/users/someone@example.com/tars/my_private_dep-4.0.0.tar.gz.
  • A felhasználónak engedélyt kell kapnia READ_FILE a fájlra a re:[UC] kötetben. Ha ezt az engedélyt minden fiókfelhasználónak megadja, az automatikusan engedélyezi ezt az új felhasználók számára.
• Helyi csomagok, mappák és Python-fájlok
  • Helyi alapú disztribúciók (.whl), forráseloszlások (.tar.gz), mappák és Python-fájlok megadhatóak local:<path>például: local:/path/to/my_private_dep-3.20.2-py3-none-any.whl, local:/path/to/my_private_dep-4.0.0.tar.gz, local:/path/to/my_folder, local:/path/to/my_file.py.
  • Az abszolút és a relatív elérési utak is támogatottak, például: local:/path/to/my_file.py vagy local:./path/to/my_file.py.

Ha egyéni függőségeket szeretne belefoglalni a UDF-be, adja meg őket egy környezetben a withDependencies segítségével, majd ezt a környezetet használja egy Spark-munkamenet létrehozásához. A függőségek telepítve vannak a Databricks-számításra, és minden olyan UDF-ben elérhetők lesznek, amelyek ezt a Spark-munkamenetet használják.

A következő kód függőségként deklarálja a PyPI-csomagot dice :

from databricks.connect import DatabricksSession, DatabricksEnv
env = DatabricksEnv().withDependencies("dice==3.1.0")
spark = DatabricksSession.builder.withEnvironment(env).getOrCreate()

Vagy egy kötetben lévő kerék függőségének megadásához:

from databricks.connect import DatabricksSession, DatabricksEnv

env = DatabricksEnv().withDependencies("/Volumes/users/someone@example.com/wheels/my_private_dep-3.20.2-py3-none-any.whl")
spark = DatabricksSession.builder.withEnvironment(env).getOrCreate()

Viselkedés a Databricks-jegyzetfüzetekben és -feladatokban

Jegyzetfüzetekben és munkamenetekben az UDF-függőségeket közvetlenül a REPL-ben kell telepíteni. A Databricks Connect ellenőrzi a REPL Python-környezetet, és ellenőrzi, hogy az összes megadott függőség már telepítve van-e, és kivételt okoz, ha nincsenek telepítve. A jegyzetfüzet-környezet érvényesítése lefut mind a PyPI-, mind a Unity Catalog-csomagfüggőségek esetében, de nem a helyi függőségek esetében.

Korlátozások

• Az UDF-függőségek támogatása a pyspark.sql.streaming.DataStreamWriter.foreach Databricks Connect for Python 18.0-s vagy újabb verzióját, valamint a Databricks Runtime 18.0-s vagy újabb verzióját futtató fürt meglétét igényli.
• Az UDF-függőségek támogatása megköveteli a pyspark.sql.streaming.DataStreamWriter.foreachBatch Databricks Connect for Python 18.0-s vagy újabb verzióját, valamint a Databricks Runtime 18.0-s vagy újabb verzióját futtató fürtöt. A szolgáltatás kiszolgáló nélküli eszközökön nem támogatott.
• A helyi csomagok, mappák és Python-fájlok UDF-függőségeinek támogatása megköveteli a Databricks Connect for Python 18.1-es vagy újabb verzióját, valamint a Databricks Runtime 18.1-es vagy újabb verzióját futtató fürtöt.
• Az UDF-függőségek nem támogatottak a pandas-aggregációs UDF-ek esetében az ablakfüggvényeken keresztül.
• A Unity Catalog csomagokat és a helyi csomagokat a PEP-427 vagy újabb által meghatározott standard Python csomagolási előírások szerint kell csomagolni a wheel típusú disztribúciókhoz, a PEP-241 vagy újabb szerint pedig a tar forrás disztribúciókhoz. A Python csomagolási szabványokkal kapcsolatos további információkért tekintse meg a PyPA dokumentációját.

Példák

Az alábbi példa definiálja a PyPI- és kötetfüggőségeket egy környezetben, létrehoz egy munkamenetet a környezettel, majd definiálja és meghívja azokat a függőségeket használó UDF-eket:

from databricks.connect import DatabricksSession, DatabricksEnv
from pyspark.sql.functions import udf, col, pandas_udf
from pyspark.sql.types import IntegerType, LongType, StringType
import pandas as pd

pypi_deps = ["pyjokes>=0.8,<1"]

volumes_deps = [
    # Example library from: https://pypi.org/project/dice/#files
    "/Volumes/main/someone@example.com/test/dice-4.0.0.tar.gz",
]

local_deps = [
    # Example library from: https://pypi.org/project/simplejson/#files
    "local:./test/simplejson-3.20.2-py3-none-any.whl",
]

env = DatabricksEnv().withDependencies(pypi_deps).withDependencies(volumes_deps).withDependencies(local_deps)
spark = DatabricksSession.builder.withEnvironment(env).getOrCreate()

# UDFs
@udf(returnType=StringType())
def get_joke():
    from pyjokes import get_joke
    return get_joke()

@udf(returnType=IntegerType())
def double_and_json_parse(x):
    import simplejson
    return simplejson.loads(simplejson.dumps(x * 2))


@pandas_udf(returnType=LongType())
def multiply_and_add_roll(a: pd.Series, b: pd.Series) -> pd.Series:
    import dice
    return a * b + dice.roll(f"1d10")[0]


df = spark.range(1, 10)
df = df.withColumns({
    "joke": get_joke(),
    "doubled": double_and_json_parse(col("id")),
    "mutliplied_with_roll": multiply_and_add_roll(col("id"), col("doubled"))
})
df.show()

UDF-függőségek automatikus kezelése

A Databricks Connect withAutoDependencies() API lehetővé teszi az UDF-ek importálási utasításaiban használt helyi modulok és nyilvános PyPI-függőségek automatikus felderítését és feltöltését. Ez megszünteti a függőségek manuális megadásának szükségességét.

Az alábbi kód lehetővé teszi az automatikus függőségkezelést:

from databricks.connect import DatabricksSession, DatabricksEnv

env = DatabricksEnv().withAutoDependencies(upload_local=True, use_index=True)
spark = DatabricksSession.builder.withEnvironment(env).getOrCreate()

A withAutoDependencies() metódus a következő paramétereket fogadja el:

• upload_local: Ha be van Trueállítva, az UDF-ekben importált helyi modulok automatikusan felderítik, csomagolják és feltöltik az UDF-tesztkörnyezetbe.
• use_index: Ha be van állítva, az TrueUDF-ekben használt nyilvános PyPI-függőségek automatikusan felderítésre és telepítésre kerülnek az Azure Databricks compute-ben. A felderítési folyamat a helyi gépen telepített csomagokat használja a verziók meghatározásához, biztosítva a helyi környezet és a távoli végrehajtási környezet közötti konzisztenciát.

Korlátozások

• A dinamikus importálások (például importlib.import_module("foo")) nem támogatottak.
• A névtércsomagok (például azure.eventhub és google.cloud.aiplatform) nem támogatottak.
• A közvetlen URL-hivatkozások használatával telepített függőségek nem támogatottak. Ide tartoznak azok a fájlok, amelyek helyi wheel fájlokból lettek telepítve.
• A privát csomagindexekből telepített függőségek nem támogatottak. Az így telepített csomagok nem különböztethetők meg a nyilvános PyPI-ból telepített csomagoktól.
• A függőségfelderítés nem működik Python-rendszerhéjban. Csak a Python-szkriptek, az IPython-rendszerhéj és a Jupyter Notebook támogatott.

Példák

Az alábbi példa az automatikus függőségkezelést mutatja be a helyi modulokkal és a PyPI-csomagokkal. Ehhez a példához telepíteni simplejson és dice (használni pip install simplejson dice) kell.

Először hozzon létre helyi segédmodulokat:

# my_helper.py
def double(x):
    return 2 * x

# my_json.py
import simplejson

def loads(x):
    return simplejson.loads(x)

def dumps(x):
    return simplejson.dumps(x)

Ezután a fő szkriptben importálja ezeket a modulokat, és használja őket UDF-ben:

# main.py
import dice as dc
from databricks.connect import DatabricksSession, DatabricksEnv
from pyspark.sql.functions import col, udf
from pyspark.sql.types import IntegerType, FloatType

import my_json
from my_helper import double

env = DatabricksEnv().withAutoDependencies(upload_local=True, use_index=True)
spark = DatabricksSession.builder.withEnvironment(env).getOrCreate()

@udf(returnType=IntegerType())
def double_and_json_parse(x):
    return my_json.loads(my_json.dumps(double(x)))

@udf(returnType=FloatType())
def sum_and_add_noise(x, y):
    return x + y + (dc.roll("d6")[0] / 6)

df = spark.range(1, 10)
df = df.withColumns({
    "doubled": double_and_json_parse(col("id")),
    "summed_with_noise": sum_and_add_noise(col("id"), col("doubled")),
})
df.show()

Logging

A felderített függőségek kimenetéhez állítsa a(z) SPARK_CONNECT_LOG_LEVEL környezeti változót info vagy debug értékre. Másik lehetőségként konfigurálja a Python naplózási keretrendszerét:

import logging
logging.basicConfig(level=logging.INFO)

A megfelelő naplókat a databricks.connect.auto_dependencies modul bocsátja ki, például:

DEBUG:databricks.connect.auto_dependencies.discovery:Discovered local module: my_json
DEBUG:databricks.connect.auto_dependencies.discovery:Discovered local module: my_helper
DEBUG:databricks.connect.auto_dependencies.discovery:Discovered distribution: simplejson for module simplejson
DEBUG:databricks.connect.auto_dependencies.discovery:Discovered distribution: dice for module dice
INFO:databricks.connect.auto_dependencies.hook:Synced zip artifact for: my_json
INFO:databricks.connect.auto_dependencies.hook:Synced zip artifact for: my_helper
INFO:databricks.connect.auto_dependencies.hook:Updated simplejson with auto-detected version ==3.20.2
INFO:databricks.connect.auto_dependencies.hook:Updated dice with auto-detected version ==4.0.0

Python alapkörnyezet

Az UDF-ek végrehajtása a Databricks-számításon történik, nem az ügyfélen. Az alap Python-környezet, amelyben az UDF-ek végrehajtása történik, a Databricks-számítástól függ.

Fürtök esetén az alap Python-környezet a fürtön futó Databricks Runtime-verzió Python-környezete. A Python-verzió és az alapkörnyezetben található csomagok listája a Databricks Runtime kibocsátási megjegyzéseinekRendszerkörnyezet és Telepített Python-kódtárak szakaszában található.

Kiszolgáló nélküli számítás esetén az alap Python-környezet a kiszolgáló nélküli környezet verziójának felel meg az alábbi táblázat szerint. A táblázatban nem szereplő Databricks Connect-verziók vagy még nem támogatják a kiszolgáló nélküli verziót, vagy elérték a támogatás végét. Lásd a verziótámogatási mátrixot és a támogatott Databricks Connect-verziókat.