udtf

Cria uma função de tabela definida pelo usuário (UDTF).

Sintaxe

Python

import pyspark.sql.functions as sf

# As a decorator
@sf.udtf(returnType=<returnType>, useArrow=<useArrow>)
class FunctionClass:
    def eval(self, *args):
        # function body
        yield row_data

# As a function wrapper
sf.udtf(cls=<class>, returnType=<returnType>, useArrow=<useArrow>)

Parâmetros

Parâmetro	Tipo	Descrição
cls	class	Opcional. Classe de manipulador de função de tabela definida pelo usuário em Python.
returnType	pyspark.sql.types.StructType ou str	Opcional. O tipo de retorno da função de tabela definida pelo usuário. O valor pode ser um objeto StructType ou uma string de tipo struct formatada em DDL. Se None, a classe manipuladora deve fornecer um método estático analyze .
useArrow	bool	Opcional. Se deve ou não utilizar o Arrow para otimizar as (des)serializações. Quando definido como None, a configuração do Spark "spark.sql.execution.pythonUDTF.arrow.enabled" é utilizada.

Exemplos

Exemplo 1 : Implementação básica de UDTF.

Python

from pyspark.sql.functions import udtf

class TestUDTF:
    def eval(self, *args):
        yield "hello", "world"

test_udtf = udtf(TestUDTF, returnType="c1: string, c2: string")
test_udtf().show()

Output

+-----+-----+
|   c1|   c2|
+-----+-----+
|hello|world|
+-----+-----+

Exemplo 2 : UDTF usando sintaxe de decorador.

Python

from pyspark.sql.functions import udtf, lit

@udtf(returnType="c1: int, c2: int")
class PlusOne:
    def eval(self, x: int):
        yield x, x + 1

PlusOne(lit(1)).show()

Output

+---+---+
| c1| c2|
+---+---+
|  1|  2|
+---+---+

Exemplo 3 : UDTF com método estático de análise.

Python

from pyspark.sql.functions import udtf, lit
from pyspark.sql.types import StructType
from pyspark.sql.udtf import AnalyzeArgument, AnalyzeResult

@udtf
class TestUDTFWithAnalyze:
    @staticmethod
    def analyze(a: AnalyzeArgument, b: AnalyzeArgument) -> AnalyzeResult:
        return AnalyzeResult(StructType().add("a", a.dataType).add("b", b.dataType))

    def eval(self, a, b):
        yield a, b

TestUDTFWithAnalyze(lit(1), lit("x")).show()

Output

+---+---+
|  a|  b|
+---+---+
|  1|  x|
+---+---+

Exemplo 4 : UDTF com argumentos nomeados.

Python

from pyspark.sql.functions import udtf, lit
from pyspark.sql.types import StructType
from pyspark.sql.udtf import AnalyzeArgument, AnalyzeResult

@udtf
class TestUDTFWithKwargs:
    @staticmethod
    def analyze(
        a: AnalyzeArgument, b: AnalyzeArgument, **kwargs: AnalyzeArgument
    ) -> AnalyzeResult:
        return AnalyzeResult(
            StructType().add("a", a.dataType)
                .add("b", b.dataType)
                .add("x", kwargs["x"].dataType)
        )

    def eval(self, a, b, **kwargs):
        yield a, b, kwargs["x"]

TestUDTFWithKwargs(lit(1), x=lit("x"), b=lit("b")).show()

Output

+---+---+---+
|  a|  b|  x|
+---+---+---+
|  1|  b|  x|
+---+---+---+

Exemplo 5 : UDTF registrada e chamada via SQL.

Python

from pyspark.sql.functions import udtf, lit
from pyspark.sql.types import StructType
from pyspark.sql.udtf import AnalyzeArgument, AnalyzeResult

@udtf
class TestUDTFWithKwargs:
    @staticmethod
    def analyze(
        a: AnalyzeArgument, b: AnalyzeArgument, **kwargs: AnalyzeArgument
    ) -> AnalyzeResult:
        return AnalyzeResult(
            StructType().add("a", a.dataType)
                .add("b", b.dataType)
                .add("x", kwargs["x"].dataType)
        )

    def eval(self, a, b, **kwargs):
        yield a, b, kwargs["x"]

_ = spark.udtf.register("test_udtf", TestUDTFWithKwargs)
spark.sql("SELECT * FROM test_udtf(1, x => 'x', b => 'b')").show()

Output

+---+---+---+
|  a|  b|  x|
+---+---+---+
|  1|  b|  x|
+---+---+---+

Exemplo 6 : UDTF com otimização Arrow ativada.

Python

from pyspark.sql.functions import udtf, lit

@udtf(returnType="c1: int, c2: int", useArrow=True)
class ArrowPlusOne:
    def eval(self, x: int):
        yield x, x + 1

ArrowPlusOne(lit(1)).show()

Output

+---+---+
| c1| c2|
+---+---+
|  1|  2|
+---+---+

Exemplo 7 : Criando uma UDTF determinística.

Python

from pyspark.sql.functions import udtf

class PlusOne:
    def eval(self, a: int):
        yield a + 1,

plus_one = udtf(PlusOne, returnType="r: int").asDeterministic()