Funciones definidas por el usuario (UDF) de Python por lotes en Unity Catalog

Los UDF de Python por lotes de Unity Catalog ya están disponibles de forma general. Operan sobre lotes de datos en lugar de una fila a la vez.

Requisitos

En cómputo clásico, los UDF de Python por lotes de Unity Catalog requieren Databricks Runtime 16.3 o superior. También son compatibles con el cómputo sin servidor y con los almacenes SQL Pro y sin servidor.

Las capacidades adicionales tienen sus propios requisitos de cálculo y versión. Consulta los requisitos de la función UDF de Python.

Crear un catálogo Batch de Unity en Python UDF

La creación de una UDF de Python del catálogo de Batch unity es similar a la creación de una UDF normal del catálogo de Unity, con las siguientes adiciones:

  • PARAMETER STYLE PANDAS: especifica que la UDF procesa los datos en lotes mediante iteradores pandas.
  • HANDLER 'handler_function': Esto especifica la función manejadora que procesa los lotes.

El siguiente ejemplo crea un UDF persistente de Python por lotes en el Catálogo de Unity. Sustituye my_catalog y my_schema con tu catálogo y esquema:

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

Después de registrar la UDF, puede llamarla mediante SQL o Python:

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

Función de controlador de UDF por lotes

Las UDF de Python del catálogo de Unity por lotes requieren una función de controlador que procesa lotes y genera resultados. Debes especificar el nombre de la función manejadora cuando crees la UDF usando la HANDLER cláusula.

La función de controlador hace lo siguiente:

  1. Acepta un argumento iterador que itera sobre uno o más pandas.Series. Cada pandas.Series contiene los parámetros de entrada de la UDF.
  2. Recorre en iteración los datos del generador y los procesa.
  3. Devuelve un iterador generador.

Las UDF de Python del catálogo de Unity por lotes deben devolver el mismo número de filas que la entrada. La función de controlador garantiza esto produciendo un pandas.Series con la misma longitud que la serie de entrada para cada lote.

Instalación de dependencias personalizadas

Puede ampliar la funcionalidad de las UDF de Python del catálogo de Unity Batch más allá del entorno de runtime de Databricks mediante la definición de dependencias personalizadas para bibliotecas externas.

Consulte Extensión de UDF mediante dependencias personalizadas.

Acceder a los secretos de Unity Catalog

Las UDF de Python por lotes de Unity Catalog pueden acceder a los secretos declarados en la cláusula SECRETS. Debes poner environment_version explícitamente como 6 o por encima. La invocación directa de un UDF que utiliza esta cláusula no es compatible con el cálculo en modo de acceso dedicado. Para la excepción de la máscara de columna, consulte Uso de UDF con secretos habilitados en máscaras de columna en proceso dedicado.

Las UDF de Batch pueden aceptar parámetros únicos o varios

Parámetro único: Cuando la función manejadora utiliza un único parámetro de entrada, recibe un iterador sobre a pandas.Series para cada lote.

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

Varios parámetros: Para varios parámetros de entrada, la función de controlador recibe un iterador que recorre en iteración varios pandas.Series. Los valores de la serie están en el mismo orden que los parámetros de entrada.

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

Optimización del rendimiento mediante la separación de operaciones costosas

Puede optimizar las operaciones costosas computacionalmente separando estas operaciones de la función del controlador. Esto garantiza que se ejecuten solo una vez en lugar de durante cada iteración sobre lotes de datos.

En el ejemplo siguiente se muestra cómo asegurarse de que un cálculo costoso se realiza solo una vez:

%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

Aislamiento del entorno

Nota:

Los entornos de aislamiento compartido requieren Databricks Runtime 17.1 y versiones posteriores. En versiones anteriores, todas las UDF de Python del catálogo de Batch Unity se ejecutan en modo de aislamiento estricto.

Las UDF de Python del catálogo de Batch unity con el mismo propietario y sesión pueden compartir un entorno de aislamiento de forma predeterminada. Esto puede mejorar el rendimiento y reducir el uso de memoria al reducir el número de entornos independientes que deben iniciarse.

Aislamiento estricto

Para asegurarse de que una UDF siempre se ejecute en su propio entorno totalmente aislado, agregue la STRICT ISOLATION cláusula característica.

