O que são funções definidas pelo usuário (UDFs)?
Funções definidas pelo usuário (UDFs) permitem reutilizar e compartilhar código que estende as funcionalidades integradas no Databricks. Use UDFs para executar tarefas específicas, como cálculos complexos, transformações ou manipulações de dados personalizadas.
Quando usar um UDF em vez de uma função do Apache Spark?
Use UDFs para lógica que seja difícil de expressar com funções integradas Apache Spark. integrada Apache Spark funções são otimizadas para processamento distribuído e oferecem melhor desempenho em escala. Para obter mais informações, consulte Funções.
A Databricks recomenda UDFs para queries ad hoc, limpeza manual de dados, análise exploratória de dados e operações em datasets pequenos a médios. Os casos de uso comuns para UDFs incluem criptografia de dados, descriptografia, hashing, análise de JSON e validação.
Use métodos do Apache Spark para operações em grandes datasets e quaisquer workloads que executam regularmente ou continuamente, incluindo Jobs ETL e operações de transmissão.
Compreender os tipos de UDF
Selecione um tipo de UDF na guia a seguir para ver uma descrição, um exemplo e um link para saber mais.
- Scalar UDF
- Batch Scalar UDFs
- Non-Scalar UDFs
- UDAF
- UDTFs
As UDFs escalares operam em uma única linha e retornam um único valor de resultado para cada linha. Eles podem ser controlados pelo Unity Catalog ou com escopo de sessão.
O exemplo a seguir usa um UDF escalar para calcular o comprimento de cada nome em uma coluna name 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 main.test.get_name_length(name STRING)
RETURNS INT
RETURN LENGTH(name);
-- Use the UDF in a SQL query
SELECT name, main.test.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 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)
Consulte funções definidas pelo usuário (UDFs) de SQL e Python no Unity Catalog e funções definidas pelo usuário (UDFs) Python escalares.
Processar dados em lotes, mantendo a paridade de linha de entrada/saída de 1:1. Isso reduz a sobrecarga das operações linha a linha para o processamento de dados em grande escala. Os UDFs de lotes também mantêm o estado entre os lotes para execução mais eficiente, reutilizam recursos e lidam com cálculos complexos que precisam de contexto em blocos de dados.
Eles podem ser controlados pelo Unity Catalog ou com escopo de sessão.
Os seguintes lotes Unity Catalog Python UDF calculam o IMC enquanto processam lotes de linhas:
+-------------+-------------+
| weight_kg | height_m |
+-------------+-------------+
| 90 | 1.8 |
| 77 | 1.6 |
| 50 | 1.5 |
+-------------+-------------+
%sql
CREATE OR REPLACE FUNCTION main.test.calculate_bmi_pandas(weight_kg DOUBLE, height_m DOUBLE)
RETURNS DOUBLE
LANGUAGE PYTHON
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)
$$;
select main.test.calculate_bmi_pandas(cast(70 as double), cast(1.8 as double));
+--------+
| BMI |
+--------+
| 27.8 |
| 30.1 |
| 22.2 |
+--------+
Consulte funções definidas pelo usuário (UDFs) de SQL e Python no Unity Catalog e funções definidas pelo usuário (UDFs) Python em lote no Unity Catalog.
Os UDFs não escalares operam em conjuntos de dados/colunas inteiros com relações de entrada/saída flexíveis (1 ou muitos).
Lotes com escopo de sessão Pandas Os UDFs podem ser dos seguintes tipos:
- Série para série
- Iterador de série para iterador de série
- Iterador de várias séries para iterador de séries
- Série em escalar
A seguir, um exemplo de uma série para série Pandas UDF.
from pyspark.sql.functions import pandas_udf
import pandas as pd
df = spark.createDataFrame([(70, 1.75), (80, 1.80), (60, 1.65)], ["Weight", "Height"])
@pandas_udf("double")
def calculate_bmi_pandas(weight: pd.Series, height: pd.Series) -> pd.Series:
return weight / (height ** 2)
df.withColumn("BMI", calculate_bmi_pandas(df["Weight"], df["Height"])).display()
Consulte Pandas funções definidas pelo usuário.
Os UDAFs operam em várias linhas e retornam um único resultado agregado. Os UDAFs têm escopo de sessão apenas.
O exemplo de UDAF a seguir agrega pontuações por tamanho de nome.
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 de agregação definidas pelo usuário (UDAFs) do Scala.
Uma UDTF usa um ou mais argumentos de entrada e retorna várias linhas (e possivelmente várias colunas) para cada linha de entrada. Eles podem ser controlados pelo Unity Catalog ou com escopo de sessão.
A UDTF a seguir cria uma tabela usando uma lista fixa de dois argumentos inteiros:
CREATE OR REPLACE FUNCTION get_sum_diff(x INT, y INT)
RETURNS TABLE (sum INT, diff INT)
LANGUAGE PYTHON
HANDLER 'GetSumDiff'
AS $$
class GetSumDiff:
def eval(self, x: int, y: int):
yield x + y, x - y
$$;
SELECT * FROM get_sum_diff(10, 3);
+-----+------+
| sum | diff |
+-----+------+
| 13 | 7 |
+-----+------+
Para implementar isso em um Notebook do Databricks usando PySpark:
from pyspark.sql.functions import lit, udtf
@udtf(returnType="sum: int, diff: int")
class GetSumDiff:
def eval(self, x: int, y: int):
yield x + y, x - y
GetSumDiff(lit(1), lit(2)).show()
Consulte UDTFs do Unity Catalog e UDTFs com escopo de sessão.
UDFs governadas pelo Unity Catalog vs. UDFs com escopo de sessão
O Unity Catalog persiste as UDFs governadas pelo Unity Catalog para uma governança, reutilização e capacidade de descoberta aprimoradas. Você define as UDFs com escopo de sessão em um Notebook ou Job, com escopo para a SparkSession atual. Você pode definir e acessar UDFs com escopo de sessão usando SQL, Python ou Scala.
Use a tabela a seguir para decidir entre as duas categorias e, em seguida, consulte as folhas de dicas a seguir para os tipos específicos de UDF em cada uma.
Consideração | UDFs regidas pelo Unity Catalog | UDFs com escopo de sessão |
|---|---|---|
Melhor para | Compartilhamento seguro de funções entre equipes, notebooks, jobs e SQL Warehouse. | Desenvolvimento rápido e iterativo em um único Notebook ou Job. |
Idiomas | SQL, Python, Scala e Java. | SQL, Python e Scala. |
Governança e compartilhamento | Regido pelas permissões do Unity Catalog e detectável no Catalog Explorer. | Com escopo para a SparkSession atual. Não governado nem compartilhado. |
Persistência | Persistido no Unity Catalog e reutilizável entre sessões. | Existem apenas para a sessão atual. |
Folha de dicas de UDFs governados pelo Unity Catalog
As UDFs governadas pelo Unity Catalog permitem que funções personalizadas sejam definidas, usadas, compartilhadas com segurança e governadas em todos os ambientes de computação. Consulte funções definidas pelo usuário (UDFs) de SQL e Python no Unity Catalog.
Tipo de UDF | Com suporte compute | Descrição |
|---|---|---|
Unity Catalog Python UDF |
| Defina um UDF em Python e registre-o em Unity Catalog para governança. As UDFs escalares operam em uma única linha e retornam um único valor de resultado para cada linha. |
lotes Unity Catalog Python UDF |
| Defina um UDF em Python e registre-o em Unity Catalog para governança. lotes operações em vários valores e retornam vários valores. Reduz a sobrecarga de operações linha a linha para processamento de dados em grande escala. |
Unity Catalog Python UDTF |
| Defina uma UDTF em Python e registre-a em Unity Catalog para governança. Uma UDTF usa um ou mais argumentos de entrada e retorna várias linhas (e possivelmente várias colunas) para cada linha de entrada. |
UDF de Scala ou Java do Unity Catalog |
| Defina uma UDF em Scala ou Java e registre-a no Unity Catalog para governança. As UDFs escalares operam em uma única linha e retornam um único valor de resultado para cada linha. Requer Scala 2.13.16, JDK 17 e Versão 4 do Ambiente. |
Folha de dicas de UDFs com escopo de sessão para usuários isolados compute
Você define UDFs com escopo de sessão em um notebook ou job, com escopo para a SparkSession atual. Você pode definir e acessar UDFs com escopo de sessão usando SQL, Python ou Scala.
Tipo de UDF | Com suporte compute | Descrição |
|---|---|---|
Escalar Python |
| As UDFs escalares operam em uma única linha e retornam um único valor de resultado para cada linha. |
Python não escalar |
| Os UDFs não escalares incluem |
Python UDTFs |
| Uma UDTF usa um ou mais argumentos de entrada e retorna várias linhas (e possivelmente várias colunas) para cada linha de entrada. |
UDFs escalares em Scala |
| As UDFs escalares operam em uma única linha e retornam um único valor de resultado para cada linha. |
UDF Scala ou Java de um JAR |
| Registre uma classe UDF pré-compilada de um JAR usando |
UDAFs em Scala |
| Os UDAFs operam em várias linhas e retornam um único resultado agregado. |
Considerações sobre o desempenho
-
As funções integradas e os UDFs do siteSQL são as opções mais eficientes.
-
As UDFs do Scala geralmente são mais rápidas do que as UDFs do Python.
- Unisolated Scala Os UDFs são executados na máquina virtual Java (JVM), portanto, evitam a sobrecarga de mover dados para dentro e para fora do JVM.
- UDFs Scala isoladas devem mover dados para dentro e para fora da JVM, mas ainda podem ser mais rápidas do que as UDFs Python porque gerenciam a memória de forma mais eficiente.
-
UDFs Python e UDFs pandas tendem a ser mais lentas que as UDFs Scala porque devem serializar dados e movê-los para fora da JVM para o interpretador 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.