auroflow commented on code in PR #28934:
URL: https://github.com/apache/flink/pull/28934#discussion_r3780599479
##########
flink-python/pyflink/dataframe/convert.py:
##########
@@ -120,10 +183,202 @@ def _validate_schema(schema: List[str]) -> None:
raise ValueError("schema field names must be unique")
+def _resolve_column_names(
+ input_names: Sequence[str], schema: Optional[List[str]]
+) -> List[str]:
+ column_names = list(input_names) if schema is None else schema
+ if (
+ schema is not None
+ and isinstance(schema, list)
+ and len(schema) != len(input_names)
+ ):
+ raise ValueError(
+ f"schema has {len(schema)} fields but data has "
+ f"{len(input_names)} columns"
+ )
+ _validate_schema(column_names)
+ return column_names
+
+
+def _infer_schema_and_create_dataframe(
+ rows: Sequence[Sequence[Any]],
+ column_names: List[str],
+ watermark: Optional[_WatermarkSpec] = None,
+) -> DataFrame:
+ row_type = _infer_schema_from_data(rows, names=column_names)
+ if watermark is None:
+ table_schema = None
+ else:
+ row_type = watermark.normalize_row_type(row_type)
+ table_schema = (
+ Schema.new_builder()
+ .from_row_data_type(row_type)
+ .watermark(*watermark)
+ .build()
+ )
+ converter = _create_converter(row_type)
+ verify_row = _create_type_verifier(row_type)
+ sql_rows = []
+ for row in rows:
+ row = converter(row)
+ verify_row(row)
+ sql_rows.append(row_type.to_sql_type(row))
+
+ table = get_or_create_table_environment()._from_elements(
+ sql_rows, row_type, table_schema
+ )
+ return DataFrame(table)
+
+
+@PublicEvolving()
+def from_table(table: Table) -> DataFrame:
+ """
+ Create a DataFrame that wraps a PyFlink Table.
+
+ :param table: Table to wrap without copying or converting it.
+ :return: A DataFrame backed by the exact supplied Table.
+ :raises TypeError: If ``table`` is not a :class:`~pyflink.table.Table`.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> table = table_env.from_elements([(1, "Alice")], ["id", "name"])
+ >>> dataframe = pf.from_table(table)
+ >>> dataframe.to_table() is table
+ True
+
+ .. versionadded:: 2.4.0
+ """
+ if not isinstance(table, Table):
+ raise TypeError("table must be a pyflink.table.Table")
+ return DataFrame(table)
+
+
+@PublicEvolving()
+def from_pandas(
+ pdf: Any,
+ schema: Optional[List[str]] = None,
+ watermark: Optional[Tuple[str, str]] = None,
+) -> DataFrame:
+ """
+ Create a DataFrame from a pandas DataFrame.
+
+ Types are inferred from the Arrow representation of the pandas columns. An
explicit ``schema``
+ renames columns positionally and must contain exactly one unique,
non-empty name per input
+ column. Empty inputs are supported when their pandas dtypes can be
converted to Flink types.
+
+ ``watermark`` declares an event-time column and its SQL watermark
expression. The selected
+ column must have a timestamp-compatible type. Its precision is normalized
to milliseconds;
+ values with finer precision are truncated to ``TIMESTAMP(3)`` or
``TIMESTAMP_LTZ(3)``.
+
+ :param pdf: pandas DataFrame to convert.
+ :param schema: Optional list of positional result column names.
+ :param watermark: Optional ``(column, expression)`` watermark declaration.
+ :return: A DataFrame containing the pandas rows.
+ :raises TypeError: If the input, schema, watermark, or inferred types are
invalid.
+ :raises ValueError: If schema width or watermark column requirements are
not met.
+
+ Example::
+
+ >>> import pandas as pd
+ >>> import pyflink.dataframe as pf
+ >>> pdf = pd.DataFrame({"identifier": [1, 2], "name": ["Alice",
"Bob"]})
+ >>> dataframe = pf.from_pandas(pdf, schema=["id", "name"])
+ >>> events = pf.from_pandas(
+ ... pd.DataFrame({"ts": pd.to_datetime(["2026-01-01T00:00:00Z"])}),
+ ... watermark=("ts", "ts - INTERVAL '5' SECOND"),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ import pandas as pd
+
+ if not isinstance(pdf, pd.DataFrame):
+ raise TypeError(
+ f"data must be a pandas.DataFrame, but was {type(pdf).__name__}"
+ )
+
+ import pyarrow as pa
+
+ return from_arrow(
+ pa.Table.from_pandas(pdf, preserve_index=False),
Review Comment:
I have assigned temporary names during construction to bypass this problem.
--
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]
