Pular para o conteúdo principal
Última atualização em

Funções definidas pelo usuário (UDFs) em Python em lote no Unity Catalog

As UDFs Python do Unity Catalog em lotes estão geralmente disponíveis. Elas operam em lotes de dados em vez de uma linha por vez.

Requisitos​

No compute clássico, as UDFs Python do Unity Catalog em lotes exigem o Databricks Runtime 16.3 ou superior. Eles também são suportados no compute serverless e em SQL warehouses Pro e serverless.

Additional capabilities have their own compute and version requirements. See Python UDF requisitos de recurso.

Criar uma UDF Python do Unity Catalog em lote​

A criação de um lote Unity Catalog Python UDF é semelhante à criação de um lote normal Unity Catalog UDF, com as seguintes adições:

  • PARAMETER STYLE PANDAS: Isso especifica que o site UDF processa dados em lotes usando iteradores Pandas.
  • HANDLER 'handler_function': Isso especifica a função de tratamento que processa os lotes.

O exemplo a seguir cria um Python UDF persistente em lotes no Unity Catalog. Substitua my_catalog e my_schema pelo seu catálogo e esquema:

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

Após registrar o UDF, o senhor pode chamá-lo usando SQL ou Python:

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

lotes UDF handler function​

As UDFs Python do Unity Catalog em lotes exigem uma função de manipulador que processe lotes e produza resultados. Você deve especificar o nome da função de manipulador ao criar a UDF usando a cláusula HANDLER.

A função de manipulador faz o seguinte:

  1. Aceita um argumento de iterador que itera sobre um ou mais pandas.Series. Cada pandas.Series contém os parâmetros de entrada da UDF.
  2. Itera sobre o gerador e processa os dados.
  3. Retorna um iterador gerador.

lotes Unity Catalog Python Os UDFs devem retornar o mesmo número de linhas que a entrada. A função handler garante isso produzindo um pandas.Series com o mesmo comprimento da série de entrada para cada lote.

Instale dependências personalizadas​

O senhor pode estender a funcionalidade dos lotes Unity Catalog Python UDFs para além do ambiente Databricks Runtime, definindo dependências personalizadas para bibliotecas externas.

Consulte Estender UDFs usando dependências personalizadas.

Access Unity Catalog secrets​

As UDFs Python do Unity Catalog em lotes podem acessar segredos declarados na cláusula SECRETS. Você deve definir explicitamente environment_version para 6 ou acima. Não há suporte para a invocação direta de uma UDF que usa esta cláusula no compute de modo de acesso dedicado. Para a exceção de máscara de coluna, consulte Use secret-enabled UDFs in column masks on dedicated compute.

lotes UDFs podem aceitar parâmetros únicos ou múltiplos​

Single parameter: When the handler function uses a single input parameter, it receives an iterator over a pandas.Series for each lotes.

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

Vários parâmetros: para vários parâmetros de entrada, a função manipuladora recebe um iterador que itera em vários pandas.Series. Os valores na série estão na mesma ordem dos parâmetros de entrada.

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

Otimizar o desempenho separando operações caras​

O senhor pode otimizar as operações computacionalmente caras separando-as da função de tratamento. Isso garante que eles sejam executados apenas uma vez, e não durante cada iteração em lotes de dados.

O exemplo a seguir mostra como garantir que uma computação cara seja executada somente uma vez:

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

Isolamento ambiental​

nota

Ambientes de isolamento compartilhado exigem Databricks Runtime 17.1 ou superior. Em versões anteriores, todos os lotes Unity Catalog Python UDFs eram executados em modo de isolamento estrito.

Por default, as UDFs (Funções Definidas pelo Usuário) Unity Catalog Python com o mesmo proprietário e sessão podem compartilhar um ambiente de isolamento. Isso pode melhorar o desempenho e reduzir o uso de memória, diminuindo o número de ambientes separados que precisam ser iniciados.

Isolamento estrito​

Para garantir que um e UDF eja sempre em seu próprio ambiente, totalmente isolado, adicione a cláusula de característica STRICT ISOLATION.

A maioria dos UDFs não precisa de isolamento estrito. As UDFs de processamento de dados padrão beneficiam do ambiente de isolamento compartilhado do default e são executadas mais rapidamente com menor consumo de memória.

