Funções definidas pelo utilizador (UDFs) em Python em lote no Catálogo Unity

Os UDFs Python do Catálogo Batch Unity estão geralmente disponíveis. Operam em lotes de dados em vez de uma linha de cada vez.

Requerimentos

No compute clássico, as UDFs Python em lote do Unity Catalog requerem o Databricks Runtime 16.3 ou posterior. Também são suportados na computação serverless e nos data warehouses SQL pro e serverless.

As capacidades adicionais têm os seus próprios requisitos de computação e versão. Consulte os requisitos de funcionalidades do Python UDF.

Criar uma UDF Python em lote no Unity Catalog

Criar um Batch Unity Catalog Python UDF é semelhante à criação de um Unity Catalog UDF regular, com as seguintes adições:

• PARAMETER STYLE PANDAS: Isso especifica que o UDF processa dados em lotes usando iteradores pandas.
• HANDLER 'handler_function': Isto especifica a função handler que processa os lotes.

O exemplo seguinte cria um UDF Python Batch persistente no Unity Catalog. Substitua my_catalog e my_schema pelo seu catálogo e pelo seu 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)
$$;

Depois de registrar o UDF, você pode chamá-lo usando SQL ou 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
);

Função do manipulador UDF em lote

UDFs Python do Batch Unity Catalog requerem uma função de manipulador que processa lotes e produz resultados. Deve especificar o nome da função de processamento ao criar a UDF, utilizando a cláusula HANDLER.

A função handler faz o seguinte:

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

UDFs Python do Batch Unity Catalog 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.

Instalar dependências personalizadas

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

Consulte Estender UDFs usando dependências personalizadas.

Aceder aos segredos do Unity Catalog

Os UDFs Python do Batch Unity Catalog podem aceder a segredos declarados na cláusula SECRETS. Deve definir environment_version explicitamente como 6 ou acima. A invocação direta de um UDF que utiliza esta cláusula não é suportada em computação em modo de acesso dedicado. Para a exceção de máscaras de coluna, veja Utilizar UDFs com segredos ativados em máscaras de coluna na computação dedicada.

UDFs de lote podem aceitar parâmetros únicos ou múltiplos

Parâmetro único: Quando a função handler usa um único parâmetro de entrada, recebe um 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;

Vários parâmetros: Para vários parâmetros de entrada, a função handler recebe um iterador que itera sobre vários pandas.Series. Os valores na série estão na mesma ordem que os 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);

Otimize o desempenho separando operações dispendiosas

Você pode otimizar operações computacionalmente caras separando essas operações da função manipulador. Isso garante que eles sejam executados apenas uma vez, em vez de durante cada iteração em lotes de dados.

O exemplo a seguir mostra como garantir que um cálculo caro seja executado apenas uma 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

Isolamento do ambiente

Os Python UDFs do Unity Catalog no modo Batch com o mesmo proprietário e sessão podem, por defeito, partilhar um ambiente de isolamento. Isso pode melhorar o desempenho e reduzir o uso de memória, reduzindo o número de ambientes separados que precisam ser iniciados.

Isolamento rigoroso

Para garantir que uma UDF seja sempre executada no seu próprio ambiente totalmente isolado, adicione a cláusula característica STRICT ISOLATION.

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

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

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

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

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 no Batch Unity Catalog Python UDFs

As UDFs de Python do Unity Catalog em lote podem usar credenciais de serviço do Unity Catalog para aceder a serviços externos na nuvem. Isso é particularmente útil para integrar funções de nuvem, como tokenizadores de segurança, em fluxos de trabalho de processamento de dados.

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

Especifique a credencial de serviço que você deseja usar na CREDENTIALS cláusula na definição 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
$$;

Permissões de credenciais de serviço

Para requisitos de criação e permissões de chamador entre tipos de computação, consulte Usar uma credencial de serviço num UDF Python.

Credenciais e aliases padrão

Você pode incluir várias credenciais na cláusula CREDENTIALS, mas apenas uma pode ser marcada como DEFAULT. Você pode aliar credenciais não padrão usando a palavra-chave AS. Cada credencial deve ter um alias exclusivo.

SDKs de nuvem corrigidos selecionam automaticamente as credenciais padrão. A credencial padrão tem precedência sobre qualquer padrão especificado na configuração do Spark da computação e persiste na definição UDF do Unity Catalog.

Você deve instalar o azure-identity pacote para usar o DefaultAzureCredential provedor. Utilize a cláusula ENVIRONMENT para instalar bibliotecas externas. Para saber mais sobre como instalar bibliotecas externas, consulte Estender UDFs usando dependências personalizadas.

Exemplo de credencial de serviço - Armazenamento de Blobs do Azure

O exemplo seguinte utiliza uma credencial de serviço para aceder ao Armazenamento de Blobs do Azure através de um Python UDF do Batch Unity Catalog:

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

Chame a UDF depois de ser registada:

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

Obter contexto de execução da tarefa

Utilize a API PySpark do TaskContext para obter informações de contexto, como identidade do utilizador, tags do cluster, ID do trabalho Spark e mais. Veja Obter o contexto da tarefa num UDF.

Definir DETERMINISTIC se a sua função produz 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 padrão, UDTFs Python do Batch Unity Catalog são assumidos como não determinísticos, a menos que explicitamente declarados. Exemplos de funções não determinísticas incluem: gerar valores aleatórios, acessar horas ou datas atuais ou fazer chamadas de API externas.

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

Limitações

• As funções Python devem manipular NULL valores de forma independente e todos os mapeamentos de tipo devem seguir os mapeamentos da linguagem SQL do Azure Databricks.
• As UDFs Python do Batch Unity Catalog são executadas em um ambiente seguro e isolado e não têm acesso a um sistema de arquivos compartilhado ou serviços internos.
• As várias invocações de UDFs dentro de um estágio são serializadas, e resultados intermediários são materializados e podem ser despejados no disco.

• Para fazer chamadas UDF Python do Batch Unity Catalog em um notebook sem servidor ou computação de trabalho, você deve configurar o controle de saída sem servidor