dianfu commented on code in PR #29234:
URL: https://github.com/apache/flink/pull/29234#discussion_r4056514987
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -135,6 +136,57 @@ class DataFrame:
def __init__(self, table: Table):
self._table = table
+ @PublicEvolving()
+ def flat_map(
+ self,
+ func: Union[Callable[[Dict[str, Any]], Any], "_DataFrameUDTFWrapper"],
+ *,
+ return_dtype: Optional["_DataTypeLike"] = None,
+ ) -> "DataFrame":
+ """
+ Apply a function to each row, emitting zero or more output rows.
+
+ A plain callable receives a dictionary keyed by column name. A
declaration
+ created with :func:`pyflink.dataframe.udtf` receives a named Flink
``Row``.
+ Output column names come from a ``TypedDict`` or an explicit named
struct;
+ scalar outputs use ``f0``. Multi-field outputs require named fields.
+
+ :param func: Row-based callable or a declaration created with
``pf.udtf``.
+ :param return_dtype: Emitted row type, inferred from annotations when
omitted.
+ Required if inference is not possible; must be
omitted
+ for a UDTF declaration.
+ :return: A DataFrame containing only the emitted output columns.
+
+ Example::
Review Comment:
It would also be useful to demonstrate the explicit `return_dtype` path, the
output column names, udfs defined with Callable Classes, Classes extending
TableFunction, etc?
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -135,6 +136,57 @@ class DataFrame:
def __init__(self, table: Table):
self._table = table
+ @PublicEvolving()
+ def flat_map(
+ self,
+ func: Union[Callable[[Dict[str, Any]], Any], "_DataFrameUDTFWrapper"],
+ *,
+ return_dtype: Optional["_DataTypeLike"] = None,
+ ) -> "DataFrame":
+ """
+ Apply a function to each row, emitting zero or more output rows.
+
+ A plain callable receives a dictionary keyed by column name. A
declaration
+ created with :func:`pyflink.dataframe.udtf` receives a named Flink
``Row``.
+ Output column names come from a ``TypedDict`` or an explicit named
struct;
+ scalar outputs use ``f0``. Multi-field outputs require named fields.
+
+ :param func: Row-based callable or a declaration created with
``pf.udtf``.
+ :param return_dtype: Emitted row type, inferred from annotations when
omitted.
+ Required if inference is not possible; must be
omitted
+ for a UDTF declaration.
+ :return: A DataFrame containing only the emitted output columns.
+
+ Example::
+
+ >>> from typing import Any, Dict, Iterator, TypedDict
+ >>> import pyflink.dataframe as pf
+ >>> class Token(TypedDict):
+ ... word: str
+ >>> def split(row: Dict[str, Any]) -> Iterator[Token]:
+ ... for word in row["text"].split():
+ ... yield {"word": word}
+ >>> df = pf.from_dict({"text": ["hello world", "flink"]})
+ >>> result = df.flat_map(split)
+
+ A reusable UDTF declaration receives a named ``Row``::
+
+ >>> from pyflink.common import Row
+ >>> @pf.udtf
Review Comment:
It's strange that when decorated with @pf.udtf, the input type is Row,
otherwise, it's a dict. Why not always make it a dict?
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -135,6 +136,57 @@ class DataFrame:
def __init__(self, table: Table):
self._table = table
+ @PublicEvolving()
+ def flat_map(
Review Comment:
Could you move it to the end of "Core Operations" group?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