Adicione a cláusula característica STRICT ISOLATION às UDFs que:

  • execução de entrada como código utilizando eval(), exec() ou funções semelhantes
  • Gravar arquivos no sistema de arquivos local
  • Modificar variáveis globais ou estado do sistema
  • Modificar variável de ambiente

O exemplo a seguir mostra uma UDF que executa a entrada como código e requer isolamento estrito:

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

credenciais de serviço em lotes Unity Catalog Python UDFs​

As UDFs Python do Unity Catalog em lotes podem usar as credenciais de serviço do Unity Catalog para acessar serviços de cloud externos. Isso é particularmente útil para integrar funções de cloud como tokenizadores de segurança em fluxos de trabalho de processamento de dados.

nota

API específica de UDF para credenciais de serviço:
Em UDFs, use databricks.service_credentials.getServiceCredentialsProvider() para acessar credenciais de serviço.

Isso difere da função dbutils.credentials.getServiceCredentialsProvider() usada no Notebook, que não está disponível em contextos de execução UDF .

Para criar uma credencial de serviço, consulte Criar credenciais de serviço.

Especifique a credencial de serviço que o senhor deseja usar na cláusula CREDENTIALS na definição do UDF:

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

Permissões de credenciais de serviço​

Para requisitos de permissão de criação e do chamador entre tipos de compute, consulte Use a service credential in a Python UDF.

credenciais e aliases padrão​

Você pode incluir várias credenciais na cláusula CREDENTIALS, mas somente uma pode ser marcada como DEFAULT. O senhor pode criar um alias para credenciais que não sejamdefault usando a palavra-chave AS. Cada credencial deve ter um alias exclusivo.

Os SDKs de nuvem corrigidos captam automaticamente as credenciais do default. A default credencial tem precedência sobre qualquer default especificada na compute Spark configuração do e persiste na Unity Catalog UDF definição .

Exemplo de credencial de serviço - Google Cloud Storage​

O exemplo a seguir usa uma credencial de serviço para acessar Google Cloud Storage a partir de um Unity Catalog Python UDF:

Python
%sql
CREATE OR REPLACE FUNCTION main.test.read_gcs_blob(blob_name STRING) RETURNS STRING LANGUAGE PYTHON
PARAMETER STYLE PANDAS
HANDLER 'batchhandler'
CREDENTIALS (
  `batch-udf-service-creds-example-cred` DEFAULT
)
ENVIRONMENT (
  dependencies = '["google-auth", "google-cloud-storage"]', environment_version = 'None'
)
AS $$
import google.auth  # This import is required to enable SDK credential integration
import pandas as pd
from google.cloud import storage


def batchhandler(it):
  # The client automatically uses the DEFAULT service credential
  client = storage.Client(project="your-project")
  bucket = client.bucket("your-bucket")

  for blob_names in it:
    results = []
    for name in blob_names:
      blob = bucket.blob(name)
      try:
        content = blob.download_as_text()
        results.append(content)
      except Exception as e:
        results.append(f"Error: {e}")
    yield pd.Series(results)
$$;

Chame o UDF depois que ele for registrado:

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

Obter o contexto de execução da tarefa​

Use o TaskContext PySpark API para obter informações de contexto, como a identidade do usuário, a tag do cluster, o ID do spark job e muito mais. Consulte Obter contexto de tarefa em um UDF.

Defina DETERMINISTIC se sua função produzir resultados consistentes​

Adicione DETERMINISTIC à sua definição de função se ela produzir as mesmas saídas para as mesmas entradas. Isso permite otimizações de consulta para melhorar o desempenho.

Por default, as UDTFs Python do Unity Catalog em lotes são consideradas não determinísticas, a menos que sejam declaradas explicitamente. Exemplos de funções não determinísticas incluem: gerar valores aleatórios, acessar horários ou datas atuais ou fazer chamadas de API externas.

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

Limitações​

  • Python devem tratar os valores de NULL de forma independente, e todos os mapeamentos de tipos devem seguir os mapeamentos de linguagem de Databricks SQL.
  • Os lotes Unity Catalog Python UDFs são executados em um ambiente seguro e isolado e não têm acesso a um sistema de arquivos compartilhado ou a um serviço interno.
  • Várias invocações do UDF em um estágio são serializadas e os resultados intermediários são materializados e podem ser transferidos para o disco.
Nesta página