auroflow commented on code in PR #28934:
URL: https://github.com/apache/flink/pull/28934#discussion_r3780650035
##########
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),
+ schema=schema,
+ watermark=watermark,
+ )
+
+
+@PublicEvolving()
+def from_arrow(
+ table: Any,
+ schema: Optional[List[str]] = None,
+ watermark: Optional[Tuple[str, str]] = None,
+) -> DataFrame:
+ """
+ Create a DataFrame from a PyArrow Table without converting through pandas.
+
+ An explicit ``schema`` renames columns positionally and must contain
exactly one unique,
+ non-empty name per input column. Empty tables are supported when their
Arrow field types 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 table: PyArrow Table to convert.
+ :param schema: Optional list of positional result column names.
+ :param watermark: Optional ``(column, expression)`` watermark declaration.
+ :return: A DataFrame containing the Arrow 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 pyarrow as pa
+ >>> import pyflink.dataframe as pf
+ >>> table = pa.table({"id": [1, 2], "name": ["Alice", "Bob"]})
+ >>> dataframe = pf.from_arrow(table)
+ >>> events = pf.from_arrow(
+ ... pa.table({"ts": pa.array([0], type=pa.timestamp("ms"))}),
+ ... watermark=("ts", "ts - INTERVAL '5' SECOND"),
+ ... )
+
+ .. versionadded:: 2.4.0
+ """
+ import pyarrow as pa
+
+ if not isinstance(table, pa.Table):
+ raise TypeError(
+ f"data must be a pyarrow.Table, but was {type(table).__name__}"
+ )
+ watermark_spec = _WatermarkSpec.parse(watermark)
+ names = _resolve_column_names(table.column_names, schema)
+ row_type = RowType(
+ [
+ RowField(name, from_arrow_type(field.type, field.nullable))
Review Comment:
This was to bypass a limitation in the Java Arrow Coder. Previously, Java
Arrow reader rejected timezone-aware Arrow timestamp vectors. Therefore, when
constructing a TIMESTAMP_LTZ column, PyFlink had to cast the Arrow timestamps
to an equivalent timezone-less representation while preserving their epoch
values. I changed the Java Arrow reader to accept timezone-aware timestamps
directly when the target type is TIMESTAMP_LTZ, so PyFlink can preserve the
Arrow timezone metadata and no longer needs that normalization.
--
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]
