Примечание.
Для доступа к этой странице требуется авторизация. Вы можете попробовать войти или изменить каталоги.
Для доступа к этой странице требуется авторизация. Вы можете попробовать изменить каталоги.
Important
Пользовательские функции Scala и Java (UDF) можно зарегистрировать в Unity Catalog для централизованного управления, повторного использования и удобного поиска. См. пользовательские функции (UDF) Scala и Java в Unity Catalog.
На этой странице описано, как создавать пользовательские функции Scala и Java уровня сеанса в Azure Databricks. Пользовательские функции (UDF) с областью действия в рамках сеанса определяются в записной книжке или задании и действуют только в пределах текущего SparkSession. Справочник по языку SQL см. в статье Внешние скалярные пользовательские функции (UDF).
Выбор подхода
Вы можете определить UDF на Scala или Java следующими способами. Чтобы сравнить все типы UDF в разных языках, режимах управления и вычислительных средах, см. UDF под управлением Unity Catalog и UDF с областью действия сеанса.
| Approach | Описание |
|---|---|
| Встроенная UDF Scala | Определите сессионный UDF с использованием функции Scala или лямбда. Поддерживается в ноутбуках Scala на классических вычислениях и в бессерверных JAR-задачах с использованием среды версии 4 и выше. |
| Java UDF из JAR-файла | Зарегистрировать заранее скомпилированный UDF-класс с сессионным диапазоном из JAR с использованием spark.udf.registerJavaFunction. Поддерживается в бессерверных JAR-заданиях. |
| Управляемый каталогом Unity Scala или Java UDF | Зарегистрируйте UDF в каталоге Unity для управления, повторного использования и обнаружения. Поддерживается в бессерверной вычислительной среде. |
Требования
- UDF-функции Scala в вычислительных ресурсах с поддержкой Unity Catalog и со стандартным режимом доступа требуют Databricks Runtime версии 14.2 или выше.
- Для поддержки Scala UDFs на экземплярах ARM в кластерах с поддержкой каталога Unity требуется Databricks Runtime версии 15.2 или выше.
- Для регистрации Java UDF из JAR-файла с помощью
spark.udf.registerJavaFunctionтребуется Databricks Runtime 18 LTS или выше. См. Регистрация Java UDF из JAR-файла. - Для доступа к секрету Unity Catalog из UDF Scala уровня сеанса требуется Databricks Runtime 19 или более поздней версии в классической вычислительной среде со стандартным режимом доступа. В бессерверной JAR-задаче доступ к секретам требует среды версии 6 или выше. См. требования и разрешения UDF.
Important
Создайте JAR-файл с теми же версиями Scala и Apache Spark, что и вычислительные ресурсы, выполняющие его. Несоответствие может привести к сбою в работе UDF при регистрации или вызове.
- Классические вычислительные ресурсы: сопоставьте версии Scala и Spark с версией Databricks Runtime. См. раздел Системная среда в примечаниях к выпуску Databricks Runtime: версии и совместимость для вашей версии. Например, Databricks Runtime 18 LTS использует Scala 2.13.16 и Apache Spark 4.0.
- Бессерверные вычисления: версия Scala должна соответствовать версии среды. См. Версии среды.
Пометьте зависимость Apache Spark как provided, чтобы она не включалась в ваш JAR-файл. Включайте только сторонние зависимости, которые использует ваша UDF.
Зарегистрируйте функцию как UDF
Регистрация функции Scala в качестве UDF с помощью spark.udf.register:
val squared = (s: Long) => {
s * s
}
spark.udf.register("square", squared)
Вызовите UDF в Spark SQL
Создайте временное представление и вызовите UDF в SQL-запросе:
spark.range(1, 20).createOrReplaceTempView("test")
%sql select id, square(id) as id_squared from test
Использование UDF с DataFrames
Вы также можете вызвать UDF с помощью API DataFrame:
import org.apache.spark.sql.functions.{col, udf}
val squared = udf((s: Long) => s * s)
display(spark.range(1, 20).select(squared(col("id")) as "id_squared"))
Доступ к секретному каталогу Unity от сессионного UDF Scala
UDF Scala с областью действия сеанса может получить секрет из Unity Catalog с помощью com.databricks.Secrets.get(catalog, schema, key). Секретный доступ использует права звонящего. См. требования и разрешения UDF для необходимых прав.
Следующий пример проверяет, настроен ли секрет без возврата значения секрета:
import org.apache.spark.sql.functions.udf
val secretIsConfigured = udf((_: Int) => {
val apiKey = com.databricks.Secrets.get("main", "default", "api_key")
apiKey != null && apiKey.nonEmpty
})
spark.udf.register("secret_is_configured", secretIsConfigured)
spark.sql("SELECT secret_is_configured(1)").show()
Не возвращайте секретные значения из функции UDF. См. Секреты в Unity Catalog.
Файлы с UDF
Important
Эта функция доступна в бета-версии. Администраторы рабочей области могут управлять доступом к этой функции на странице "Предварительные версии ". См. статью "Управление предварительными версиями Azure Databricks".
Тип Scala для файла — FileRef. Используйте его как параметр или тип возврата в UDF, либо как тип верхнего уровня, либо вложенный. Сведения о типе и правилах его вложения см. в FILEтип.
Чтобы прочитать содержимое файла в UDF, вызовите одну из следующих кнопок на FileRef:
-
asLocalFile(): Возвращаетjava.io.File, который вы можете передать в любую библиотеку, которая принимает путь. -
open(): Возвращаетjava.io.InputStreamдля чтения байтов файла. Звонящий закрывает его.
Чтобы получить новый FileRef в UDF, вызовем один из следующих статических методов:
-
FileRef.create(uri): Создаёт ссылку на файл вuri. -
FileRef.fromBytes(bytes, destinationPath, contentType): Загружаетbytesпо пути томаdestinationPathкак внешний файл и возвращает ссылку. -
FileRef.fromLocalFile(localFile, destinationPath, contentType): Загружает локальный файл по пути томаdestinationPathкак внешний файл и возвращает ссылку.
Возврат значения FileRef из UDF, записывающей в столбец FILE MANAGED, не поддерживается.
Для примеров в Python, Scala и SQL, включая обработку изображений, обнаружение типов файлов и извлечение видеокадров, см. раздел Процессные файлы с UDF.
Регистрация Java UDF из JAR-файла
Упакуйте UDF в JAR-файл, добавьте его в текущий сеанс с помощью spark.addArtifact и зарегистрируйте класс UDF с помощью spark.udf.registerJavaFunction.
Примечание.
Поддерживается в стандартном режиме доступа и бессерверных вычислениях в Databricks Runtime 18 LTS или более поздней версии. Зарегистрированная функция имеет область видимости в пределах сеанса и не зарегистрирована в Unity Catalog.
Ниже описано, как создать проект, написать класс UDF, создать толстый JAR-файл и зарегистрировать его.
Шаг 1. Создание проекта
Настройте проект в Scala или Java.
Scala
Создание проекта Scala с помощью sbt:
sbt new scala/scala-seed.g8
Замените содержимое файла build.sbt следующим образом. Задайте scalaVersion и версию spark-sql в соответствии с вашими вычислительными ресурсами:
scalaVersion := "2.13.16"
ThisBuild / organization := "com.example"
lazy val myUDF = (project in file("."))
.settings(
name := "my-udf",
libraryDependencies += "org.apache.spark" %% "spark-sql" % "4.0.0" % "provided"
)
Включите плагин sbt-assembly для создания толстого JAR-файла. Создайте или измените project/assembly.sbt и добавьте:
addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.0.0")
Java
Создайте проект Maven с помощью архетипа быстрого запуска:
mvn archetype:generate \
-DgroupId=com.example \
-DartifactId=my-udf \
-DarchetypeArtifactId=maven-archetype-quickstart \
-DinteractiveMode=false
Эта команда создает стандартную структуру проекта Maven с src/main/java каталогами и src/test/java каталогами.
В созданных pom.xml<project></project> тегах добавьте <properties> блок и настройте maven-shade-plugin для сборки fat JAR:
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.5.0</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
Шаг 2: Напишите свой класс UDF
Класс UDF должен реализовать один из org.apache.spark.sql.api.java.UDF интерфейсов (UDF1 через UDF22), где число указывает, сколько входных аргументов принимает UDF. Реализуйте метод call(), используя свою логику.
Обработчик должен быть классом Java.
spark.udf.registerJavaFunction загружает класс через рефлексию, поэтому это должен быть либо public-класс верхнего уровня, либо static вложенный public-класс с публичным конструктором без аргументов. Объект Scala class или object не соответствует этому требованию и приводит к сбою при вызове. Вы можете создать JAR-файл с помощью sbt, но сам класс UDF должен быть записан в Java.
Создайте src/main/java/com/example/MyIntegerUDF.java:
package com.example;
import org.apache.spark.sql.api.java.UDF1;
public class MyIntegerUDF implements UDF1<Integer, Integer> {
@Override
public Integer call(Integer x) {
return x + 1;
}
}
Шаг 3: Создайте свой fat JAR-файл
Упаковайте скомпилированный UDF в толстый JAR-файл.
Scala
Из корневого каталога проекта выполните следующую команду:
sbt clean assembly
Толстый JAR-файл создается в target/scala-2.13/ под именем, например my-udf-assembly-0.1.0-SNAPSHOT.jar.
Java
Из корневого каталога проекта выполните следующую команду:
mvn clean package
Толстый JAR-файл создается в target/ под именем, например my-udf-1.0-SNAPSHOT.jar.
Шаг 4. Отправка JAR-файла в том каталога Unity
Загрузите JAR-файл в том Unity Catalog, чтобы ваши вычисления могли получить к нему доступ. Если у вас еще нет тома, создайте его:
CREATE VOLUME IF NOT EXISTS my_catalog.my_schema.udf_jars
COMMENT 'Storage for UDF JAR files';
Отправьте JAR-файл в том с помощью обозревателя каталогов:
- В рабочей области Azure Databricks щелкните
Catalog, чтобы открыть обозреватель каталогов.
- Выберите каталог, а затем выберите схему, содержащую том.
- Щелкните имя тома.
- Нажмите кнопку "Отправить в этот том " и выберите JAR-файл.
- Нажмите кнопку Отправить.
- После завершения отправки щелкните имя JAR-файла, а затем нажмите кнопку "Копировать путь ", чтобы скопировать путь тома. Например:
/Volumes/my_catalog/my_schema/udf_jars/my-udf-assembly-0.1.0-SNAPSHOT.jar. Этот путь понадобится вам на следующем шаге.
Шаг 5. Регистрация и вызов UDF
Добавьте JAR-файл в сеанс с помощью пути тома, зарегистрируйте класс UDF и вызовите его из Spark SQL:
# Add the JAR containing your UDF class to the session
spark.addArtifact("/Volumes/my_catalog/my_schema/udf_jars/my-udf-assembly-0.1.0-SNAPSHOT.jar")
# Register the UDF class, providing the SQL function name,
# the fully qualified class name, and the return type
from pyspark.sql.types import IntegerType
spark.udf.registerJavaFunction(
"my_udf",
"com.example.MyIntegerUDF",
IntegerType(),
)
# Call the UDF from Spark SQL
spark.sql("SELECT my_udf(21)").show()
При вычислении бессерверного и стандартного режима доступа необходимо передать явный тип возвращаемого значения. Пропуск возвращаемого типа приводит к UC_COMMAND_NOT_SUPPORTED_IN_SHARED_ACCESS_MODEошибке. Определяемые пользователем агрегатные функции (UDAFs) не поддерживаются в registerJavaFunction.
Запрос возвращает выходные данные UDF, подтверждающие, что функция зарегистрирована и вызываема:
+----------+
| my_udf(21)|
+----------+
| 22|
+----------+
Порядок вычисления и проверка на NULL
Spark SQL (включая SQL и API DataFrame и Dataset) не гарантирует порядок вычисления подвыражений. Spark не оценивает входные данные оператора или функции слева направо. Логические AND и OR выражения не имеют семантики короткого замыкания слева направо.
Не полагайтесь на побочные эффекты, порядок вычисления булевых выражений или порядок следования частей WHERE и HAVING. Оптимизатор запросов может изменить порядок этих выражений и предложений. Если UDF опирается на семантику короткого замыкания для проверки на null, Spark не гарантирует, что проверка на null будет выполнена до вызова UDF. Рассмотрим пример.
spark.udf.register("strlen", (s: String) => s.length)
spark.sql("select s from test1 where s is not null and strlen(s) > 1") // no guarantee
Это WHERE условие не гарантирует, что Spark вызывает выполнение strlen UDF-функции после того, как он отфильтровывает значения NULL.
Для проверки на NULL Databricks рекомендует использовать любой из следующих способов:
- Сделайте саму UDF способной обрабатывать значения NULL и выполняйте проверку на NULL внутри UDF
- Используйте выражения
IFилиCASE WHENдля проверки значения NULL и вызова UDF в условном блоке.
spark.udf.register("strlen_nullsafe", (s: String) => if (s != null) s.length else -1)
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
API типизированного набора данных
Примечание.
Эта функция поддерживается кластерами с включенным каталогом Unity и стандартным режимом доступа в Databricks Runtime 15.4 и выше.
Используйте типизированные API набора данных для выполнения преобразований, таких как сопоставление, фильтрация и агрегирование наборов данных с определяемой пользователем функцией.
В следующем примере API используется map() для изменения числа в столбце результатов до префиксной строки:
spark.range(3).map(f => s"row-$f").show()
В этом примере используетсяmap(), но тот же шаблон применяется к другим типизированным API набора данных, таким как filter(), , mapPartitions(), foreach()foreachPartition()reduce()и .flatMap()
Совместимость функций Scala UDF и Databricks Runtime
Для следующих функций требуются минимальные версии среды выполнения Databricks в кластерах с поддержкой каталога Unity в стандартном (общем) режиме доступа.
| Функция | Минимальная версия среды выполнения Databricks |
|---|---|
| Определяемые пользователем скалярные функции | Databricks Runtime версия 14.2 |
Dataset.map, , Dataset.mapPartitionsDataset.filter, Dataset.reduceDataset.flatMap |
Databricks Runtime 15.4 |
KeyValueGroupedDataset.flatMapGroups, KeyValueGroupedDataset.mapGroups |
Databricks Runtime 15.4 |
(Потоковая передача) foreachWriter Sink |
Databricks Runtime 15.4 |
(Потоковая передача) foreachBatch |
Databricks Runtime 16.1 |
(Потоковая передача) KeyValueGroupedDataset.flatMapGroupsWithState |
Databricks Runtime 16.2 |
spark.udf.registerJavaFunction(Java UDF из JAR-файла) |
Databricks Runtime 18 LTS |