La mayoría de las UDF no necesitan aislamiento estricto. Las UDF de procesamiento de datos estándar se benefician del entorno de aislamiento compartido predeterminado y se ejecutan más rápido con un menor consumo de memoria.

Agregue la cláusula característica STRICT ISOLATION a las UDFs que:

  • Ejecutar la entrada como código mediante eval(), exec()o funciones similares
  • Escribir archivos en el sistema de archivos local
  • Modificación de variables globales o estado del sistema
  • Modificación de variables de entorno

En el ejemplo siguiente se muestra una UDF que ejecuta la entrada como código y requiere un aislamiento estricto:

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

Credenciales de servicio en las UDF de Python del catálogo de Unity de Batch

Los UDF de Python del Catálogo de Unity por lotes pueden usar las credenciales del servicio del Catálogo de Unity para acceder a servicios en la nube externos. Esto es especialmente útil para integrar funciones en la nube como tokenizadores de seguridad en flujos de trabajo de procesamiento de datos.

Nota:

API específica de UDF para credenciales de servicio:
En UDF, use databricks.service_credentials.getServiceCredentialsProvider() para acceder a las credenciales de servicio.

Esto difiere de la dbutils.credentials.getServiceCredentialsProvider() función usada en cuadernos, que no está disponible en contextos de ejecución de UDF.

Para crear una credencial de servicio, consulte Creación de credenciales de servicio.

Especifique la credencial de servicio que desea usar en la CREDENTIALS cláusula en la definición de 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
$$;

Permisos de credenciales de servicio

Para requisitos de creación y permisos para llamadas entre tipos de cálculo, véase Usar una credencial de servicio en un UDF de Python.

Credenciales y alias predeterminados

Puede incluir varias credenciales en la cláusula CREDENTIALS, pero solo una se puede marcar como DEFAULT. Puede usar alias para las credenciales no predeterminadas mediante la palabra clave AS. Cada credencial debe tener un alias único.

Los SDK de nube revisados seleccionan automáticamente las credenciales predeterminadas. La credencial predeterminada tiene prioridad sobre cualquier valor predeterminado especificado en la configuración de Spark del proceso y persiste en la definición de UDF del catálogo de Unity.

Debe instalar el azure-identity paquete para usar el DefaultAzureCredential proveedor. Use la ENVIRONMENT cláusula para instalar bibliotecas externas. Para obtener más información sobre la instalación de bibliotecas externas, consulte Extensión de las UDF mediante dependencias personalizadas.

Ejemplo de credenciales de servicio: Azure Blob Storage

En el ejemplo siguiente se usa una credencial de servicio para acceder a Azure Blob Storage desde una UDF de Python del catálogo de Batch 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)
$$;

Llame a la UDF después de registrarla:

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

Obtención del contexto de ejecución de tareas

Use taskContext PySpark API para obtener información de contexto, como la identidad del usuario, las etiquetas de clúster, el identificador de trabajo de Spark, etc. Consulte Obtener contexto de tarea en una UDF.

Establecer DETERMINISTIC si la función genera resultados coherentes

Agregue DETERMINISTIC a la definición de función si genera las mismas salidas para las mismas entradas. Esto permite que las optimizaciones de consultas mejoren el rendimiento.

De forma predeterminada, se supone que las UDF de Python del catálogo de Batch Unity no son deterministas a menos que se declaren explícitamente. Entre los ejemplos de funciones no deterministas se incluyen: generar valores aleatorios, acceder a las horas o fechas actuales o realizar llamadas API externas.

Consulte CREATE FUNCTION (SQL, Python, Scala y Java)

Limitaciones

  • Las funciones de Python deben controlar NULL los valores de forma independiente y todas las asignaciones de tipos deben seguir las asignaciones de lenguaje SQL de Azure Databricks.
  • Las UDF de Python del catálogo de Batch unity se ejecutan en un entorno seguro y aislado y no tienen acceso a un sistema de archivos compartido ni a servicios internos.
  • Se serializan varias invocaciones de UDF dentro de una fase y los resultados intermedios se materializan y pueden desbordarse en el disco.
  • Para ejecutar llamadas UDF de Python en el Batch Unity Catalog en un cuaderno o cómputo de tareas sin servidor, debe configurar el control de salida sin servidor.
  • Last updated on