Yicong-Huang commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4031070937
##########
python/pyspark/sql/conversion.py:
##########
@@ -410,105 +401,161 @@ def convert_column(
assign_cols_by_name=assign_cols_by_name,
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- is_legacy=is_legacy,
+ use_legacy_error_handling=use_legacy_error_handling,
)
# Wrap the nested RecordBatch as a single StructArray column
- return
ArrowBatchTransformer.wrap_struct(nested_batch).column(0)
+
converted.append(ArrowBatchTransformer.wrap_struct(nested_batch).column(0))
+ else:
+ assert isinstance(col, pd.Series)
+ converted.append(
+ cls._convert_column(
+ col,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
+ )
- series = col
- field_name = field.name
- ret_type = field.dataType
+ # pa.Array.from_pandas returns a pa.ChunkedArray for a chunked
arrow-backed Series
+ # (e.g. a pyarrow-backed extension dtype), which
pa.RecordBatch.from_arrays rejects.
+ arrays = [a.combine_chunks() if isinstance(a, pa.ChunkedArray) else a
for a in converted]
+ return pa.RecordBatch.from_arrays(arrays, schema.names)
- if isinstance(series.dtype, pd.CategoricalDtype):
- series = series.astype(series.dtype.categories.dtype)
+ @classmethod
+ def _convert_column(
+ cls,
+ series: "pd.Series",
+ field: StructField,
+ *,
+ timezone: Optional[str] = None,
+ safecheck: bool = True,
+ arrow_cast: bool = False,
+ prefers_large_types: bool = False,
+ int_to_decimal_coercion_enabled: bool = False,
+ ignore_unexpected_complex_type_values: bool = False,
+ use_legacy_error_handling: bool = False,
+ ) -> Union["pa.Array", "pa.ChunkedArray"]:
+ """Dispatch a pandas column to its conversion strategy."""
+ # Future NumPy and PyArrow strategies will be selected here; use
legacy for now.
+ return cls._convert_column_legacy(
+ series,
+ field,
+ timezone=timezone,
+ safecheck=safecheck,
+ arrow_cast=arrow_cast,
+ prefers_large_types=prefers_large_types,
+ int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
+
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
+ use_legacy_error_handling=use_legacy_error_handling,
+ )
- arrow_type = to_arrow_type(
- ret_type, timezone=timezone,
prefers_large_types=prefers_large_types
- )
- series = _create_converter_from_pandas(
- ret_type,
- timezone=timezone,
- error_on_duplicated_field_names=False,
-
int_to_decimal_coercion_enabled=int_to_decimal_coercion_enabled,
-
ignore_unexpected_complex_type_values=ignore_unexpected_complex_type_values,
- )(series)
+ @classmethod
+ def _convert_column_legacy(
+ cls,
+ series: "pd.Series",
+ field: StructField,
+ *,
+ timezone: Optional[str] = None,
+ safecheck: bool = True,
+ arrow_cast: bool = False,
+ prefers_large_types: bool = False,
+ int_to_decimal_coercion_enabled: bool = False,
+ ignore_unexpected_complex_type_values: bool = False,
+ use_legacy_error_handling: bool = False,
+ ) -> Union["pa.Array", "pa.ChunkedArray"]:
+ """Convert a pandas column using the legacy conversion strategy."""
+ import pandas as pd
+ import pyarrow as pa
+
+ from pyspark.errors import PySparkTypeError, PySparkValueError
+ from pyspark.sql.pandas.types import _create_converter_from_pandas,
to_arrow_type
+
+ field_name = field.name
Review Comment:
Optional, non-blocking (follow-up): the old `convert_column` docstring
explained *why* it uses `field.name` instead of `series.name` here — avoiding a
`Series.rename()` copy, ~20% on the hot path. That rationale was dropped in the
refactor while `field.name` is still used. Consider re-adding a short comment
so the perf reason isn't lost for future maintainers.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
