Spenserrrr commented on code in PR #58751:
URL: https://github.com/apache/spark/pull/58751#discussion_r4031107934
##########
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:
Got it. I will add this back in later PRs. Thanks for pointing this out!
--
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]
