O que são UDFs (funções definidas pelo usuário)?

As UDFs (funções definidas pelo usuário) permitem que você reutilize e compartilhe código que estende a funcionalidade interna no Azure Databricks. Use UDFs para executar tarefas específicas, como cálculos complexos, transformações ou manipulações de dados personalizadas.

Quando você deve usar uma UDF?

Use UDFs para lógica difícil de expressar com funções internas do Apache Spark. As funções internas do Apache Spark são otimizadas para processamento distribuído e geralmente oferecem melhor desempenho em escala. Para obter mais informações, consulte Funções.

A Databricks recomenda UDFs para consultas ad hoc, limpeza manual de dados, análise exploratória de dados e operações em conjuntos de dados de pequeno a médio porte. Casos de uso comuns para UDFs incluem criptografia e descriptografia de dados, hash, análise JSON e validação.

Use métodos do Apache Spark para operações em conjuntos de dados muito grandes e quaisquer cargas de trabalho executadas regular ou continuamente, incluindo trabalhos de ETL e operações de streaming.

UDFs registradas e com escopo de sessão

As UDFs criadas usando SQL são registradas no Catálogo do Unity e têm permissões associadas, enquanto as UDFs criadas em seu notebook são baseadas em sessão e têm como escopo o SparkSession atual.

Você pode definir e acessar UDFs baseadas em sessão usando qualquer linguagem compatível com o Azure Databricks. As UDFs podem ser escalares ou não escalares.

UDFs escalares

As UDFs escalares operam em uma única linha e retornam um único valor para cada linha. O exemplo a seguir usa uma UDF escalar para calcular o comprimento de cada nome em uma name coluna e adicionar o valor em uma nova coluna name_length:

+-------+-------+
| name  | score |
+-------+-------+
| alice |  10.0 |
| bob   |  20.0 |
| carol |  30.0 |
| dave  |  40.0 |
| eve   |  50.0 |
+-------+-------+

-- Create a SQL UDF for name length
CREATE OR REPLACE FUNCTION get_name_length(name STRING)
RETURNS INT
RETURN LENGTH(name);

-- Use the UDF in a SQL query
SELECT name, get_name_length(name) AS name_length
FROM your_table;

+-------+-------+-------------+
| name  | score | name_length |
+-------+-------+-------------+
| alice |  10.0 |      5      |
|  bob  |  20.0 |      3      |
| carol |  30.0 |      5      |
| dave  |  40.0 |      4      |
|  eve  |  50.0 |      3      |
+-------+-------+-------------+

Para implementar isso em um notebook do Databricks usando o PySpark:

from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType

@udf(returnType=IntegerType())
def get_name_length(name):
   return len(name)

df = df.withColumn("name_length", get_name_length(df.name))

# Show the result
display(df)

Para obter mais informações, consulte UDFs (funções definidas pelo usuário) no Catálogo do Unity e Funções escalares definidas pelo usuário – Python.

UDAFs (Funções de agregação definidas pelo usuário)

As UDAFs (funções agregadas definidas pelo usuário) operam em várias linhas e retornam um único resultado agregado. No exemplo a seguir, é definida uma UDAF que agrega pontuações.

from pyspark.sql.functions import pandas_udf
from pyspark.sql import SparkSession
import pandas as pd

# Define a pandas UDF for aggregating scores
@pandas_udf("int")
def total_score_udf(scores: pd.Series) -> int:
    return scores.sum()

# Group by name length and aggregate
result_df = (df.groupBy("name_length")
              .agg(total_score_udf(df["score"]).alias("total_score")))

display(result_df)

+-------------+-------------+
| name_length | total_score |
+-------------+-------------+
|      3      |     70.0    |
|      4      |     40.0    |
|      5      |     40.0    |
+-------------+-------------+

Consulte funções definidas pelo usuário do pandas para Python e funções agregadas definidas pelo usuário - Scala.

UDTFs (funções de tabela definidas pelo usuário) do Python

As funções de tabela definidas pelo usuário (UDTFs) do Python podem retornar várias linhas e colunas para cada linha de entrada. No exemplo a seguir, cada valor na coluna de pontuação corresponde a uma lista de categorias. Uma UDTF é definida para dividir a lista separada por vírgulas em várias linhas. Consulte UDTFs (funções de tabela definidas pelo usuário) do Python

+-------+-------+-----------------+
| name  | score |   categories    |
+-------+-------+-----------------+
| alice |  10.0 |  math,science   |
|  bob  |  20.0 |  history,math   |
| carol |  30.0 | science,history |
| dave  |  40.0 |    math,art     |
|  eve  |  50.0 |  science,art    |
+-------+-------+-----------------+

from pyspark.sql.functions import udtf

@udtf(returnType="score: int, categories: string, name: string")
class ScoreCategoriesUDTF:
    def eval(self, name: str, score: float, categories: str):
        category_list = categories.split(',')
        for category in category_list:
            yield (name, score, category)

# Apply the UDTF
result_df = df.select(ScoreCategoriesUDTF(df.score, df.categories, df.name))
display(result_df)

+-------+-------+----------+
| name  | score | category |
+-------+-------+----------+
| alice |  10.0 |   math   |
| alice |  10.0 | science  |
|  bob  |  20.0 | history  |
|  bob  |  20.0 |   math   |
| carol |  30.0 | science  |
| carol |  30.0 | history  |
| dave  |  40.0 |   math   |
| dave  |  40.0 |   art    |
|  eve  |  50.0 | science  |
|  eve  |  50.0 |   art    |
+-------+-------+----------+

Considerações sobre desempenho

• Funções integradas e UDFs SQL são a opção mais eficiente disponível.
• As UDFs Scala geralmente são mais rápidas , pois são executadas na Java Virtual Machine (JVM) e evitam a sobrecarga de mover dados para dentro e para fora da JVM.
• As UDFs do Python e do Pandas tendem a ser mais lentas do que as UDFs do Scala porque exigem que os dados sejam serializados e movidos da JVM para o interpretador do Python. As UDFs do Pandas são até 100 vezes mais rápidas do que as UDFs do Python porque usam o Apache Arrow para reduzir os custos de serialização.