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

As funções definidas pelo utilizador (UDFs) permitem-lhe reutilizar e partilhar código que estende a funcionalidade incorporada no Azure Databricks. Use UDFs para executar tarefas específicas, como cálculos complexos, transformações ou manipulações de dados personalizadas.

Quando se deve usar um UDF?

Use UDFs para lógica difícil de expressar com funções integradas do Apache Spark. As funções integradas 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.

O 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. Os casos de uso comuns para UDFs incluem criptografia e descriptografia de dados, hashing, análise JSON e validação.

Use os métodos Apache Spark para operações em conjuntos de dados muito grandes e quaisquer cargas de trabalho que são executadas regularmente ou continuamente, incluindo trabalhos ETL e operações de streaming.

UDFs registrados e com escopo de sessão

As UDFs criadas usando SQL são registradas no Unity Catalog e têm permissões associadas, enquanto as UDFs criadas em seu bloco de anotações são baseadas em sessão e têm como escopo a SparkSession atual.

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

UDFs escalares

UDFs escalares operam em uma única linha e retornam um único valor para cada linha. O exemplo a seguir usa um 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 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 Funções definidas pelo usuário (UDFs) no Unity Catalog e Funções escalares definidas pelo usuário - Python.

Funções de agregação definidas pelo utilizador (UDAFs)

As funções de agregação definidas pelo usuário (UDAFs) 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 pandas definidas pelo usuário para Python e funções agregadas definidas pelo usuário - Scala.

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

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 Funções de tabela definidas pelo usuário Python (UDTFs)

+-------+-------+-----------------+
| 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 internas 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 dentro da Java Virtual Machine (JVM) e evitam a sobrecarga de mover dados para dentro e para fora da JVM.
• UDFs Python e UDFs Pandas tendem a ser mais lentos do que UDFs Scala porque exigem que os dados sejam serializados e movidos da JVM para o interpretador Python. Pandas UDFs até 100x mais rápido do que Python UDFs porque eles usam Apache Arrow para reduzir os custos de serialização.