Пользовательские функции Scala и Java уровня сеанса

На этой странице описано, как создавать пользовательские функции Scala и Java уровня сеанса в Azure Databricks. Пользовательские функции (UDF) с областью действия в рамках сеанса определяются в записной книжке или задании и действуют только в пределах текущего SparkSession. Справочник по языку SQL см. в статье Внешние скалярные пользовательские функции (UDF).

Выбор подхода

Вы можете определить UDF на Scala или Java следующими способами. Чтобы сравнить все типы UDF в разных языках, режимах управления и вычислительных средах, см. UDF под управлением Unity Catalog и UDF с областью действия сеанса.

Требования

• 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.3 или более поздней версии. См. Регистрация Java UDF из JAR-файла.

Зарегистрируйте функцию как 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"))

Регистрация Java UDF из JAR-файла

Упакуйте UDF в JAR-файл, добавьте его в текущий сеанс с помощью spark.addArtifact и зарегистрируйте класс UDF с помощью spark.udf.registerJavaFunction.

Ниже описано, как создать проект, написать класс 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

Fat 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-файл в том с помощью обозревателя каталогов:

1. В рабочей области Azure Databricks щелкните Catalog, чтобы открыть обозреватель каталогов.
2. Выберите каталог, а затем выберите схему, содержащую том.
3. Щелкните имя тома.
4. Нажмите кнопку "Отправить в этот том " и выберите JAR-файл.
5. Нажмите кнопку Отправить.
6. После завершения отправки щелкните имя 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 типизированного набора данных

Используйте типизированные 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 в стандартном (общем) режиме доступа.