pyflink.dataframe.DataFrame.flat_map

DataFrame.flat_map(func: Union[Callable[[Dict[str, Any]], Any], DataFrameUDTFWrapper], *, return_dtype: Optional[DataFrameDataTypeLike] = None, concurrency: Optional[int] = None, num_gpus: Optional[float] = None, gpu_type: Optional[str] = None) → DataFrame

Apply a row-based UDTF to each row, producing zero or more rows.

Raw Python functions receive a dict[str, Any] keyed by DataFrame column name. The result contains only the emitted UDTF columns. When type hints are present, raw Python functions should annotate the row parameter as dict-like, while DataFrameUDTFWrapper functions should annotate it as Row or leave it unannotated. Output column names are inferred from named TypedDict or struct return types. Scalar outputs and unnamed single-field struct outputs use f0. Unnamed multi-field outputs are rejected; use a TypedDict return type or named struct return_dtype instead. When func is a DataFrameUDTFWrapper, return_dtype must not be specified because the wrapper already carries its return type. Resource arguments supplied to flat_map override resources defined on the wrapper for this operation.

Example::

>>> from typing import Any, Dict, Iterator, TypedDict
>>> from pyflink.dataframe import DataType
>>>
>>> class Output(TypedDict):
...     value: int
>>>
>>> def expand(row: Dict[str, Any]) -> Iterator[Output]:
...     yield {"value": row["x"]}
...     yield {"value": row["x"] + 1}
>>>
>>> df.flat_map(expand)
>>>
>>> def expand_with_dtype(row):
...     yield {"value": row["x"]}
...     yield {"value": row["x"] + 1}
>>>
>>> df.flat_map(
...     expand_with_dtype,
...     return_dtype=DataType.struct({"value": DataType.int64()}),
... )