Пакетные пользовательские функции Python (UDF) в Unity Catalog

Это важно

Эта функция доступна в общедоступной предварительной версии.

Определяемые пользователем функции (UDF) каталога Unity для пакетной обработки с помощью Python расширяют возможности стандартных UDF каталога Unity, позволяя писать код на Python для обработки пакетов данных, что значительно повышает эффективность за счет снижения накладных расходов, связанных с обработкой данных по одной строке. Эти оптимизации делают UDF на Python в Unity Catalog оптимальными для обработки данных в крупном масштабе.

Требования

Для пакетных пользовательских функций Unity Catalog на Python требуется Databricks Runtime версии 16.3 и выше.

Создание Python UDF для каталога пакетной обработки Unity

Создание пользовательской функции (UDF) пакетного каталога Unity аналогично созданию обычной пользовательской функции каталога Unity со следующими дополнениями:

  • PARAMETER STYLE PANDAS: это указывает, что UDF обрабатывает данные в пакетах с помощью итераторов pandas.
  • HANDLER 'handler_function': это указывает функцию обработчика, которая вызывается для обработки пакетов.

В следующем примере показано, как создать Python UDF для каталога Unity для пакетных заданий.

%sql
CREATE OR REPLACE TEMPORARY FUNCTION 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)
$$;

После регистрации UDF его можно вызвать с помощью SQL или Python:

SELECT person_id, 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
);

Функция обработчика UDF для пакетной обработки

Для Python UDF каталога Unity требуется функция обработчика, которая обрабатывает пакеты и возвращает результаты. Необходимо указать имя функции обработчика при создании UDF, используя ключ HANDLER.

Функция обработчика выполняет следующие действия:

  1. Принимает аргумент итератора, который выполняет итерацию по одному или нескольким pandas.Series. Каждый pandas.Series содержит входные параметры UDF.
  2. Выполняет итерацию по генератору и обрабатывает данные.
  3. Возвращает генераторный итератор.

ППФ каталога Batch Unity должны возвращать то же количество строк, что и вход. Эта функция обработчика обеспечивает это путем получения pandas.Series той же длины, что и входной ряд для каждого пакета.

Установка пользовательских зависимостей

Вы можете расширить функциональные возможности пользовательских Python UDF в Unity Catalog для пакетной обработки за пределами среды выполнения Databricks, определив пользовательские зависимости для внешних библиотек.

См. Расширение UDF с использованием настраиваемых зависимостей.

Пакетные функции, определяемые пользователем, могут принимать один или несколько параметров.

Один параметр: Когда функция обработчика использует один входной параметр, он получает итератор, который выполняет итерацию по каждому пакету pandas.Series .

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

Несколько параметров: Для нескольких входных параметров функция обработчика получает итератор, который выполняет итерацию по нескольким pandas.Series. Значения в серии находятся в том же порядке, что и входные параметры.

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

Оптимизация производительности путем разделения дорогостоящих операций

Вы можете оптимизировать вычислительные дорогостоящие операции, разделив эти операции от функции обработчика. Это гарантирует, что они выполняются только один раз, а не во время каждой итерации по пакетам данных.

В следующем примере показано, как обеспечить выполнение дорогостоящих вычислений только один раз:

%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

Изоляция среды

Замечание

Для общих изолированных сред требуется Databricks Runtime 17.1 и более поздних версий. В более ранних версиях все Python UDF Каталога Unity для пакетных заданий выполняются в строгом режиме изоляции.

Пакетные UDF Python в Unity Catalog с одинаковым владельцем и сеансом могут по умолчанию совместно использовать одну и ту же среду изоляции. Это может повысить производительность и уменьшить использование памяти, уменьшая количество отдельных сред, которые необходимо запустить.

Строгая изоляция

Чтобы убедиться, что UDF всегда выполняется в собственной, полностью изолированной среде, добавьте STRICT ISOLATION предложение характеристик.

Большинству определяемых пользователем функций не требуется строгая изоляция. Стандартные пользовательские функции обработки данных используют общую изолированную среду по умолчанию и выполняются быстрее с меньшим потреблением памяти.

Добавьте STRICT ISOLATION характерное условие к определяемым пользователем функциям, которые:

  • Выполнение входных данных в виде кода с помощью eval()exec()или аналогичных функций
  • Запись файлов в локальную файловую систему
  • Изменение глобальных переменных или состояния системы
  • Изменение переменных среды

В следующем примере показан UDF, который выполняет входные данные в виде кода и требует строгой изоляции:

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

Учетные данные службы в Python UDFs Единого каталога Unity для пакетной обработки

