udtf

Vytvoří uživatelem definovanou funkci tabulky (UDTF).

Syntaxe

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>)

Parametry

Parameter	Typ	Description
cls	class	Optional. Třída obslužné rutiny funkce definované uživatelem v Pythonu
returnType	pyspark.sql.types.StructType nebo str	Optional. Návratový typ uživatelem definované funkce tabulky. Hodnota může být buď objekt StructType nebo řetězec typu struktury ve formátu DDL. Pokud None, třída obslužné rutiny musí poskytovat analyze statickou metodu.
useArrow	bool	Optional. Zda použít šipku k optimalizaci (de)serializace. Pokud je nastavena na Hodnotu Žádné, použije se konfigurace Sparku "spark.sql.execution.pythonUDTF.arrow.enabled".

Examples

Příklad 1: Základní implementace UDTF.

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()

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

Příklad 2: UDTF pomocí syntaxe dekorátoru

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()

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

Příklad 3: UDTF s analýzou statické metody

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()

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

Příklad 4: UDTF s argumenty klíčových slov

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()

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

Příklad 5: UDTF zaregistrováno a volána prostřednictvím SQL.

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()

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

Příklad 6: UDTF s povolenou optimalizací šipky

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()

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

Příklad 7: Vytvoření deterministické funkce UDTF

from pyspark.sql.functions import udtf

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

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