pyflink.dataframe.udtf#

udtf(func: Union[Callable[[...], Any], pyflink.table.udf.TableFunction, Type[pyflink.table.udf.TableFunction]], *, return_dtype: Optional[Union[pyflink.dataframe.datatype.DataType, type, str]] = None, deterministic: bool = True, name: Optional[str] = None, concurrency: Optional[int] = None, num_gpus: Optional[float] = None, gpu_type: Optional[str] = None) → pyflink.dataframe.udf.DataFrameUDTFWrapper[source]#

udtf(func: None = None, *, return_dtype: Optional[Union[pyflink.dataframe.datatype.DataType, type, str]] = None, deterministic: bool = True, name: Optional[str] = None, concurrency: Optional[int] = None, num_gpus: Optional[float] = None, gpu_type: Optional[str] = None) → Callable[[Union[Callable[[...], Any], pyflink.table.udf.TableFunction, Type[pyflink.table.udf.TableFunction]]], pyflink.dataframe.udf.DataFrameUDTFWrapper]

Create a user-defined table function for DataFrame operations.

This decorator wraps: - Plain Python functions and lambdas - TableFunction subclasses or instances, using the eval method - Plain callable class instances, using the __call__ method

The wrapped object must emit zero or more rows by returning an Iterable/Iterator/list, or by yielding from a generator. Each emitted row may be a scalar value, tuple/Row, or dict when output fields are named.

Parameters

func – The Python function, TableFunction instance/subclass, or callable class instance to wrap as a UDTF.
return_dtype – Optional emitted row type. Can be: - DataType instance (e.g., DataType.string()) - Python type (e.g., str, int) - Struct DataType for multi-column output If not specified, inferred from the function’s return type hint (eval for TableFunction, __call__ for callable class instances).
deterministic – Whether the function is deterministic (default: True).
name – Optional name for the UDTF.
concurrency – Optional concurrency (parallelism) for the UDTF operator. If specified, the operator running this UDTF will use this parallelism.
num_gpus – Optional number of GPUs requested for this UDTF (e.g., 0.5, 1.0).
gpu_type – Optional GPU type (e.g., ‘A10’, ‘V100’). Required when num_gpus is set.

Returns

A DataFrameUDTFWrapper that can be called with Expressions and passed to join_lateral, or passed directly to flat_map.

Example::

>>> from typing import Iterator, TypedDict
>>> from pyflink.dataframe import DataType, col, udtf
>>> from pyflink.table.udf import TableFunction
>>>
>>> # 1. Plain generator function with explicit return_dtype
>>> @udtf(return_dtype=DataType.string())
... def split_words(text):
...     for word in text.split():
...         yield word
>>>
>>> df.join_lateral(split_words(col("text")).alias("word"))
>>>
>>> # 2. Plain function with Iterator[T] type hint inference
>>> @udtf
... def chars(text: str) -> Iterator[str]:
...     for ch in text:
...         yield ch
>>>
>>> df.join_lateral(chars(col("text")).alias("ch"))
>>>
>>> # 3. TypedDict output supplies output column names
>>> class Token(TypedDict):
...     word: str
...     length: int
>>>
>>> @udtf
... def tokenize(text: str) -> Iterator[Token]:
...     for word in text.split():
...         yield {"word": word, "length": len(word)}
>>>
>>> df.join_lateral(tokenize(col("text")))
>>>
>>> # 4. Tuple output infers field types; provide output aliases
>>> from typing import Tuple
>>> @udtf
... def pair(x: int) -> Iterator[Tuple[int, str]]:
...     yield x, str(x)
>>>
>>> df.join_lateral(pair(col("a")).alias("value", "text"))
>>>
>>> # 5. Parameterized decorator with resources
>>> @udtf(return_dtype=DataType.string(), concurrency=4,
...       num_gpus=0.5, gpu_type="A10")
... def gpu_split(text):
...     return text.split()
>>>
>>> df.join_lateral(gpu_split(col("text")).alias("word"))
>>>
>>> # 6. Direct call form
>>> split_words = udtf(lambda text: text.split(),
...                    return_dtype=DataType.string())
>>> df.join_lateral(split_words(col("text")).alias("word"))
>>>
>>> # 7. TableFunction instance or class
>>> class Split(TableFunction):
...     def eval(self, text: str) -> Iterator[str]:
...         for word in text.split():
...             yield word
>>>
>>> split = udtf(Split())
>>> split = udtf(Split)  # auto-instantiated
>>> df.join_lateral(split(col("text")).alias("word"))
>>>
>>> # 8. Callable class instance
>>> class Repeat:
...     def __call__(self, value: str) -> Iterator[str]:
...         yield value
...         yield value
>>>
>>> repeat = udtf(Repeat())
>>> df.join_lateral(repeat(col("text")).alias("value"))
>>>
>>> # 9. Left outer lateral join semantics
>>> df.join_lateral(chars(col("text")).alias("ch"), ignore_empty=False)