UDFs (funções definidas pelo usuário) do Python em lote no Catálogo do Unity

As UDFs em Python no Catálogo do Unity em lote estendem as funcionalidades das UDFs do Catálogo do Unity, permitindo que você escreva um código Python para operar em lotes de dados, aprimorando significativamente a eficiência ao reduzir a sobrecarga associada a UDFs de linha por linha. Essas otimizações tornam as UDFs do Python em lote do Unity Catalog ideais para processamento de dados em grande escala.

Requisitos

As UDFs em Python no Catálogo do Unity em lote exigem versões do Databricks Runtime 16.3 e superiores.

Criar UDF em Python no Catálogo do Unity em lote

A criação de um UDF do Python do Catálogo do Unity em lote é semelhante à criação de um UDF de Catálogo do Unity regular, com as seguintes adições:

• PARAMETER STYLE PANDAS: especifica que a UDF processa dados em lotes usando iteradores pandas.
• HANDLER 'handler_function': especifica a função de manipulador que é chamada para processar os lotes.

O exemplo a seguir mostra como criar um UDF Python do Catálogo do Unity em lote:

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

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

Função de manipulador de UDF em lote

UDFs em Python do Catálogo do Unity em lote exigem uma função de manipulador que processa lotes e gera resultados. Você deve especificar o nome da função de manipulador ao criar a UDF usando a HANDLER chave.

A função de manipulador faz o seguinte:

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

UDFs em Python no Catálogo do Unity em lote devem retornar o mesmo número de linhas que a entrada. A função de manipulador garante isso gerando um pandas.Series com o mesmo comprimento que a série de entrada para cada lote.

Instalar dependências personalizadas

Você pode estender a funcionalidade de UDFs em Python no Catálogo do Unity para além do ambiente do Databricks Runtime definindo dependências personalizadas para bibliotecas externas.

Consulte Estender UDFs usando dependências personalizadas.

UDFs em Lote podem aceitar um único parâmetro ou múltiplos parâmetros

Parâmetro único: Quando a função manipuladora usa um único parâmetro de entrada, ela recebe um iterador que percorre um 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 de manipulador recebe um iterador que itera em 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);

Otimizar o desempenho separando operações caras

Você pode otimizar operações computacionalmente caras separando essas operações da função de 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 uma computação cara seja executada 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 de ambiente

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

Isolamento estrito

Para garantir que uma UDF sempre seja executada em seu próprio ambiente totalmente isolado, adicione a cláusula de 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 nas UDFs que:

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

O exemplo a seguir mostra uma 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 em UDFs do Python do Catálogo do Unity do Lote

Os UDFs em Python do Catálogo do Unity em lote podem usar as Credenciais do Serviço de Catálogo do Unity para acessar serviços de nuvem externos. Isso é particularmente útil para integrar funções de nuvem, como tokens 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 da 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

O criador da UDF deve ter a permissão ACCESS na credencial de serviço do Catálogo do Unity. No entanto, para os chamadores da UDF, é suficiente conceder-lhes a permissão EXECUTE na UDF. Os chamadores da UDF, em particular, não precisam de acesso à credencial de serviço subjacente, pois a UDF é executada usando as permissões de credencial do criador da UDF.

Para funções temporárias, o criador é sempre o invocador. As UDFs executadas no escopo No-PE, também conhecidos como clusters dedicados, usam as permissões do chamador.

Credenciais e aliases padrão

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

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

Você deve instalar o azure-identity pacote para usar o DefaultAzureCredential provedor. Use a ENVIRONMENT cláusula 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 a seguir usa uma credencial de serviço para acessar o Armazenamento de Blobs do Azure de um UDF Python do Unity Catalog do Batch:

%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 que ela for registrada:

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

Obter o contexto de execução da tarefa

Use a API do TaskContext PySpark para obter informações de contexto, como identidade do usuário, marcas de cluster, ID do trabalho do Spark e muito mais. Veja Como obter o contexto de tarefa em uma UDF.

Definir DETERMINISTIC se 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, as UDTFs do Python do Catálogo do Unity do Lote são consideradas não deterministioc, a menos que explicitamente declaradas. Exemplos de funções não determinísticas incluem: gerar valores aleatórios, acessar datas ou horários atuais ou fazer chamadas à API externa.

Consulte CREATE FUNCTION (SQL e Python)

Limitações

• As funções do Python devem lidar com NULL valores de forma independente e todos os mapeamentos de tipo devem seguir mapeamentos de linguagem SQL do Azure Databricks.
• As UDFs em Python do Catálogo do Unity em lote 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.
• Várias invocações de UDF em um estágio são serializadas e os resultados intermediários são materializados e podem ser despejados em disco.
• As credenciais de serviço estão disponíveis apenas em UDFs Python do Catálogo Unity Batch e UDFs Python escalares. Eles não são suportados nas UDFs padrão do Python no Unity Catalog.
• Em clusters dedicados e para funções temporárias, o chamador de função deve ter permissões ACCESS nas credenciais de serviço. Veja Conceder permissões para usar uma credencial de serviço para acessar um serviço de nuvem externo.
• Habilite o recurso visualização pública habilitar a rede para UDFs em SQL Warehouses sem servidor na página Visualizações do workspace para fazer chamadas de UDF do Python do Catálogo do Lote Unity para serviços externos na computação do SQL Warehouse sem servidor.

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