Пакетные пользовательские функции Python каталога Unity Catalog могут использовать учетные данные службы Unity Catalog для доступа к внешним облачным сервисам. Это особенно полезно для интеграции облачных функций, таких как маркеризаторы безопасности, в рабочие процессы обработки данных.

Замечание

API, специфичный для UDF, для учетных данных службы:
В пользовательских функциях используйте databricks.service_credentials.getServiceCredentialsProvider() для доступа к учетным данным службы.

Это отличается от функции, используемой dbutils.credentials.getServiceCredentialsProvider() в записных книжках, которая недоступна в контекстах выполнения UDF.

Сведения о создании учетных данных службы см. в разделе "Создание учетных данных службы".

Укажите учетные данные службы, которые вы хотите использовать в CREDENTIALS части определения UDF:

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

Разрешения учетных данных службы

Создатель UDF должен иметь ACCESS разрешение на учетные данные службы каталога Unity. Однако для пользователей, вызывающих UDF, достаточно предоставить им EXECUTE разрешение на UDF. В частности, вызывающие UDF не нуждаются в доступе к учетным данным базовой службы, так как UDF выполняется с помощью разрешений учетных данных создателя UDF.

Для временных функций создатель всегда является вызывающим элементом. Определяемые пользователем функции, которые выполняются в области No-PE, также известной как выделенные кластеры, используют разрешения вызывающего пользователя.

Учетные данные и псевдонимы по умолчанию

В условие можно включить несколько CREDENTIALS идентификаторов, но только один может быть помечен как DEFAULT. Вы можете псевдонимить учетные данные, отличные от по умолчанию, с помощью ключевого AS слова. У каждой учетной записи должен быть уникальный псевдоним.

Исправленные облачные SDK автоматически используют учетные данные по умолчанию. Учетные данные по умолчанию имеют приоритет над любой конфигурацией по умолчанию, указанной в настройках вычислений Spark, и сохраняются в определении UDF Каталога Unity.

Для использования поставщика azure-identity необходимо установить пакет DefaultAzureCredential. Используйте директиву ENVIRONMENT, чтобы установить внешние библиотеки. Дополнительные сведения об установке внешних библиотек см. в разделе «Расширение UDF с использованием пользовательских зависимостей».

Пример учетных данных службы — хранилище BLOB-объектов Azure

В следующем примере используются учетные данные службы для доступа к Хранилище BLOB-объектов Azure из UDF каталога Unity для пакетных заданий.

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

Вызовите UDF после его регистрации:

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

Получить контекст выполнения задачи

Используйте TaskContext PySpark API, чтобы получить информацию о контексте, такую как идентификация пользователя, метки кластера, идентификатор задания Spark и многое другое. См. Получение контекста задачи в UDF.

Установить DETERMINISTIC если функция производит последовательные результаты

Добавьте DETERMINISTIC в определение функции, если он создает те же выходные данные для одних и того же входных данных. Это позволяет оптимизировать запросы для повышения производительности.

По умолчанию определяемые пользователем UDTF каталога Batch Unity считаются недетерминированными, если явно не объявлены. Примеры недетерминированных функций: создание случайных значений, доступ к текущим времени или датам или вызовы внешних API.

См. CREATE FUNCTION (SQL, Python, Scala и Java)

Ограничения

  • Функции Python должны обрабатывать NULL значения независимо, и все сопоставления типов должны соответствовать сопоставлениям языка SQL Azure Databricks.
  • Пакетные UDF Python для каталога Unity выполняются в безопасной, изолированной среде и не имеют доступа к общей файловой системе или внутренним службам.
  • Несколько вызовов UDF на стадии сериализуются, а промежуточные результаты материализуются и могут записываться на диск.
  • Учетные данные службы доступны только в определенных пользователем функциях (UDF) для Batch Unity Catalog и скалярных функциях Python UDF. Они не поддерживаются в стандартных Python UDF каталога Unity.
  • В выделенных кластерах и для временных функций вызывающий функцию должен иметь ACCESS разрешения на учетные данные службы. См . раздел "Предоставление разрешений на использование учетных данных службы для доступа к внешней облачной службе".
  • Включение функции общедоступной предварительной версии позволяет включить сеть для изолированных рабочих нагрузок в бессерверных хранилищах SQL на странице предварительной версии рабочей области, чтобы сделать вызовы UDF каталога Пакетной службы Unity для внешних служб на бессерверных вычислительных ресурсах хранилища SQL.
  • Чтобы выполнять вызовы UDF каталога Unity Batch на бессерверной записной книжке или в вычислительной среде заданий, необходимо настроить управление бессерверным исходящим трафиком.
  • Last updated on