You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "wengh (via GitHub)" <gi...@apache.org> on 2025/02/13 20:30:38 UTC
[PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
wengh opened a new pull request, #49941:
URL: https://github.com/apache/spark/pull/49941
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Refactor `pyspark.sql.connect.conversion` to move `LocalDataToArrowConversion` and `ArrowTableToRowsConversion` into `pyspark.sql.conversion`.
The reason is that `pyspark.sql.connect.conversion` checks for Spark Connect dependencies such as `grpcio` and `pandas`, but `LocalDataToArrowConversion` and `ArrowTableToRowsConversion` don't need these dependencies.
`pyspark.sql.connect.conversion` still re-exports the two classes for backward compatibility.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Python Data Sources should work without Spark Connect dependencies but currently it imports `LocalDataToArrowConversion` and `ArrowTableToRowsConversion` from `pyspark.sql.connect.conversion` making it require unnecessary dependencies. This change moves these two classes to `pyspark.sql.conversion` so that Python Data Sources runs without Spark Connect dependencies.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Existing tests should make sure that the changes don't break anything.
TODO: Add new test to ensure that Python Data Sources can be used with minimal dependencies.
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960790988
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
Review Comment:
Changed to only import where it's used
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960586608
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
+from pyspark.errors import PySparkValueError
+from pyspark.sql.pandas.types import _dedup_names, _deduplicate_field_names, to_arrow_schema
+from pyspark.sql.types import (
+ ArrayType,
+ BinaryType,
+ DataType,
+ DecimalType,
+ MapType,
+ NullType,
+ Row,
+ StringType,
+ StructField,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ UserDefinedType,
+ VariantType,
+ VariantVal,
+ _create_row,
+)
+
+
+class LocalDataToArrowConversion:
+ """
+ Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
+ """
+
+ @staticmethod
+ def _need_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ ) -> bool:
+ if not nullable:
+ # always check the nullability
+ return True
+ elif isinstance(dataType, NullType):
+ # always check the nullability
+ return True
+ elif isinstance(dataType, StructType):
+ # Struct maybe rows, should convert to dict.
+ return True
+ elif isinstance(dataType, ArrayType):
+ return LocalDataToArrowConversion._need_converter(
+ dataType.elementType, dataType.containsNull
+ )
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
+ # since an Arrow Map requires a list of tuples.
+ return True
+ elif isinstance(dataType, BinaryType):
+ return True
+ elif isinstance(dataType, (TimestampType, TimestampNTZType)):
+ # Always truncate
+ return True
+ elif isinstance(dataType, DecimalType):
+ # Convert Decimal('NaN') to None
+ return True
+ elif isinstance(dataType, StringType):
+ # Coercion to StringType is allowed
+ return True
+ elif isinstance(dataType, UserDefinedType):
+ return True
+ elif isinstance(dataType, VariantType):
+ return True
+ else:
+ return False
+
+ @staticmethod
+ def _create_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ variants_as_dicts: bool = False, # some code paths may require python internal types
+ ) -> Callable:
+ assert dataType is not None and isinstance(dataType, DataType)
+ assert isinstance(nullable, bool)
+
+ if not LocalDataToArrowConversion._need_converter(dataType, nullable):
+ return lambda value: value
+
+ if isinstance(dataType, NullType):
+
+ def convert_null(value: Any) -> Any:
+ if value is not None:
+ raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
+ return None
+
+ return convert_null
+
+ elif isinstance(dataType, StructType):
+ field_names = dataType.fieldNames()
+ dedup_field_names = _dedup_names(dataType.names)
+
+ field_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts
+ )
+ for field in dataType.fields
+ ]
+
+ def convert_struct(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (tuple, dict)) or hasattr(
+ value, "__dict__"
+ ), f"{type(value)} {value}"
+
+ _dict = {}
+ if (
+ not isinstance(value, Row)
+ and not isinstance(value, tuple) # inherited namedtuple
+ and hasattr(value, "__dict__")
+ ):
+ value = value.__dict__
+ if isinstance(value, dict):
+ for i, field in enumerate(field_names):
+ _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
+ else:
+ if len(value) != len(field_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(field_names)),
+ "actual_length": str(len(value)),
+ },
+ )
+ for i in range(len(field_names)):
+ _dict[dedup_field_names[i]] = field_convs[i](value[i])
+
+ return _dict
+
+ return convert_struct
+
+ elif isinstance(dataType, ArrayType):
+ element_conv = LocalDataToArrowConversion._create_converter(
+ dataType.elementType, dataType.containsNull, variants_as_dicts
+ )
+
+ def convert_array(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (list, array.array))
+ return [element_conv(v) for v in value]
+
+ return convert_array
+
+ elif isinstance(dataType, MapType):
+ key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
+ value_conv = LocalDataToArrowConversion._create_converter(
+ dataType.valueType, dataType.valueContainsNull, variants_as_dicts
+ )
+
+ def convert_map(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, dict)
+
+ _tuples = []
+ for k, v in value.items():
+ _tuples.append((key_conv(k), value_conv(v)))
+
+ return _tuples
+
+ return convert_map
+
+ elif isinstance(dataType, BinaryType):
+
+ def convert_binary(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (bytes, bytearray))
+ return bytes(value)
+
+ return convert_binary
+
+ elif isinstance(dataType, TimestampType):
+
+ def convert_timestamp(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime)
+ return value.astimezone(datetime.timezone.utc)
+
+ return convert_timestamp
+
+ elif isinstance(dataType, TimestampNTZType):
+
+ def convert_timestamp_ntz(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime) and value.tzinfo is None
+ return value
+
+ return convert_timestamp_ntz
+
+ elif isinstance(dataType, DecimalType):
+
+ def convert_decimal(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, decimal.Decimal)
+ return None if value.is_nan() else value
+
+ return convert_decimal
+
+ elif isinstance(dataType, StringType):
+
+ def convert_string(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ if isinstance(value, bool):
+ # To match the PySpark which convert bool to string in
+ # the JVM side (python.EvaluatePython.makeFromJava)
+ return str(value).lower()
+ else:
+ return str(value)
+
+ return convert_string
+
+ elif isinstance(dataType, UserDefinedType):
+ udt: UserDefinedType = dataType
+
+ conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
+
+ def convert_udt(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ return conv(udt.serialize(value))
+
+ return convert_udt
+
+ elif isinstance(dataType, VariantType):
+
+ def convert_variant(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ elif (
+ isinstance(value, dict)
+ and all(key in value for key in ["value", "metadata"])
+ and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
+ and not variants_as_dicts
+ ):
+ return VariantVal(value["value"], value["metadata"])
+ elif isinstance(value, VariantVal) and variants_as_dicts:
+ return VariantType().toInternal(value)
+ else:
+ raise PySparkValueError(errorClass="MALFORMED_VARIANT")
+
+ return convert_variant
+
+ elif not nullable:
+
+ def convert_other(value: Any) -> Any:
+ if value is None:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return value
+
+ return convert_other
+ else:
+ return lambda value: value
+
+ @staticmethod
+ def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
+ assert isinstance(data, list) and len(data) > 0
+
+ assert schema is not None and isinstance(schema, StructType)
+
+ column_names = schema.fieldNames()
+
+ column_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts=True
+ )
+ for field in schema.fields
+ ]
+
+ pylist: List[List] = [[] for _ in range(len(column_names))]
+
+ for item in data:
+ if isinstance(item, VariantVal):
+ raise PySparkValueError("Rows cannot be of type VariantVal")
+ if (
+ not isinstance(item, Row)
+ and not isinstance(item, tuple) # inherited namedtuple
+ and hasattr(item, "__dict__")
+ ):
+ item = item.__dict__
+ if isinstance(item, dict):
+ for i, col in enumerate(column_names):
+ pylist[i].append(column_convs[i](item.get(col)))
+ else:
+ if len(item) != len(column_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(column_names)),
+ "actual_length": str(len(item)),
+ },
+ )
+
+ for i in range(len(column_names)):
+ pylist[i].append(column_convs[i](item[i]))
+
+ pa_schema = to_arrow_schema(
+ StructType(
+ [
+ StructField(
+ field.name, _deduplicate_field_names(field.dataType), field.nullable
+ )
+ for field in schema.fields
+ ]
+ ),
+ prefers_large_types=use_large_var_types,
+ )
+
+ return pa.Table.from_arrays(pylist, schema=pa_schema)
+
+
+class ArrowTableToRowsConversion:
+ """
+ Conversion from Arrow Table to Rows.
+ """
+
+ @staticmethod
+ def _need_converter(dataType: DataType) -> bool:
+ if isinstance(dataType, NullType):
+ return True
+ elif isinstance(dataType, StructType):
+ return True
+ elif isinstance(dataType, ArrayType):
+ return ArrowTableToRowsConversion._need_converter(dataType.elementType)
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
Review Comment:
Can we check where `_create_converter`is used and see if we can merge these two methods?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960798422
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
+from pyspark.errors import PySparkValueError
+from pyspark.sql.pandas.types import _dedup_names, _deduplicate_field_names, to_arrow_schema
+from pyspark.sql.types import (
+ ArrayType,
+ BinaryType,
+ DataType,
+ DecimalType,
+ MapType,
+ NullType,
+ Row,
+ StringType,
+ StructField,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ UserDefinedType,
+ VariantType,
+ VariantVal,
+ _create_row,
+)
+
+
+class LocalDataToArrowConversion:
+ """
+ Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
+ """
+
+ @staticmethod
+ def _need_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ ) -> bool:
+ if not nullable:
+ # always check the nullability
+ return True
+ elif isinstance(dataType, NullType):
+ # always check the nullability
+ return True
+ elif isinstance(dataType, StructType):
+ # Struct maybe rows, should convert to dict.
+ return True
+ elif isinstance(dataType, ArrayType):
+ return LocalDataToArrowConversion._need_converter(
+ dataType.elementType, dataType.containsNull
+ )
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
+ # since an Arrow Map requires a list of tuples.
+ return True
+ elif isinstance(dataType, BinaryType):
+ return True
+ elif isinstance(dataType, (TimestampType, TimestampNTZType)):
+ # Always truncate
+ return True
+ elif isinstance(dataType, DecimalType):
+ # Convert Decimal('NaN') to None
+ return True
+ elif isinstance(dataType, StringType):
+ # Coercion to StringType is allowed
+ return True
+ elif isinstance(dataType, UserDefinedType):
+ return True
+ elif isinstance(dataType, VariantType):
+ return True
+ else:
+ return False
+
+ @staticmethod
+ def _create_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ variants_as_dicts: bool = False, # some code paths may require python internal types
+ ) -> Callable:
+ assert dataType is not None and isinstance(dataType, DataType)
+ assert isinstance(nullable, bool)
+
+ if not LocalDataToArrowConversion._need_converter(dataType, nullable):
+ return lambda value: value
+
+ if isinstance(dataType, NullType):
+
+ def convert_null(value: Any) -> Any:
+ if value is not None:
+ raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
+ return None
+
+ return convert_null
+
+ elif isinstance(dataType, StructType):
+ field_names = dataType.fieldNames()
+ dedup_field_names = _dedup_names(dataType.names)
+
+ field_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts
+ )
+ for field in dataType.fields
+ ]
+
+ def convert_struct(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (tuple, dict)) or hasattr(
+ value, "__dict__"
+ ), f"{type(value)} {value}"
+
+ _dict = {}
+ if (
+ not isinstance(value, Row)
+ and not isinstance(value, tuple) # inherited namedtuple
+ and hasattr(value, "__dict__")
+ ):
+ value = value.__dict__
+ if isinstance(value, dict):
+ for i, field in enumerate(field_names):
+ _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
+ else:
+ if len(value) != len(field_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(field_names)),
+ "actual_length": str(len(value)),
+ },
+ )
+ for i in range(len(field_names)):
+ _dict[dedup_field_names[i]] = field_convs[i](value[i])
+
+ return _dict
+
+ return convert_struct
+
+ elif isinstance(dataType, ArrayType):
+ element_conv = LocalDataToArrowConversion._create_converter(
+ dataType.elementType, dataType.containsNull, variants_as_dicts
+ )
+
+ def convert_array(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (list, array.array))
+ return [element_conv(v) for v in value]
+
+ return convert_array
+
+ elif isinstance(dataType, MapType):
+ key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
+ value_conv = LocalDataToArrowConversion._create_converter(
+ dataType.valueType, dataType.valueContainsNull, variants_as_dicts
+ )
+
+ def convert_map(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, dict)
+
+ _tuples = []
+ for k, v in value.items():
+ _tuples.append((key_conv(k), value_conv(v)))
+
+ return _tuples
+
+ return convert_map
+
+ elif isinstance(dataType, BinaryType):
+
+ def convert_binary(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (bytes, bytearray))
+ return bytes(value)
+
+ return convert_binary
+
+ elif isinstance(dataType, TimestampType):
+
+ def convert_timestamp(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime)
+ return value.astimezone(datetime.timezone.utc)
+
+ return convert_timestamp
+
+ elif isinstance(dataType, TimestampNTZType):
+
+ def convert_timestamp_ntz(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime) and value.tzinfo is None
+ return value
+
+ return convert_timestamp_ntz
+
+ elif isinstance(dataType, DecimalType):
+
+ def convert_decimal(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, decimal.Decimal)
+ return None if value.is_nan() else value
+
+ return convert_decimal
+
+ elif isinstance(dataType, StringType):
+
+ def convert_string(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ if isinstance(value, bool):
+ # To match the PySpark which convert bool to string in
+ # the JVM side (python.EvaluatePython.makeFromJava)
+ return str(value).lower()
+ else:
+ return str(value)
+
+ return convert_string
+
+ elif isinstance(dataType, UserDefinedType):
+ udt: UserDefinedType = dataType
+
+ conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
+
+ def convert_udt(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ return conv(udt.serialize(value))
+
+ return convert_udt
+
+ elif isinstance(dataType, VariantType):
+
+ def convert_variant(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ elif (
+ isinstance(value, dict)
+ and all(key in value for key in ["value", "metadata"])
+ and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
+ and not variants_as_dicts
+ ):
+ return VariantVal(value["value"], value["metadata"])
+ elif isinstance(value, VariantVal) and variants_as_dicts:
+ return VariantType().toInternal(value)
+ else:
+ raise PySparkValueError(errorClass="MALFORMED_VARIANT")
+
+ return convert_variant
+
+ elif not nullable:
+
+ def convert_other(value: Any) -> Any:
+ if value is None:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return value
+
+ return convert_other
+ else:
+ return lambda value: value
+
+ @staticmethod
+ def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
+ assert isinstance(data, list) and len(data) > 0
+
+ assert schema is not None and isinstance(schema, StructType)
+
+ column_names = schema.fieldNames()
+
+ column_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts=True
+ )
+ for field in schema.fields
+ ]
+
+ pylist: List[List] = [[] for _ in range(len(column_names))]
+
+ for item in data:
+ if isinstance(item, VariantVal):
+ raise PySparkValueError("Rows cannot be of type VariantVal")
+ if (
+ not isinstance(item, Row)
+ and not isinstance(item, tuple) # inherited namedtuple
+ and hasattr(item, "__dict__")
+ ):
+ item = item.__dict__
+ if isinstance(item, dict):
+ for i, col in enumerate(column_names):
+ pylist[i].append(column_convs[i](item.get(col)))
+ else:
+ if len(item) != len(column_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(column_names)),
+ "actual_length": str(len(item)),
+ },
+ )
+
+ for i in range(len(column_names)):
+ pylist[i].append(column_convs[i](item[i]))
+
+ pa_schema = to_arrow_schema(
+ StructType(
+ [
+ StructField(
+ field.name, _deduplicate_field_names(field.dataType), field.nullable
+ )
+ for field in schema.fields
+ ]
+ ),
+ prefers_large_types=use_large_var_types,
+ )
+
+ return pa.Table.from_arrays(pylist, schema=pa_schema)
+
+
+class ArrowTableToRowsConversion:
+ """
+ Conversion from Arrow Table to Rows.
+ """
+
+ @staticmethod
+ def _need_converter(dataType: DataType) -> bool:
+ if isinstance(dataType, NullType):
+ return True
+ elif isinstance(dataType, StructType):
+ return True
+ elif isinstance(dataType, ArrayType):
+ return ArrowTableToRowsConversion._need_converter(dataType.elementType)
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
Review Comment:
That _create_converter turns dicts (StructType) into tuples of values. It has nothing to do with arrow conversion. Sorry for the confusion 🤦
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960750061
##########
python/pyspark/sql/connect/conversion.py:
##########
@@ -14,541 +14,31 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-from pyspark.errors import PySparkValueError
from pyspark.sql.connect.utils import check_dependencies
check_dependencies(__name__)
-import array
-import datetime
-import decimal
-import pyarrow as pa
+from typing import TYPE_CHECKING
-from pyspark.sql.types import (
- _create_row,
- Row,
- DataType,
- TimestampType,
- TimestampNTZType,
- MapType,
- StructField,
- StructType,
- ArrayType,
- BinaryType,
- NullType,
- DecimalType,
- StringType,
- UserDefinedType,
- VariantType,
- VariantVal,
+import pyspark.sql.connect.proto as pb2
+from pyspark.sql.conversion import (
+ ArrowTableToRowsConversion,
+ LocalDataToArrowConversion,
)
-
from pyspark.storagelevel import StorageLevel
-import pyspark.sql.connect.proto as pb2
-from pyspark.sql.pandas.types import to_arrow_schema, _dedup_names, _deduplicate_field_names
-
-from typing import Any, Callable, Sequence, List, TYPE_CHECKING
if TYPE_CHECKING:
from pyspark.sql.connect.dataframe import DataFrame
-class LocalDataToArrowConversion:
- """
- Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
- Currently, only :class:`SparkSession` in Spark Connect can use this class.
- """
-
- @staticmethod
- def _need_converter(
- dataType: DataType,
- nullable: bool = True,
- ) -> bool:
- if not nullable:
- # always check the nullability
- return True
- elif isinstance(dataType, NullType):
- # always check the nullability
- return True
- elif isinstance(dataType, StructType):
- # Struct maybe rows, should convert to dict.
- return True
- elif isinstance(dataType, ArrayType):
- return LocalDataToArrowConversion._need_converter(
- dataType.elementType, dataType.containsNull
- )
- elif isinstance(dataType, MapType):
- # Different from PySpark, here always needs conversion,
- # since an Arrow Map requires a list of tuples.
- return True
- elif isinstance(dataType, BinaryType):
- return True
- elif isinstance(dataType, (TimestampType, TimestampNTZType)):
- # Always truncate
- return True
- elif isinstance(dataType, DecimalType):
- # Convert Decimal('NaN') to None
- return True
- elif isinstance(dataType, StringType):
- # Coercion to StringType is allowed
- return True
- elif isinstance(dataType, UserDefinedType):
- return True
- elif isinstance(dataType, VariantType):
- return True
- else:
- return False
-
- @staticmethod
- def _create_converter(
- dataType: DataType,
- nullable: bool = True,
- variants_as_dicts: bool = False, # some code paths may require python internal types
- ) -> Callable:
- assert dataType is not None and isinstance(dataType, DataType)
- assert isinstance(nullable, bool)
-
- if not LocalDataToArrowConversion._need_converter(dataType, nullable):
- return lambda value: value
-
- if isinstance(dataType, NullType):
-
- def convert_null(value: Any) -> Any:
- if value is not None:
- raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
- return None
-
- return convert_null
-
- elif isinstance(dataType, StructType):
- field_names = dataType.fieldNames()
- dedup_field_names = _dedup_names(dataType.names)
-
- field_convs = [
- LocalDataToArrowConversion._create_converter(
- field.dataType, field.nullable, variants_as_dicts
- )
- for field in dataType.fields
- ]
-
- def convert_struct(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, (tuple, dict)) or hasattr(
- value, "__dict__"
- ), f"{type(value)} {value}"
-
- _dict = {}
- if (
- not isinstance(value, Row)
- and not isinstance(value, tuple) # inherited namedtuple
- and hasattr(value, "__dict__")
- ):
- value = value.__dict__
- if isinstance(value, dict):
- for i, field in enumerate(field_names):
- _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
- else:
- if len(value) != len(field_names):
- raise PySparkValueError(
- errorClass="AXIS_LENGTH_MISMATCH",
- messageParameters={
- "expected_length": str(len(field_names)),
- "actual_length": str(len(value)),
- },
- )
- for i in range(len(field_names)):
- _dict[dedup_field_names[i]] = field_convs[i](value[i])
-
- return _dict
-
- return convert_struct
-
- elif isinstance(dataType, ArrayType):
- element_conv = LocalDataToArrowConversion._create_converter(
- dataType.elementType, dataType.containsNull, variants_as_dicts
- )
-
- def convert_array(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, (list, array.array))
- return [element_conv(v) for v in value]
-
- return convert_array
-
- elif isinstance(dataType, MapType):
- key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
- value_conv = LocalDataToArrowConversion._create_converter(
- dataType.valueType, dataType.valueContainsNull, variants_as_dicts
- )
-
- def convert_map(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, dict)
-
- _tuples = []
- for k, v in value.items():
- _tuples.append((key_conv(k), value_conv(v)))
-
- return _tuples
-
- return convert_map
-
- elif isinstance(dataType, BinaryType):
-
- def convert_binary(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, (bytes, bytearray))
- return bytes(value)
-
- return convert_binary
-
- elif isinstance(dataType, TimestampType):
-
- def convert_timestamp(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, datetime.datetime)
- return value.astimezone(datetime.timezone.utc)
-
- return convert_timestamp
-
- elif isinstance(dataType, TimestampNTZType):
-
- def convert_timestamp_ntz(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, datetime.datetime) and value.tzinfo is None
- return value
-
- return convert_timestamp_ntz
-
- elif isinstance(dataType, DecimalType):
-
- def convert_decimal(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, decimal.Decimal)
- return None if value.is_nan() else value
-
- return convert_decimal
-
- elif isinstance(dataType, StringType):
-
- def convert_string(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- if isinstance(value, bool):
- # To match the PySpark which convert bool to string in
- # the JVM side (python.EvaluatePython.makeFromJava)
- return str(value).lower()
- else:
- return str(value)
-
- return convert_string
-
- elif isinstance(dataType, UserDefinedType):
- udt: UserDefinedType = dataType
-
- conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
-
- def convert_udt(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- return conv(udt.serialize(value))
-
- return convert_udt
-
- elif isinstance(dataType, VariantType):
-
- def convert_variant(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- elif (
- isinstance(value, dict)
- and all(key in value for key in ["value", "metadata"])
- and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
- and not variants_as_dicts
- ):
- return VariantVal(value["value"], value["metadata"])
- elif isinstance(value, VariantVal) and variants_as_dicts:
- return VariantType().toInternal(value)
- else:
- raise PySparkValueError(errorClass="MALFORMED_VARIANT")
-
- return convert_variant
-
- elif not nullable:
-
- def convert_other(value: Any) -> Any:
- if value is None:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return value
-
- return convert_other
- else:
- return lambda value: value
-
- @staticmethod
- def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
- assert isinstance(data, list) and len(data) > 0
-
- assert schema is not None and isinstance(schema, StructType)
-
- column_names = schema.fieldNames()
-
- column_convs = [
- LocalDataToArrowConversion._create_converter(
- field.dataType, field.nullable, variants_as_dicts=True
- )
- for field in schema.fields
- ]
-
- pylist: List[List] = [[] for _ in range(len(column_names))]
-
- for item in data:
- if isinstance(item, VariantVal):
- raise PySparkValueError("Rows cannot be of type VariantVal")
- if (
- not isinstance(item, Row)
- and not isinstance(item, tuple) # inherited namedtuple
- and hasattr(item, "__dict__")
- ):
- item = item.__dict__
- if isinstance(item, dict):
- for i, col in enumerate(column_names):
- pylist[i].append(column_convs[i](item.get(col)))
- else:
- if len(item) != len(column_names):
- raise PySparkValueError(
- errorClass="AXIS_LENGTH_MISMATCH",
- messageParameters={
- "expected_length": str(len(column_names)),
- "actual_length": str(len(item)),
- },
- )
-
- for i in range(len(column_names)):
- pylist[i].append(column_convs[i](item[i]))
-
- pa_schema = to_arrow_schema(
- StructType(
- [
- StructField(
- field.name, _deduplicate_field_names(field.dataType), field.nullable
- )
- for field in schema.fields
- ]
- ),
- prefers_large_types=use_large_var_types,
- )
-
- return pa.Table.from_arrays(pylist, schema=pa_schema)
-
-
-class ArrowTableToRowsConversion:
- """
- Conversion from Arrow Table to Rows.
- Currently, only :class:`DataFrame` in Spark Connect can use this class.
- """
-
- @staticmethod
- def _need_converter(dataType: DataType) -> bool:
- if isinstance(dataType, NullType):
- return True
- elif isinstance(dataType, StructType):
- return True
- elif isinstance(dataType, ArrayType):
- return ArrowTableToRowsConversion._need_converter(dataType.elementType)
- elif isinstance(dataType, MapType):
- # Different from PySpark, here always needs conversion,
- # since the input from Arrow is a list of tuples.
- return True
- elif isinstance(dataType, BinaryType):
- return True
- elif isinstance(dataType, (TimestampType, TimestampNTZType)):
- # Always remove the time zone info for now
- return True
- elif isinstance(dataType, UserDefinedType):
- return True
- elif isinstance(dataType, VariantType):
- return True
- else:
- return False
-
- @staticmethod
- def _create_converter(dataType: DataType) -> Callable:
- assert dataType is not None and isinstance(dataType, DataType)
-
- if not ArrowTableToRowsConversion._need_converter(dataType):
- return lambda value: value
-
- if isinstance(dataType, NullType):
- return lambda value: None
-
- elif isinstance(dataType, StructType):
- field_names = dataType.names
- dedup_field_names = _dedup_names(field_names)
-
- field_convs = [
- ArrowTableToRowsConversion._create_converter(f.dataType) for f in dataType.fields
- ]
-
- def convert_struct(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, dict)
-
- _values = [
- field_convs[i](value.get(name, None))
- for i, name in enumerate(dedup_field_names)
- ]
- return _create_row(field_names, _values)
-
- return convert_struct
-
- elif isinstance(dataType, ArrayType):
- element_conv = ArrowTableToRowsConversion._create_converter(dataType.elementType)
-
- def convert_array(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, list)
- return [element_conv(v) for v in value]
-
- return convert_array
-
- elif isinstance(dataType, MapType):
- key_conv = ArrowTableToRowsConversion._create_converter(dataType.keyType)
- value_conv = ArrowTableToRowsConversion._create_converter(dataType.valueType)
-
- def convert_map(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, list)
- assert all(isinstance(t, tuple) and len(t) == 2 for t in value)
- return dict((key_conv(t[0]), value_conv(t[1])) for t in value)
-
- return convert_map
-
- elif isinstance(dataType, BinaryType):
-
- def convert_binary(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, bytes)
- return bytearray(value)
-
- return convert_binary
-
- elif isinstance(dataType, TimestampType):
-
- def convert_timestample(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, datetime.datetime)
- return value.astimezone().replace(tzinfo=None)
-
- return convert_timestample
-
- elif isinstance(dataType, TimestampNTZType):
-
- def convert_timestample_ntz(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, datetime.datetime)
- return value
-
- return convert_timestample_ntz
-
- elif isinstance(dataType, UserDefinedType):
- udt: UserDefinedType = dataType
-
- conv = ArrowTableToRowsConversion._create_converter(udt.sqlType())
-
- def convert_udt(value: Any) -> Any:
- if value is None:
- return None
- else:
- return udt.deserialize(conv(value))
-
- return convert_udt
-
- elif isinstance(dataType, VariantType):
-
- def convert_variant(value: Any) -> Any:
- if value is None:
- return None
- elif (
- isinstance(value, dict)
- and all(key in value for key in ["value", "metadata"])
- and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
- ):
- return VariantVal(value["value"], value["metadata"])
- else:
- raise PySparkValueError(errorClass="MALFORMED_VARIANT")
-
- return convert_variant
-
- else:
- return lambda value: value
-
- @staticmethod
- def convert(table: "pa.Table", schema: StructType) -> List[Row]:
- assert isinstance(table, pa.Table)
-
- assert schema is not None and isinstance(schema, StructType)
-
- field_converters = [
- ArrowTableToRowsConversion._create_converter(f.dataType) for f in schema.fields
- ]
-
- columnar_data = [column.to_pylist() for column in table.columns]
-
- rows: List[Row] = []
- for i in range(0, table.num_rows):
- values = [field_converters[j](columnar_data[j][i]) for j in range(table.num_columns)]
- rows.append(_create_row(fields=schema.fieldNames(), values=values))
- return rows
+__all__ = [
Review Comment:
If we remove this it will break forks that import from `pyspark.sql.connect.conversion` even if there's no merge conflict. Is that ok?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960735174
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
Review Comment:
You will have to import inplace where it's used, e.g., within the method.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on PR #49941:
URL: https://github.com/apache/spark/pull/49941#issuecomment-2660361211
cc @HyukjinKwon
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960286586
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
+from pyspark.errors import PySparkValueError
+from pyspark.sql.pandas.types import _dedup_names, _deduplicate_field_names, to_arrow_schema
+from pyspark.sql.types import (
+ ArrayType,
+ BinaryType,
+ DataType,
+ DecimalType,
+ MapType,
+ NullType,
+ Row,
+ StringType,
+ StructField,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ UserDefinedType,
+ VariantType,
+ VariantVal,
+ _create_row,
+)
+
+
+class LocalDataToArrowConversion:
+ """
+ Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
+ """
+
+ @staticmethod
+ def _need_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ ) -> bool:
+ if not nullable:
+ # always check the nullability
+ return True
+ elif isinstance(dataType, NullType):
+ # always check the nullability
+ return True
+ elif isinstance(dataType, StructType):
+ # Struct maybe rows, should convert to dict.
+ return True
+ elif isinstance(dataType, ArrayType):
+ return LocalDataToArrowConversion._need_converter(
+ dataType.elementType, dataType.containsNull
+ )
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
+ # since an Arrow Map requires a list of tuples.
+ return True
+ elif isinstance(dataType, BinaryType):
+ return True
+ elif isinstance(dataType, (TimestampType, TimestampNTZType)):
+ # Always truncate
+ return True
+ elif isinstance(dataType, DecimalType):
+ # Convert Decimal('NaN') to None
+ return True
+ elif isinstance(dataType, StringType):
+ # Coercion to StringType is allowed
+ return True
+ elif isinstance(dataType, UserDefinedType):
+ return True
+ elif isinstance(dataType, VariantType):
+ return True
+ else:
+ return False
+
+ @staticmethod
+ def _create_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ variants_as_dicts: bool = False, # some code paths may require python internal types
+ ) -> Callable:
+ assert dataType is not None and isinstance(dataType, DataType)
+ assert isinstance(nullable, bool)
+
+ if not LocalDataToArrowConversion._need_converter(dataType, nullable):
+ return lambda value: value
+
+ if isinstance(dataType, NullType):
+
+ def convert_null(value: Any) -> Any:
+ if value is not None:
+ raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
+ return None
+
+ return convert_null
+
+ elif isinstance(dataType, StructType):
+ field_names = dataType.fieldNames()
+ dedup_field_names = _dedup_names(dataType.names)
+
+ field_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts
+ )
+ for field in dataType.fields
+ ]
+
+ def convert_struct(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (tuple, dict)) or hasattr(
+ value, "__dict__"
+ ), f"{type(value)} {value}"
+
+ _dict = {}
+ if (
+ not isinstance(value, Row)
+ and not isinstance(value, tuple) # inherited namedtuple
+ and hasattr(value, "__dict__")
+ ):
+ value = value.__dict__
+ if isinstance(value, dict):
+ for i, field in enumerate(field_names):
+ _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
+ else:
+ if len(value) != len(field_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(field_names)),
+ "actual_length": str(len(value)),
+ },
+ )
+ for i in range(len(field_names)):
+ _dict[dedup_field_names[i]] = field_convs[i](value[i])
+
+ return _dict
+
+ return convert_struct
+
+ elif isinstance(dataType, ArrayType):
+ element_conv = LocalDataToArrowConversion._create_converter(
+ dataType.elementType, dataType.containsNull, variants_as_dicts
+ )
+
+ def convert_array(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (list, array.array))
+ return [element_conv(v) for v in value]
+
+ return convert_array
+
+ elif isinstance(dataType, MapType):
+ key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
+ value_conv = LocalDataToArrowConversion._create_converter(
+ dataType.valueType, dataType.valueContainsNull, variants_as_dicts
+ )
+
+ def convert_map(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, dict)
+
+ _tuples = []
+ for k, v in value.items():
+ _tuples.append((key_conv(k), value_conv(v)))
+
+ return _tuples
+
+ return convert_map
+
+ elif isinstance(dataType, BinaryType):
+
+ def convert_binary(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (bytes, bytearray))
+ return bytes(value)
+
+ return convert_binary
+
+ elif isinstance(dataType, TimestampType):
+
+ def convert_timestamp(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime)
+ return value.astimezone(datetime.timezone.utc)
+
+ return convert_timestamp
+
+ elif isinstance(dataType, TimestampNTZType):
+
+ def convert_timestamp_ntz(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime) and value.tzinfo is None
+ return value
+
+ return convert_timestamp_ntz
+
+ elif isinstance(dataType, DecimalType):
+
+ def convert_decimal(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, decimal.Decimal)
+ return None if value.is_nan() else value
+
+ return convert_decimal
+
+ elif isinstance(dataType, StringType):
+
+ def convert_string(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ if isinstance(value, bool):
+ # To match the PySpark which convert bool to string in
+ # the JVM side (python.EvaluatePython.makeFromJava)
+ return str(value).lower()
+ else:
+ return str(value)
+
+ return convert_string
+
+ elif isinstance(dataType, UserDefinedType):
+ udt: UserDefinedType = dataType
+
+ conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
+
+ def convert_udt(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ return conv(udt.serialize(value))
+
+ return convert_udt
+
+ elif isinstance(dataType, VariantType):
+
+ def convert_variant(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ elif (
+ isinstance(value, dict)
+ and all(key in value for key in ["value", "metadata"])
+ and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
+ and not variants_as_dicts
+ ):
+ return VariantVal(value["value"], value["metadata"])
+ elif isinstance(value, VariantVal) and variants_as_dicts:
+ return VariantType().toInternal(value)
+ else:
+ raise PySparkValueError(errorClass="MALFORMED_VARIANT")
+
+ return convert_variant
+
+ elif not nullable:
+
+ def convert_other(value: Any) -> Any:
+ if value is None:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return value
+
+ return convert_other
+ else:
+ return lambda value: value
+
+ @staticmethod
+ def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
+ assert isinstance(data, list) and len(data) > 0
+
+ assert schema is not None and isinstance(schema, StructType)
+
+ column_names = schema.fieldNames()
+
+ column_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts=True
+ )
+ for field in schema.fields
+ ]
+
+ pylist: List[List] = [[] for _ in range(len(column_names))]
+
+ for item in data:
+ if isinstance(item, VariantVal):
+ raise PySparkValueError("Rows cannot be of type VariantVal")
+ if (
+ not isinstance(item, Row)
+ and not isinstance(item, tuple) # inherited namedtuple
+ and hasattr(item, "__dict__")
+ ):
+ item = item.__dict__
+ if isinstance(item, dict):
+ for i, col in enumerate(column_names):
+ pylist[i].append(column_convs[i](item.get(col)))
+ else:
+ if len(item) != len(column_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(column_names)),
+ "actual_length": str(len(item)),
+ },
+ )
+
+ for i in range(len(column_names)):
+ pylist[i].append(column_convs[i](item[i]))
+
+ pa_schema = to_arrow_schema(
+ StructType(
+ [
+ StructField(
+ field.name, _deduplicate_field_names(field.dataType), field.nullable
+ )
+ for field in schema.fields
+ ]
+ ),
+ prefers_large_types=use_large_var_types,
+ )
+
+ return pa.Table.from_arrays(pylist, schema=pa_schema)
+
+
+class ArrowTableToRowsConversion:
+ """
+ Conversion from Arrow Table to Rows.
+ """
+
+ @staticmethod
+ def _need_converter(dataType: DataType) -> bool:
+ if isinstance(dataType, NullType):
+ return True
+ elif isinstance(dataType, StructType):
+ return True
+ elif isinstance(dataType, ArrayType):
+ return ArrowTableToRowsConversion._need_converter(dataType.elementType)
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
Review Comment:
Are you referring to [`pyspark.sql.types._create_converter`](https://src.dev.databricks.com/databricks-eng/runtime@72929e1baf2ed733e300437b784ca4f2e32f3d7a/-/blob/python/pyspark/sql/types.py?L2567-2617)? Seems like there are some differences, for example `ArrowTableToRowsConversion` handles UDT and variant but the one in `types.py` doesn't.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1967091437
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
+from pyspark.errors import PySparkValueError
+from pyspark.sql.pandas.types import _dedup_names, _deduplicate_field_names, to_arrow_schema
+from pyspark.sql.types import (
+ ArrayType,
+ BinaryType,
+ DataType,
+ DecimalType,
+ MapType,
+ NullType,
+ Row,
+ StringType,
+ StructField,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ UserDefinedType,
+ VariantType,
+ VariantVal,
+ _create_row,
+)
+
+
+class LocalDataToArrowConversion:
+ """
+ Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
+ """
+
+ @staticmethod
+ def _need_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ ) -> bool:
+ if not nullable:
+ # always check the nullability
+ return True
+ elif isinstance(dataType, NullType):
+ # always check the nullability
+ return True
+ elif isinstance(dataType, StructType):
+ # Struct maybe rows, should convert to dict.
+ return True
+ elif isinstance(dataType, ArrayType):
+ return LocalDataToArrowConversion._need_converter(
+ dataType.elementType, dataType.containsNull
+ )
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
+ # since an Arrow Map requires a list of tuples.
+ return True
+ elif isinstance(dataType, BinaryType):
+ return True
+ elif isinstance(dataType, (TimestampType, TimestampNTZType)):
+ # Always truncate
+ return True
+ elif isinstance(dataType, DecimalType):
+ # Convert Decimal('NaN') to None
+ return True
+ elif isinstance(dataType, StringType):
+ # Coercion to StringType is allowed
+ return True
+ elif isinstance(dataType, UserDefinedType):
+ return True
+ elif isinstance(dataType, VariantType):
+ return True
+ else:
+ return False
+
+ @staticmethod
+ def _create_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ variants_as_dicts: bool = False, # some code paths may require python internal types
+ ) -> Callable:
+ assert dataType is not None and isinstance(dataType, DataType)
+ assert isinstance(nullable, bool)
+
+ if not LocalDataToArrowConversion._need_converter(dataType, nullable):
+ return lambda value: value
+
+ if isinstance(dataType, NullType):
+
+ def convert_null(value: Any) -> Any:
+ if value is not None:
+ raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
+ return None
+
+ return convert_null
+
+ elif isinstance(dataType, StructType):
+ field_names = dataType.fieldNames()
+ dedup_field_names = _dedup_names(dataType.names)
+
+ field_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts
+ )
+ for field in dataType.fields
+ ]
+
+ def convert_struct(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (tuple, dict)) or hasattr(
+ value, "__dict__"
+ ), f"{type(value)} {value}"
+
+ _dict = {}
+ if (
+ not isinstance(value, Row)
+ and not isinstance(value, tuple) # inherited namedtuple
+ and hasattr(value, "__dict__")
+ ):
+ value = value.__dict__
+ if isinstance(value, dict):
+ for i, field in enumerate(field_names):
+ _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
+ else:
+ if len(value) != len(field_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(field_names)),
+ "actual_length": str(len(value)),
+ },
+ )
+ for i in range(len(field_names)):
+ _dict[dedup_field_names[i]] = field_convs[i](value[i])
+
+ return _dict
+
+ return convert_struct
+
+ elif isinstance(dataType, ArrayType):
+ element_conv = LocalDataToArrowConversion._create_converter(
+ dataType.elementType, dataType.containsNull, variants_as_dicts
+ )
+
+ def convert_array(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (list, array.array))
+ return [element_conv(v) for v in value]
+
+ return convert_array
+
+ elif isinstance(dataType, MapType):
+ key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
+ value_conv = LocalDataToArrowConversion._create_converter(
+ dataType.valueType, dataType.valueContainsNull, variants_as_dicts
+ )
+
+ def convert_map(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, dict)
+
+ _tuples = []
+ for k, v in value.items():
+ _tuples.append((key_conv(k), value_conv(v)))
+
+ return _tuples
+
+ return convert_map
+
+ elif isinstance(dataType, BinaryType):
+
+ def convert_binary(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (bytes, bytearray))
+ return bytes(value)
+
+ return convert_binary
+
+ elif isinstance(dataType, TimestampType):
+
+ def convert_timestamp(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime)
+ return value.astimezone(datetime.timezone.utc)
+
+ return convert_timestamp
+
+ elif isinstance(dataType, TimestampNTZType):
+
+ def convert_timestamp_ntz(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime) and value.tzinfo is None
+ return value
+
+ return convert_timestamp_ntz
+
+ elif isinstance(dataType, DecimalType):
+
+ def convert_decimal(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, decimal.Decimal)
+ return None if value.is_nan() else value
+
+ return convert_decimal
+
+ elif isinstance(dataType, StringType):
+
+ def convert_string(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ if isinstance(value, bool):
+ # To match the PySpark which convert bool to string in
+ # the JVM side (python.EvaluatePython.makeFromJava)
+ return str(value).lower()
+ else:
+ return str(value)
+
+ return convert_string
+
+ elif isinstance(dataType, UserDefinedType):
+ udt: UserDefinedType = dataType
+
+ conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
+
+ def convert_udt(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ return conv(udt.serialize(value))
+
+ return convert_udt
+
+ elif isinstance(dataType, VariantType):
+
+ def convert_variant(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ elif (
+ isinstance(value, dict)
+ and all(key in value for key in ["value", "metadata"])
+ and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
+ and not variants_as_dicts
+ ):
+ return VariantVal(value["value"], value["metadata"])
+ elif isinstance(value, VariantVal) and variants_as_dicts:
+ return VariantType().toInternal(value)
+ else:
+ raise PySparkValueError(errorClass="MALFORMED_VARIANT")
+
+ return convert_variant
+
+ elif not nullable:
+
+ def convert_other(value: Any) -> Any:
+ if value is None:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return value
+
+ return convert_other
+ else:
+ return lambda value: value
+
+ @staticmethod
+ def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
+ assert isinstance(data, list) and len(data) > 0
+
+ assert schema is not None and isinstance(schema, StructType)
+
+ column_names = schema.fieldNames()
+
+ column_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts=True
+ )
+ for field in schema.fields
+ ]
+
+ pylist: List[List] = [[] for _ in range(len(column_names))]
+
+ for item in data:
+ if isinstance(item, VariantVal):
+ raise PySparkValueError("Rows cannot be of type VariantVal")
+ if (
+ not isinstance(item, Row)
+ and not isinstance(item, tuple) # inherited namedtuple
+ and hasattr(item, "__dict__")
+ ):
+ item = item.__dict__
+ if isinstance(item, dict):
+ for i, col in enumerate(column_names):
+ pylist[i].append(column_convs[i](item.get(col)))
+ else:
+ if len(item) != len(column_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(column_names)),
+ "actual_length": str(len(item)),
+ },
+ )
+
+ for i in range(len(column_names)):
+ pylist[i].append(column_convs[i](item[i]))
+
+ pa_schema = to_arrow_schema(
+ StructType(
+ [
+ StructField(
+ field.name, _deduplicate_field_names(field.dataType), field.nullable
+ )
+ for field in schema.fields
+ ]
+ ),
+ prefers_large_types=use_large_var_types,
+ )
+
+ return pa.Table.from_arrays(pylist, schema=pa_schema)
+
+
+class ArrowTableToRowsConversion:
+ """
+ Conversion from Arrow Table to Rows.
+ """
+
+ @staticmethod
+ def _need_converter(dataType: DataType) -> bool:
+ if isinstance(dataType, NullType):
+ return True
+ elif isinstance(dataType, StructType):
+ return True
+ elif isinstance(dataType, ArrayType):
+ return ArrowTableToRowsConversion._need_converter(dataType.elementType)
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
Review Comment:
actually, I meant `python/pyspark/sql/pandas/conversion.py`. We can investigate it later.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960286586
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
+from pyspark.errors import PySparkValueError
+from pyspark.sql.pandas.types import _dedup_names, _deduplicate_field_names, to_arrow_schema
+from pyspark.sql.types import (
+ ArrayType,
+ BinaryType,
+ DataType,
+ DecimalType,
+ MapType,
+ NullType,
+ Row,
+ StringType,
+ StructField,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ UserDefinedType,
+ VariantType,
+ VariantVal,
+ _create_row,
+)
+
+
+class LocalDataToArrowConversion:
+ """
+ Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
+ """
+
+ @staticmethod
+ def _need_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ ) -> bool:
+ if not nullable:
+ # always check the nullability
+ return True
+ elif isinstance(dataType, NullType):
+ # always check the nullability
+ return True
+ elif isinstance(dataType, StructType):
+ # Struct maybe rows, should convert to dict.
+ return True
+ elif isinstance(dataType, ArrayType):
+ return LocalDataToArrowConversion._need_converter(
+ dataType.elementType, dataType.containsNull
+ )
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
+ # since an Arrow Map requires a list of tuples.
+ return True
+ elif isinstance(dataType, BinaryType):
+ return True
+ elif isinstance(dataType, (TimestampType, TimestampNTZType)):
+ # Always truncate
+ return True
+ elif isinstance(dataType, DecimalType):
+ # Convert Decimal('NaN') to None
+ return True
+ elif isinstance(dataType, StringType):
+ # Coercion to StringType is allowed
+ return True
+ elif isinstance(dataType, UserDefinedType):
+ return True
+ elif isinstance(dataType, VariantType):
+ return True
+ else:
+ return False
+
+ @staticmethod
+ def _create_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ variants_as_dicts: bool = False, # some code paths may require python internal types
+ ) -> Callable:
+ assert dataType is not None and isinstance(dataType, DataType)
+ assert isinstance(nullable, bool)
+
+ if not LocalDataToArrowConversion._need_converter(dataType, nullable):
+ return lambda value: value
+
+ if isinstance(dataType, NullType):
+
+ def convert_null(value: Any) -> Any:
+ if value is not None:
+ raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
+ return None
+
+ return convert_null
+
+ elif isinstance(dataType, StructType):
+ field_names = dataType.fieldNames()
+ dedup_field_names = _dedup_names(dataType.names)
+
+ field_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts
+ )
+ for field in dataType.fields
+ ]
+
+ def convert_struct(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (tuple, dict)) or hasattr(
+ value, "__dict__"
+ ), f"{type(value)} {value}"
+
+ _dict = {}
+ if (
+ not isinstance(value, Row)
+ and not isinstance(value, tuple) # inherited namedtuple
+ and hasattr(value, "__dict__")
+ ):
+ value = value.__dict__
+ if isinstance(value, dict):
+ for i, field in enumerate(field_names):
+ _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
+ else:
+ if len(value) != len(field_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(field_names)),
+ "actual_length": str(len(value)),
+ },
+ )
+ for i in range(len(field_names)):
+ _dict[dedup_field_names[i]] = field_convs[i](value[i])
+
+ return _dict
+
+ return convert_struct
+
+ elif isinstance(dataType, ArrayType):
+ element_conv = LocalDataToArrowConversion._create_converter(
+ dataType.elementType, dataType.containsNull, variants_as_dicts
+ )
+
+ def convert_array(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (list, array.array))
+ return [element_conv(v) for v in value]
+
+ return convert_array
+
+ elif isinstance(dataType, MapType):
+ key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
+ value_conv = LocalDataToArrowConversion._create_converter(
+ dataType.valueType, dataType.valueContainsNull, variants_as_dicts
+ )
+
+ def convert_map(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, dict)
+
+ _tuples = []
+ for k, v in value.items():
+ _tuples.append((key_conv(k), value_conv(v)))
+
+ return _tuples
+
+ return convert_map
+
+ elif isinstance(dataType, BinaryType):
+
+ def convert_binary(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (bytes, bytearray))
+ return bytes(value)
+
+ return convert_binary
+
+ elif isinstance(dataType, TimestampType):
+
+ def convert_timestamp(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime)
+ return value.astimezone(datetime.timezone.utc)
+
+ return convert_timestamp
+
+ elif isinstance(dataType, TimestampNTZType):
+
+ def convert_timestamp_ntz(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime) and value.tzinfo is None
+ return value
+
+ return convert_timestamp_ntz
+
+ elif isinstance(dataType, DecimalType):
+
+ def convert_decimal(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, decimal.Decimal)
+ return None if value.is_nan() else value
+
+ return convert_decimal
+
+ elif isinstance(dataType, StringType):
+
+ def convert_string(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ if isinstance(value, bool):
+ # To match the PySpark which convert bool to string in
+ # the JVM side (python.EvaluatePython.makeFromJava)
+ return str(value).lower()
+ else:
+ return str(value)
+
+ return convert_string
+
+ elif isinstance(dataType, UserDefinedType):
+ udt: UserDefinedType = dataType
+
+ conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
+
+ def convert_udt(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ return conv(udt.serialize(value))
+
+ return convert_udt
+
+ elif isinstance(dataType, VariantType):
+
+ def convert_variant(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ elif (
+ isinstance(value, dict)
+ and all(key in value for key in ["value", "metadata"])
+ and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
+ and not variants_as_dicts
+ ):
+ return VariantVal(value["value"], value["metadata"])
+ elif isinstance(value, VariantVal) and variants_as_dicts:
+ return VariantType().toInternal(value)
+ else:
+ raise PySparkValueError(errorClass="MALFORMED_VARIANT")
+
+ return convert_variant
+
+ elif not nullable:
+
+ def convert_other(value: Any) -> Any:
+ if value is None:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return value
+
+ return convert_other
+ else:
+ return lambda value: value
+
+ @staticmethod
+ def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
+ assert isinstance(data, list) and len(data) > 0
+
+ assert schema is not None and isinstance(schema, StructType)
+
+ column_names = schema.fieldNames()
+
+ column_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts=True
+ )
+ for field in schema.fields
+ ]
+
+ pylist: List[List] = [[] for _ in range(len(column_names))]
+
+ for item in data:
+ if isinstance(item, VariantVal):
+ raise PySparkValueError("Rows cannot be of type VariantVal")
+ if (
+ not isinstance(item, Row)
+ and not isinstance(item, tuple) # inherited namedtuple
+ and hasattr(item, "__dict__")
+ ):
+ item = item.__dict__
+ if isinstance(item, dict):
+ for i, col in enumerate(column_names):
+ pylist[i].append(column_convs[i](item.get(col)))
+ else:
+ if len(item) != len(column_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(column_names)),
+ "actual_length": str(len(item)),
+ },
+ )
+
+ for i in range(len(column_names)):
+ pylist[i].append(column_convs[i](item[i]))
+
+ pa_schema = to_arrow_schema(
+ StructType(
+ [
+ StructField(
+ field.name, _deduplicate_field_names(field.dataType), field.nullable
+ )
+ for field in schema.fields
+ ]
+ ),
+ prefers_large_types=use_large_var_types,
+ )
+
+ return pa.Table.from_arrays(pylist, schema=pa_schema)
+
+
+class ArrowTableToRowsConversion:
+ """
+ Conversion from Arrow Table to Rows.
+ """
+
+ @staticmethod
+ def _need_converter(dataType: DataType) -> bool:
+ if isinstance(dataType, NullType):
+ return True
+ elif isinstance(dataType, StructType):
+ return True
+ elif isinstance(dataType, ArrayType):
+ return ArrowTableToRowsConversion._need_converter(dataType.elementType)
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
Review Comment:
Are you referring to [`pyspark.sql.types._create_converter`](https://github.com/databricks-eng/runtime/blob/5f41ce66c67a4cfa972ab6a3b91adb65bc61f83b/python/pyspark/sql/types.py#L2567-L2617)? Seems like there are some differences, for example `ArrowTableToRowsConversion` handles UDT and variant but the one in `types.py` doesn't.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960286586
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
+from pyspark.errors import PySparkValueError
+from pyspark.sql.pandas.types import _dedup_names, _deduplicate_field_names, to_arrow_schema
+from pyspark.sql.types import (
+ ArrayType,
+ BinaryType,
+ DataType,
+ DecimalType,
+ MapType,
+ NullType,
+ Row,
+ StringType,
+ StructField,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ UserDefinedType,
+ VariantType,
+ VariantVal,
+ _create_row,
+)
+
+
+class LocalDataToArrowConversion:
+ """
+ Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
+ """
+
+ @staticmethod
+ def _need_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ ) -> bool:
+ if not nullable:
+ # always check the nullability
+ return True
+ elif isinstance(dataType, NullType):
+ # always check the nullability
+ return True
+ elif isinstance(dataType, StructType):
+ # Struct maybe rows, should convert to dict.
+ return True
+ elif isinstance(dataType, ArrayType):
+ return LocalDataToArrowConversion._need_converter(
+ dataType.elementType, dataType.containsNull
+ )
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
+ # since an Arrow Map requires a list of tuples.
+ return True
+ elif isinstance(dataType, BinaryType):
+ return True
+ elif isinstance(dataType, (TimestampType, TimestampNTZType)):
+ # Always truncate
+ return True
+ elif isinstance(dataType, DecimalType):
+ # Convert Decimal('NaN') to None
+ return True
+ elif isinstance(dataType, StringType):
+ # Coercion to StringType is allowed
+ return True
+ elif isinstance(dataType, UserDefinedType):
+ return True
+ elif isinstance(dataType, VariantType):
+ return True
+ else:
+ return False
+
+ @staticmethod
+ def _create_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ variants_as_dicts: bool = False, # some code paths may require python internal types
+ ) -> Callable:
+ assert dataType is not None and isinstance(dataType, DataType)
+ assert isinstance(nullable, bool)
+
+ if not LocalDataToArrowConversion._need_converter(dataType, nullable):
+ return lambda value: value
+
+ if isinstance(dataType, NullType):
+
+ def convert_null(value: Any) -> Any:
+ if value is not None:
+ raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
+ return None
+
+ return convert_null
+
+ elif isinstance(dataType, StructType):
+ field_names = dataType.fieldNames()
+ dedup_field_names = _dedup_names(dataType.names)
+
+ field_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts
+ )
+ for field in dataType.fields
+ ]
+
+ def convert_struct(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (tuple, dict)) or hasattr(
+ value, "__dict__"
+ ), f"{type(value)} {value}"
+
+ _dict = {}
+ if (
+ not isinstance(value, Row)
+ and not isinstance(value, tuple) # inherited namedtuple
+ and hasattr(value, "__dict__")
+ ):
+ value = value.__dict__
+ if isinstance(value, dict):
+ for i, field in enumerate(field_names):
+ _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
+ else:
+ if len(value) != len(field_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(field_names)),
+ "actual_length": str(len(value)),
+ },
+ )
+ for i in range(len(field_names)):
+ _dict[dedup_field_names[i]] = field_convs[i](value[i])
+
+ return _dict
+
+ return convert_struct
+
+ elif isinstance(dataType, ArrayType):
+ element_conv = LocalDataToArrowConversion._create_converter(
+ dataType.elementType, dataType.containsNull, variants_as_dicts
+ )
+
+ def convert_array(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (list, array.array))
+ return [element_conv(v) for v in value]
+
+ return convert_array
+
+ elif isinstance(dataType, MapType):
+ key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
+ value_conv = LocalDataToArrowConversion._create_converter(
+ dataType.valueType, dataType.valueContainsNull, variants_as_dicts
+ )
+
+ def convert_map(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, dict)
+
+ _tuples = []
+ for k, v in value.items():
+ _tuples.append((key_conv(k), value_conv(v)))
+
+ return _tuples
+
+ return convert_map
+
+ elif isinstance(dataType, BinaryType):
+
+ def convert_binary(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (bytes, bytearray))
+ return bytes(value)
+
+ return convert_binary
+
+ elif isinstance(dataType, TimestampType):
+
+ def convert_timestamp(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime)
+ return value.astimezone(datetime.timezone.utc)
+
+ return convert_timestamp
+
+ elif isinstance(dataType, TimestampNTZType):
+
+ def convert_timestamp_ntz(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime) and value.tzinfo is None
+ return value
+
+ return convert_timestamp_ntz
+
+ elif isinstance(dataType, DecimalType):
+
+ def convert_decimal(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, decimal.Decimal)
+ return None if value.is_nan() else value
+
+ return convert_decimal
+
+ elif isinstance(dataType, StringType):
+
+ def convert_string(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ if isinstance(value, bool):
+ # To match the PySpark which convert bool to string in
+ # the JVM side (python.EvaluatePython.makeFromJava)
+ return str(value).lower()
+ else:
+ return str(value)
+
+ return convert_string
+
+ elif isinstance(dataType, UserDefinedType):
+ udt: UserDefinedType = dataType
+
+ conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
+
+ def convert_udt(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ return conv(udt.serialize(value))
+
+ return convert_udt
+
+ elif isinstance(dataType, VariantType):
+
+ def convert_variant(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ elif (
+ isinstance(value, dict)
+ and all(key in value for key in ["value", "metadata"])
+ and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
+ and not variants_as_dicts
+ ):
+ return VariantVal(value["value"], value["metadata"])
+ elif isinstance(value, VariantVal) and variants_as_dicts:
+ return VariantType().toInternal(value)
+ else:
+ raise PySparkValueError(errorClass="MALFORMED_VARIANT")
+
+ return convert_variant
+
+ elif not nullable:
+
+ def convert_other(value: Any) -> Any:
+ if value is None:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return value
+
+ return convert_other
+ else:
+ return lambda value: value
+
+ @staticmethod
+ def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
+ assert isinstance(data, list) and len(data) > 0
+
+ assert schema is not None and isinstance(schema, StructType)
+
+ column_names = schema.fieldNames()
+
+ column_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts=True
+ )
+ for field in schema.fields
+ ]
+
+ pylist: List[List] = [[] for _ in range(len(column_names))]
+
+ for item in data:
+ if isinstance(item, VariantVal):
+ raise PySparkValueError("Rows cannot be of type VariantVal")
+ if (
+ not isinstance(item, Row)
+ and not isinstance(item, tuple) # inherited namedtuple
+ and hasattr(item, "__dict__")
+ ):
+ item = item.__dict__
+ if isinstance(item, dict):
+ for i, col in enumerate(column_names):
+ pylist[i].append(column_convs[i](item.get(col)))
+ else:
+ if len(item) != len(column_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(column_names)),
+ "actual_length": str(len(item)),
+ },
+ )
+
+ for i in range(len(column_names)):
+ pylist[i].append(column_convs[i](item[i]))
+
+ pa_schema = to_arrow_schema(
+ StructType(
+ [
+ StructField(
+ field.name, _deduplicate_field_names(field.dataType), field.nullable
+ )
+ for field in schema.fields
+ ]
+ ),
+ prefers_large_types=use_large_var_types,
+ )
+
+ return pa.Table.from_arrays(pylist, schema=pa_schema)
+
+
+class ArrowTableToRowsConversion:
+ """
+ Conversion from Arrow Table to Rows.
+ """
+
+ @staticmethod
+ def _need_converter(dataType: DataType) -> bool:
+ if isinstance(dataType, NullType):
+ return True
+ elif isinstance(dataType, StructType):
+ return True
+ elif isinstance(dataType, ArrayType):
+ return ArrowTableToRowsConversion._need_converter(dataType.elementType)
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
Review Comment:
Are you referring to [`pyspark.sql.types._create_converter`](https://github.com/apache/spark/blob/aa37f89811878d9993962c75b73025bcadcbb6f5/python/pyspark/sql/types.py#L2513-L2563)? Seems like there are some differences, for example `ArrowTableToRowsConversion` handles UDT and variant but the one in `types.py` doesn't.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960734709
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
Review Comment:
Still Arrow is an optional package. Without installing it, it fails when you import PySpark without Arrow.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #49941: [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect
URL: https://github.com/apache/spark/pull/49941
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #49941:
URL: https://github.com/apache/spark/pull/49941#issuecomment-2686706677
Merged to master and branch-4.0.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1956769837
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,544 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
Review Comment:
Users might not have pyarrow installed as it's not a required package. We need to throw a user friendly error message similar to check_dependencies when pyarrow is not installed.
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,544 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
+
+from pyspark.errors import PySparkValueError
+from pyspark.sql.pandas.types import (
+ _dedup_names,
+ _deduplicate_field_names,
+ to_arrow_schema,
+)
+from pyspark.sql.types import (
+ ArrayType,
+ BinaryType,
+ DataType,
+ DecimalType,
+ MapType,
+ NullType,
+ Row,
+ StringType,
+ StructField,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ UserDefinedType,
+ VariantType,
+ VariantVal,
+ _create_row,
+)
+
+
+class LocalDataToArrowConversion:
+ """
+ Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
+ Currently, only :class:`SparkSession` in Spark Connect can use this class.
Review Comment:
Can we update this comment?
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,544 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
+
+from pyspark.errors import PySparkValueError
+from pyspark.sql.pandas.types import (
+ _dedup_names,
+ _deduplicate_field_names,
+ to_arrow_schema,
+)
+from pyspark.sql.types import (
+ ArrayType,
+ BinaryType,
+ DataType,
+ DecimalType,
+ MapType,
+ NullType,
+ Row,
+ StringType,
+ StructField,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ UserDefinedType,
+ VariantType,
+ VariantVal,
+ _create_row,
+)
+
+
+class LocalDataToArrowConversion:
+ """
+ Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
+ Currently, only :class:`SparkSession` in Spark Connect can use this class.
+ """
+
+ @staticmethod
+ def _need_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ ) -> bool:
+ if not nullable:
+ # always check the nullability
+ return True
+ elif isinstance(dataType, NullType):
+ # always check the nullability
+ return True
+ elif isinstance(dataType, StructType):
+ # Struct maybe rows, should convert to dict.
+ return True
+ elif isinstance(dataType, ArrayType):
+ return LocalDataToArrowConversion._need_converter(
+ dataType.elementType, dataType.containsNull
+ )
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
+ # since an Arrow Map requires a list of tuples.
+ return True
+ elif isinstance(dataType, BinaryType):
+ return True
+ elif isinstance(dataType, (TimestampType, TimestampNTZType)):
+ # Always truncate
+ return True
+ elif isinstance(dataType, DecimalType):
+ # Convert Decimal('NaN') to None
+ return True
+ elif isinstance(dataType, StringType):
+ # Coercion to StringType is allowed
+ return True
+ elif isinstance(dataType, UserDefinedType):
+ return True
+ elif isinstance(dataType, VariantType):
+ return True
+ else:
+ return False
+
+ @staticmethod
+ def _create_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ variants_as_dicts: bool = False, # some code paths may require python internal types
+ ) -> Callable:
+ assert dataType is not None and isinstance(dataType, DataType)
+ assert isinstance(nullable, bool)
+
+ if not LocalDataToArrowConversion._need_converter(dataType, nullable):
+ return lambda value: value
+
+ if isinstance(dataType, NullType):
+
+ def convert_null(value: Any) -> Any:
+ if value is not None:
+ raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
+ return None
+
+ return convert_null
+
+ elif isinstance(dataType, StructType):
+ field_names = dataType.fieldNames()
+ dedup_field_names = _dedup_names(dataType.names)
+
+ field_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts
+ )
+ for field in dataType.fields
+ ]
+
+ def convert_struct(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (tuple, dict)) or hasattr(
+ value, "__dict__"
+ ), f"{type(value)} {value}"
+
+ _dict = {}
+ if (
+ not isinstance(value, Row)
+ and not isinstance(value, tuple) # inherited namedtuple
+ and hasattr(value, "__dict__")
+ ):
+ value = value.__dict__
+ if isinstance(value, dict):
+ for i, field in enumerate(field_names):
+ _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
+ else:
+ if len(value) != len(field_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(field_names)),
+ "actual_length": str(len(value)),
+ },
+ )
+ for i in range(len(field_names)):
+ _dict[dedup_field_names[i]] = field_convs[i](value[i])
+
+ return _dict
+
+ return convert_struct
+
+ elif isinstance(dataType, ArrayType):
+ element_conv = LocalDataToArrowConversion._create_converter(
+ dataType.elementType, dataType.containsNull, variants_as_dicts
+ )
+
+ def convert_array(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (list, array.array))
+ return [element_conv(v) for v in value]
+
+ return convert_array
+
+ elif isinstance(dataType, MapType):
+ key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
+ value_conv = LocalDataToArrowConversion._create_converter(
+ dataType.valueType, dataType.valueContainsNull, variants_as_dicts
+ )
+
+ def convert_map(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, dict)
+
+ _tuples = []
+ for k, v in value.items():
+ _tuples.append((key_conv(k), value_conv(v)))
+
+ return _tuples
+
+ return convert_map
+
+ elif isinstance(dataType, BinaryType):
+
+ def convert_binary(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (bytes, bytearray))
+ return bytes(value)
+
+ return convert_binary
+
+ elif isinstance(dataType, TimestampType):
+
+ def convert_timestamp(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime)
+ return value.astimezone(datetime.timezone.utc)
+
+ return convert_timestamp
+
+ elif isinstance(dataType, TimestampNTZType):
+
+ def convert_timestamp_ntz(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime) and value.tzinfo is None
+ return value
+
+ return convert_timestamp_ntz
+
+ elif isinstance(dataType, DecimalType):
+
+ def convert_decimal(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, decimal.Decimal)
+ return None if value.is_nan() else value
+
+ return convert_decimal
+
+ elif isinstance(dataType, StringType):
+
+ def convert_string(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ if isinstance(value, bool):
+ # To match the PySpark which convert bool to string in
+ # the JVM side (python.EvaluatePython.makeFromJava)
+ return str(value).lower()
+ else:
+ return str(value)
+
+ return convert_string
+
+ elif isinstance(dataType, UserDefinedType):
+ udt: UserDefinedType = dataType
+
+ conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
+
+ def convert_udt(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ return conv(udt.serialize(value))
+
+ return convert_udt
+
+ elif isinstance(dataType, VariantType):
+
+ def convert_variant(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ elif (
+ isinstance(value, dict)
+ and all(key in value for key in ["value", "metadata"])
+ and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
+ and not variants_as_dicts
+ ):
+ return VariantVal(value["value"], value["metadata"])
+ elif isinstance(value, VariantVal) and variants_as_dicts:
+ return VariantType().toInternal(value)
+ else:
+ raise PySparkValueError(errorClass="MALFORMED_VARIANT")
+
+ return convert_variant
+
+ elif not nullable:
+
+ def convert_other(value: Any) -> Any:
+ if value is None:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return value
+
+ return convert_other
+ else:
+ return lambda value: value
+
+ @staticmethod
+ def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
+ assert isinstance(data, list) and len(data) > 0
+
+ assert schema is not None and isinstance(schema, StructType)
+
+ column_names = schema.fieldNames()
+
+ column_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts=True
+ )
+ for field in schema.fields
+ ]
+
+ pylist: List[List] = [[] for _ in range(len(column_names))]
+
+ for item in data:
+ if isinstance(item, VariantVal):
+ raise PySparkValueError("Rows cannot be of type VariantVal")
+ if (
+ not isinstance(item, Row)
+ and not isinstance(item, tuple) # inherited namedtuple
+ and hasattr(item, "__dict__")
+ ):
+ item = item.__dict__
+ if isinstance(item, dict):
+ for i, col in enumerate(column_names):
+ pylist[i].append(column_convs[i](item.get(col)))
+ else:
+ if len(item) != len(column_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(column_names)),
+ "actual_length": str(len(item)),
+ },
+ )
+
+ for i in range(len(column_names)):
+ pylist[i].append(column_convs[i](item[i]))
+
+ pa_schema = to_arrow_schema(
+ StructType(
+ [
+ StructField(
+ field.name, _deduplicate_field_names(field.dataType), field.nullable
+ )
+ for field in schema.fields
+ ]
+ ),
+ prefers_large_types=use_large_var_types,
+ )
+
+ return pa.Table.from_arrays(pylist, schema=pa_schema)
+
+
+class ArrowTableToRowsConversion:
+ """
+ Conversion from Arrow Table to Rows.
+ Currently, only :class:`DataFrame` in Spark Connect can use this class.
Review Comment:
nit: this also needs update
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960748551
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
Review Comment:
We have called `pyspark.sql.pandas.utils.require_minimum_pyarrow_version()` before this import to make sure arrow is present.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960773807
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
Review Comment:
`require_minimum_pyarrow_version` throws an exception, and `pyspark.sql` doesn't require PyArrow by default. If this package is imported somewhere within `pyspark.sql`, it will fail to import `pyspark.sql.` at all.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1957500704
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,543 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+# Arrow is required for arrow conversion.
+require_minimum_pyarrow_version()
+
+
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
+from pyspark.errors import PySparkValueError
+from pyspark.sql.pandas.types import _dedup_names, _deduplicate_field_names, to_arrow_schema
+from pyspark.sql.types import (
+ ArrayType,
+ BinaryType,
+ DataType,
+ DecimalType,
+ MapType,
+ NullType,
+ Row,
+ StringType,
+ StructField,
+ StructType,
+ TimestampNTZType,
+ TimestampType,
+ UserDefinedType,
+ VariantType,
+ VariantVal,
+ _create_row,
+)
+
+
+class LocalDataToArrowConversion:
+ """
+ Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
+ """
+
+ @staticmethod
+ def _need_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ ) -> bool:
+ if not nullable:
+ # always check the nullability
+ return True
+ elif isinstance(dataType, NullType):
+ # always check the nullability
+ return True
+ elif isinstance(dataType, StructType):
+ # Struct maybe rows, should convert to dict.
+ return True
+ elif isinstance(dataType, ArrayType):
+ return LocalDataToArrowConversion._need_converter(
+ dataType.elementType, dataType.containsNull
+ )
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
+ # since an Arrow Map requires a list of tuples.
+ return True
+ elif isinstance(dataType, BinaryType):
+ return True
+ elif isinstance(dataType, (TimestampType, TimestampNTZType)):
+ # Always truncate
+ return True
+ elif isinstance(dataType, DecimalType):
+ # Convert Decimal('NaN') to None
+ return True
+ elif isinstance(dataType, StringType):
+ # Coercion to StringType is allowed
+ return True
+ elif isinstance(dataType, UserDefinedType):
+ return True
+ elif isinstance(dataType, VariantType):
+ return True
+ else:
+ return False
+
+ @staticmethod
+ def _create_converter(
+ dataType: DataType,
+ nullable: bool = True,
+ variants_as_dicts: bool = False, # some code paths may require python internal types
+ ) -> Callable:
+ assert dataType is not None and isinstance(dataType, DataType)
+ assert isinstance(nullable, bool)
+
+ if not LocalDataToArrowConversion._need_converter(dataType, nullable):
+ return lambda value: value
+
+ if isinstance(dataType, NullType):
+
+ def convert_null(value: Any) -> Any:
+ if value is not None:
+ raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
+ return None
+
+ return convert_null
+
+ elif isinstance(dataType, StructType):
+ field_names = dataType.fieldNames()
+ dedup_field_names = _dedup_names(dataType.names)
+
+ field_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts
+ )
+ for field in dataType.fields
+ ]
+
+ def convert_struct(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (tuple, dict)) or hasattr(
+ value, "__dict__"
+ ), f"{type(value)} {value}"
+
+ _dict = {}
+ if (
+ not isinstance(value, Row)
+ and not isinstance(value, tuple) # inherited namedtuple
+ and hasattr(value, "__dict__")
+ ):
+ value = value.__dict__
+ if isinstance(value, dict):
+ for i, field in enumerate(field_names):
+ _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
+ else:
+ if len(value) != len(field_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(field_names)),
+ "actual_length": str(len(value)),
+ },
+ )
+ for i in range(len(field_names)):
+ _dict[dedup_field_names[i]] = field_convs[i](value[i])
+
+ return _dict
+
+ return convert_struct
+
+ elif isinstance(dataType, ArrayType):
+ element_conv = LocalDataToArrowConversion._create_converter(
+ dataType.elementType, dataType.containsNull, variants_as_dicts
+ )
+
+ def convert_array(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (list, array.array))
+ return [element_conv(v) for v in value]
+
+ return convert_array
+
+ elif isinstance(dataType, MapType):
+ key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
+ value_conv = LocalDataToArrowConversion._create_converter(
+ dataType.valueType, dataType.valueContainsNull, variants_as_dicts
+ )
+
+ def convert_map(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, dict)
+
+ _tuples = []
+ for k, v in value.items():
+ _tuples.append((key_conv(k), value_conv(v)))
+
+ return _tuples
+
+ return convert_map
+
+ elif isinstance(dataType, BinaryType):
+
+ def convert_binary(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, (bytes, bytearray))
+ return bytes(value)
+
+ return convert_binary
+
+ elif isinstance(dataType, TimestampType):
+
+ def convert_timestamp(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime)
+ return value.astimezone(datetime.timezone.utc)
+
+ return convert_timestamp
+
+ elif isinstance(dataType, TimestampNTZType):
+
+ def convert_timestamp_ntz(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, datetime.datetime) and value.tzinfo is None
+ return value
+
+ return convert_timestamp_ntz
+
+ elif isinstance(dataType, DecimalType):
+
+ def convert_decimal(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ assert isinstance(value, decimal.Decimal)
+ return None if value.is_nan() else value
+
+ return convert_decimal
+
+ elif isinstance(dataType, StringType):
+
+ def convert_string(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ if isinstance(value, bool):
+ # To match the PySpark which convert bool to string in
+ # the JVM side (python.EvaluatePython.makeFromJava)
+ return str(value).lower()
+ else:
+ return str(value)
+
+ return convert_string
+
+ elif isinstance(dataType, UserDefinedType):
+ udt: UserDefinedType = dataType
+
+ conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
+
+ def convert_udt(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ else:
+ return conv(udt.serialize(value))
+
+ return convert_udt
+
+ elif isinstance(dataType, VariantType):
+
+ def convert_variant(value: Any) -> Any:
+ if value is None:
+ if not nullable:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return None
+ elif (
+ isinstance(value, dict)
+ and all(key in value for key in ["value", "metadata"])
+ and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
+ and not variants_as_dicts
+ ):
+ return VariantVal(value["value"], value["metadata"])
+ elif isinstance(value, VariantVal) and variants_as_dicts:
+ return VariantType().toInternal(value)
+ else:
+ raise PySparkValueError(errorClass="MALFORMED_VARIANT")
+
+ return convert_variant
+
+ elif not nullable:
+
+ def convert_other(value: Any) -> Any:
+ if value is None:
+ raise PySparkValueError(f"input for {dataType} must not be None")
+ return value
+
+ return convert_other
+ else:
+ return lambda value: value
+
+ @staticmethod
+ def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
+ assert isinstance(data, list) and len(data) > 0
+
+ assert schema is not None and isinstance(schema, StructType)
+
+ column_names = schema.fieldNames()
+
+ column_convs = [
+ LocalDataToArrowConversion._create_converter(
+ field.dataType, field.nullable, variants_as_dicts=True
+ )
+ for field in schema.fields
+ ]
+
+ pylist: List[List] = [[] for _ in range(len(column_names))]
+
+ for item in data:
+ if isinstance(item, VariantVal):
+ raise PySparkValueError("Rows cannot be of type VariantVal")
+ if (
+ not isinstance(item, Row)
+ and not isinstance(item, tuple) # inherited namedtuple
+ and hasattr(item, "__dict__")
+ ):
+ item = item.__dict__
+ if isinstance(item, dict):
+ for i, col in enumerate(column_names):
+ pylist[i].append(column_convs[i](item.get(col)))
+ else:
+ if len(item) != len(column_names):
+ raise PySparkValueError(
+ errorClass="AXIS_LENGTH_MISMATCH",
+ messageParameters={
+ "expected_length": str(len(column_names)),
+ "actual_length": str(len(item)),
+ },
+ )
+
+ for i in range(len(column_names)):
+ pylist[i].append(column_convs[i](item[i]))
+
+ pa_schema = to_arrow_schema(
+ StructType(
+ [
+ StructField(
+ field.name, _deduplicate_field_names(field.dataType), field.nullable
+ )
+ for field in schema.fields
+ ]
+ ),
+ prefers_large_types=use_large_var_types,
+ )
+
+ return pa.Table.from_arrays(pylist, schema=pa_schema)
+
+
+class ArrowTableToRowsConversion:
+ """
+ Conversion from Arrow Table to Rows.
+ """
+
+ @staticmethod
+ def _need_converter(dataType: DataType) -> bool:
+ if isinstance(dataType, NullType):
+ return True
+ elif isinstance(dataType, StructType):
+ return True
+ elif isinstance(dataType, ArrayType):
+ return ArrowTableToRowsConversion._need_converter(dataType.elementType)
+ elif isinstance(dataType, MapType):
+ # Different from PySpark, here always needs conversion,
Review Comment:
I remember there is another version of conversion for PySpark classic, do we plan to unify them in the future?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960733966
##########
python/pyspark/sql/connect/conversion.py:
##########
@@ -14,541 +14,31 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-from pyspark.errors import PySparkValueError
from pyspark.sql.connect.utils import check_dependencies
check_dependencies(__name__)
-import array
-import datetime
-import decimal
-import pyarrow as pa
+from typing import TYPE_CHECKING
-from pyspark.sql.types import (
- _create_row,
- Row,
- DataType,
- TimestampType,
- TimestampNTZType,
- MapType,
- StructField,
- StructType,
- ArrayType,
- BinaryType,
- NullType,
- DecimalType,
- StringType,
- UserDefinedType,
- VariantType,
- VariantVal,
+import pyspark.sql.connect.proto as pb2
+from pyspark.sql.conversion import (
+ ArrowTableToRowsConversion,
+ LocalDataToArrowConversion,
)
-
from pyspark.storagelevel import StorageLevel
-import pyspark.sql.connect.proto as pb2
-from pyspark.sql.pandas.types import to_arrow_schema, _dedup_names, _deduplicate_field_names
-
-from typing import Any, Callable, Sequence, List, TYPE_CHECKING
if TYPE_CHECKING:
from pyspark.sql.connect.dataframe import DataFrame
-class LocalDataToArrowConversion:
- """
- Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
- Currently, only :class:`SparkSession` in Spark Connect can use this class.
- """
-
- @staticmethod
- def _need_converter(
- dataType: DataType,
- nullable: bool = True,
- ) -> bool:
- if not nullable:
- # always check the nullability
- return True
- elif isinstance(dataType, NullType):
- # always check the nullability
- return True
- elif isinstance(dataType, StructType):
- # Struct maybe rows, should convert to dict.
- return True
- elif isinstance(dataType, ArrayType):
- return LocalDataToArrowConversion._need_converter(
- dataType.elementType, dataType.containsNull
- )
- elif isinstance(dataType, MapType):
- # Different from PySpark, here always needs conversion,
- # since an Arrow Map requires a list of tuples.
- return True
- elif isinstance(dataType, BinaryType):
- return True
- elif isinstance(dataType, (TimestampType, TimestampNTZType)):
- # Always truncate
- return True
- elif isinstance(dataType, DecimalType):
- # Convert Decimal('NaN') to None
- return True
- elif isinstance(dataType, StringType):
- # Coercion to StringType is allowed
- return True
- elif isinstance(dataType, UserDefinedType):
- return True
- elif isinstance(dataType, VariantType):
- return True
- else:
- return False
-
- @staticmethod
- def _create_converter(
- dataType: DataType,
- nullable: bool = True,
- variants_as_dicts: bool = False, # some code paths may require python internal types
- ) -> Callable:
- assert dataType is not None and isinstance(dataType, DataType)
- assert isinstance(nullable, bool)
-
- if not LocalDataToArrowConversion._need_converter(dataType, nullable):
- return lambda value: value
-
- if isinstance(dataType, NullType):
-
- def convert_null(value: Any) -> Any:
- if value is not None:
- raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
- return None
-
- return convert_null
-
- elif isinstance(dataType, StructType):
- field_names = dataType.fieldNames()
- dedup_field_names = _dedup_names(dataType.names)
-
- field_convs = [
- LocalDataToArrowConversion._create_converter(
- field.dataType, field.nullable, variants_as_dicts
- )
- for field in dataType.fields
- ]
-
- def convert_struct(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, (tuple, dict)) or hasattr(
- value, "__dict__"
- ), f"{type(value)} {value}"
-
- _dict = {}
- if (
- not isinstance(value, Row)
- and not isinstance(value, tuple) # inherited namedtuple
- and hasattr(value, "__dict__")
- ):
- value = value.__dict__
- if isinstance(value, dict):
- for i, field in enumerate(field_names):
- _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
- else:
- if len(value) != len(field_names):
- raise PySparkValueError(
- errorClass="AXIS_LENGTH_MISMATCH",
- messageParameters={
- "expected_length": str(len(field_names)),
- "actual_length": str(len(value)),
- },
- )
- for i in range(len(field_names)):
- _dict[dedup_field_names[i]] = field_convs[i](value[i])
-
- return _dict
-
- return convert_struct
-
- elif isinstance(dataType, ArrayType):
- element_conv = LocalDataToArrowConversion._create_converter(
- dataType.elementType, dataType.containsNull, variants_as_dicts
- )
-
- def convert_array(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, (list, array.array))
- return [element_conv(v) for v in value]
-
- return convert_array
-
- elif isinstance(dataType, MapType):
- key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
- value_conv = LocalDataToArrowConversion._create_converter(
- dataType.valueType, dataType.valueContainsNull, variants_as_dicts
- )
-
- def convert_map(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, dict)
-
- _tuples = []
- for k, v in value.items():
- _tuples.append((key_conv(k), value_conv(v)))
-
- return _tuples
-
- return convert_map
-
- elif isinstance(dataType, BinaryType):
-
- def convert_binary(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, (bytes, bytearray))
- return bytes(value)
-
- return convert_binary
-
- elif isinstance(dataType, TimestampType):
-
- def convert_timestamp(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, datetime.datetime)
- return value.astimezone(datetime.timezone.utc)
-
- return convert_timestamp
-
- elif isinstance(dataType, TimestampNTZType):
-
- def convert_timestamp_ntz(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, datetime.datetime) and value.tzinfo is None
- return value
-
- return convert_timestamp_ntz
-
- elif isinstance(dataType, DecimalType):
-
- def convert_decimal(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, decimal.Decimal)
- return None if value.is_nan() else value
-
- return convert_decimal
-
- elif isinstance(dataType, StringType):
-
- def convert_string(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- if isinstance(value, bool):
- # To match the PySpark which convert bool to string in
- # the JVM side (python.EvaluatePython.makeFromJava)
- return str(value).lower()
- else:
- return str(value)
-
- return convert_string
-
- elif isinstance(dataType, UserDefinedType):
- udt: UserDefinedType = dataType
-
- conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
-
- def convert_udt(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- return conv(udt.serialize(value))
-
- return convert_udt
-
- elif isinstance(dataType, VariantType):
-
- def convert_variant(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- elif (
- isinstance(value, dict)
- and all(key in value for key in ["value", "metadata"])
- and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
- and not variants_as_dicts
- ):
- return VariantVal(value["value"], value["metadata"])
- elif isinstance(value, VariantVal) and variants_as_dicts:
- return VariantType().toInternal(value)
- else:
- raise PySparkValueError(errorClass="MALFORMED_VARIANT")
-
- return convert_variant
-
- elif not nullable:
-
- def convert_other(value: Any) -> Any:
- if value is None:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return value
-
- return convert_other
- else:
- return lambda value: value
-
- @staticmethod
- def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
- assert isinstance(data, list) and len(data) > 0
-
- assert schema is not None and isinstance(schema, StructType)
-
- column_names = schema.fieldNames()
-
- column_convs = [
- LocalDataToArrowConversion._create_converter(
- field.dataType, field.nullable, variants_as_dicts=True
- )
- for field in schema.fields
- ]
-
- pylist: List[List] = [[] for _ in range(len(column_names))]
-
- for item in data:
- if isinstance(item, VariantVal):
- raise PySparkValueError("Rows cannot be of type VariantVal")
- if (
- not isinstance(item, Row)
- and not isinstance(item, tuple) # inherited namedtuple
- and hasattr(item, "__dict__")
- ):
- item = item.__dict__
- if isinstance(item, dict):
- for i, col in enumerate(column_names):
- pylist[i].append(column_convs[i](item.get(col)))
- else:
- if len(item) != len(column_names):
- raise PySparkValueError(
- errorClass="AXIS_LENGTH_MISMATCH",
- messageParameters={
- "expected_length": str(len(column_names)),
- "actual_length": str(len(item)),
- },
- )
-
- for i in range(len(column_names)):
- pylist[i].append(column_convs[i](item[i]))
-
- pa_schema = to_arrow_schema(
- StructType(
- [
- StructField(
- field.name, _deduplicate_field_names(field.dataType), field.nullable
- )
- for field in schema.fields
- ]
- ),
- prefers_large_types=use_large_var_types,
- )
-
- return pa.Table.from_arrays(pylist, schema=pa_schema)
-
-
-class ArrowTableToRowsConversion:
- """
- Conversion from Arrow Table to Rows.
- Currently, only :class:`DataFrame` in Spark Connect can use this class.
- """
-
- @staticmethod
- def _need_converter(dataType: DataType) -> bool:
- if isinstance(dataType, NullType):
- return True
- elif isinstance(dataType, StructType):
- return True
- elif isinstance(dataType, ArrayType):
- return ArrowTableToRowsConversion._need_converter(dataType.elementType)
- elif isinstance(dataType, MapType):
- # Different from PySpark, here always needs conversion,
- # since the input from Arrow is a list of tuples.
- return True
- elif isinstance(dataType, BinaryType):
- return True
- elif isinstance(dataType, (TimestampType, TimestampNTZType)):
- # Always remove the time zone info for now
- return True
- elif isinstance(dataType, UserDefinedType):
- return True
- elif isinstance(dataType, VariantType):
- return True
- else:
- return False
-
- @staticmethod
- def _create_converter(dataType: DataType) -> Callable:
- assert dataType is not None and isinstance(dataType, DataType)
-
- if not ArrowTableToRowsConversion._need_converter(dataType):
- return lambda value: value
-
- if isinstance(dataType, NullType):
- return lambda value: None
-
- elif isinstance(dataType, StructType):
- field_names = dataType.names
- dedup_field_names = _dedup_names(field_names)
-
- field_convs = [
- ArrowTableToRowsConversion._create_converter(f.dataType) for f in dataType.fields
- ]
-
- def convert_struct(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, dict)
-
- _values = [
- field_convs[i](value.get(name, None))
- for i, name in enumerate(dedup_field_names)
- ]
- return _create_row(field_names, _values)
-
- return convert_struct
-
- elif isinstance(dataType, ArrayType):
- element_conv = ArrowTableToRowsConversion._create_converter(dataType.elementType)
-
- def convert_array(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, list)
- return [element_conv(v) for v in value]
-
- return convert_array
-
- elif isinstance(dataType, MapType):
- key_conv = ArrowTableToRowsConversion._create_converter(dataType.keyType)
- value_conv = ArrowTableToRowsConversion._create_converter(dataType.valueType)
-
- def convert_map(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, list)
- assert all(isinstance(t, tuple) and len(t) == 2 for t in value)
- return dict((key_conv(t[0]), value_conv(t[1])) for t in value)
-
- return convert_map
-
- elif isinstance(dataType, BinaryType):
-
- def convert_binary(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, bytes)
- return bytearray(value)
-
- return convert_binary
-
- elif isinstance(dataType, TimestampType):
-
- def convert_timestample(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, datetime.datetime)
- return value.astimezone().replace(tzinfo=None)
-
- return convert_timestample
-
- elif isinstance(dataType, TimestampNTZType):
-
- def convert_timestample_ntz(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, datetime.datetime)
- return value
-
- return convert_timestample_ntz
-
- elif isinstance(dataType, UserDefinedType):
- udt: UserDefinedType = dataType
-
- conv = ArrowTableToRowsConversion._create_converter(udt.sqlType())
-
- def convert_udt(value: Any) -> Any:
- if value is None:
- return None
- else:
- return udt.deserialize(conv(value))
-
- return convert_udt
-
- elif isinstance(dataType, VariantType):
-
- def convert_variant(value: Any) -> Any:
- if value is None:
- return None
- elif (
- isinstance(value, dict)
- and all(key in value for key in ["value", "metadata"])
- and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
- ):
- return VariantVal(value["value"], value["metadata"])
- else:
- raise PySparkValueError(errorClass="MALFORMED_VARIANT")
-
- return convert_variant
-
- else:
- return lambda value: value
-
- @staticmethod
- def convert(table: "pa.Table", schema: StructType) -> List[Row]:
- assert isinstance(table, pa.Table)
-
- assert schema is not None and isinstance(schema, StructType)
-
- field_converters = [
- ArrowTableToRowsConversion._create_converter(f.dataType) for f in schema.fields
- ]
-
- columnar_data = [column.to_pylist() for column in table.columns]
-
- rows: List[Row] = []
- for i in range(0, table.num_rows):
- values = [field_converters[j](columnar_data[j][i]) for j in range(table.num_columns)]
- rows.append(_create_row(fields=schema.fieldNames(), values=values))
- return rows
+__all__ = [
Review Comment:
I think you can remove this. this conversion package isn't an API, and documented. It's just an internal package.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960735653
##########
python/pyspark/sql/worker/plan_data_source_read.py:
##########
@@ -32,7 +32,10 @@
SpecialLengths,
)
from pyspark.sql import Row
-from pyspark.sql.connect.conversion import ArrowTableToRowsConversion, LocalDataToArrowConversion
+from pyspark.sql.conversion import (
+ ArrowTableToRowsConversion,
+ LocalDataToArrowConversion,
+)
Review Comment:
Shall we make the import single line?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1960774711
##########
python/pyspark/sql/connect/conversion.py:
##########
@@ -14,541 +14,31 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-from pyspark.errors import PySparkValueError
from pyspark.sql.connect.utils import check_dependencies
check_dependencies(__name__)
-import array
-import datetime
-import decimal
-import pyarrow as pa
+from typing import TYPE_CHECKING
-from pyspark.sql.types import (
- _create_row,
- Row,
- DataType,
- TimestampType,
- TimestampNTZType,
- MapType,
- StructField,
- StructType,
- ArrayType,
- BinaryType,
- NullType,
- DecimalType,
- StringType,
- UserDefinedType,
- VariantType,
- VariantVal,
+import pyspark.sql.connect.proto as pb2
+from pyspark.sql.conversion import (
+ ArrowTableToRowsConversion,
+ LocalDataToArrowConversion,
)
-
from pyspark.storagelevel import StorageLevel
-import pyspark.sql.connect.proto as pb2
-from pyspark.sql.pandas.types import to_arrow_schema, _dedup_names, _deduplicate_field_names
-
-from typing import Any, Callable, Sequence, List, TYPE_CHECKING
if TYPE_CHECKING:
from pyspark.sql.connect.dataframe import DataFrame
-class LocalDataToArrowConversion:
- """
- Conversion from local data (except pandas DataFrame and numpy ndarray) to Arrow.
- Currently, only :class:`SparkSession` in Spark Connect can use this class.
- """
-
- @staticmethod
- def _need_converter(
- dataType: DataType,
- nullable: bool = True,
- ) -> bool:
- if not nullable:
- # always check the nullability
- return True
- elif isinstance(dataType, NullType):
- # always check the nullability
- return True
- elif isinstance(dataType, StructType):
- # Struct maybe rows, should convert to dict.
- return True
- elif isinstance(dataType, ArrayType):
- return LocalDataToArrowConversion._need_converter(
- dataType.elementType, dataType.containsNull
- )
- elif isinstance(dataType, MapType):
- # Different from PySpark, here always needs conversion,
- # since an Arrow Map requires a list of tuples.
- return True
- elif isinstance(dataType, BinaryType):
- return True
- elif isinstance(dataType, (TimestampType, TimestampNTZType)):
- # Always truncate
- return True
- elif isinstance(dataType, DecimalType):
- # Convert Decimal('NaN') to None
- return True
- elif isinstance(dataType, StringType):
- # Coercion to StringType is allowed
- return True
- elif isinstance(dataType, UserDefinedType):
- return True
- elif isinstance(dataType, VariantType):
- return True
- else:
- return False
-
- @staticmethod
- def _create_converter(
- dataType: DataType,
- nullable: bool = True,
- variants_as_dicts: bool = False, # some code paths may require python internal types
- ) -> Callable:
- assert dataType is not None and isinstance(dataType, DataType)
- assert isinstance(nullable, bool)
-
- if not LocalDataToArrowConversion._need_converter(dataType, nullable):
- return lambda value: value
-
- if isinstance(dataType, NullType):
-
- def convert_null(value: Any) -> Any:
- if value is not None:
- raise PySparkValueError(f"input for {dataType} must be None, but got {value}")
- return None
-
- return convert_null
-
- elif isinstance(dataType, StructType):
- field_names = dataType.fieldNames()
- dedup_field_names = _dedup_names(dataType.names)
-
- field_convs = [
- LocalDataToArrowConversion._create_converter(
- field.dataType, field.nullable, variants_as_dicts
- )
- for field in dataType.fields
- ]
-
- def convert_struct(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, (tuple, dict)) or hasattr(
- value, "__dict__"
- ), f"{type(value)} {value}"
-
- _dict = {}
- if (
- not isinstance(value, Row)
- and not isinstance(value, tuple) # inherited namedtuple
- and hasattr(value, "__dict__")
- ):
- value = value.__dict__
- if isinstance(value, dict):
- for i, field in enumerate(field_names):
- _dict[dedup_field_names[i]] = field_convs[i](value.get(field))
- else:
- if len(value) != len(field_names):
- raise PySparkValueError(
- errorClass="AXIS_LENGTH_MISMATCH",
- messageParameters={
- "expected_length": str(len(field_names)),
- "actual_length": str(len(value)),
- },
- )
- for i in range(len(field_names)):
- _dict[dedup_field_names[i]] = field_convs[i](value[i])
-
- return _dict
-
- return convert_struct
-
- elif isinstance(dataType, ArrayType):
- element_conv = LocalDataToArrowConversion._create_converter(
- dataType.elementType, dataType.containsNull, variants_as_dicts
- )
-
- def convert_array(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, (list, array.array))
- return [element_conv(v) for v in value]
-
- return convert_array
-
- elif isinstance(dataType, MapType):
- key_conv = LocalDataToArrowConversion._create_converter(dataType.keyType)
- value_conv = LocalDataToArrowConversion._create_converter(
- dataType.valueType, dataType.valueContainsNull, variants_as_dicts
- )
-
- def convert_map(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, dict)
-
- _tuples = []
- for k, v in value.items():
- _tuples.append((key_conv(k), value_conv(v)))
-
- return _tuples
-
- return convert_map
-
- elif isinstance(dataType, BinaryType):
-
- def convert_binary(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, (bytes, bytearray))
- return bytes(value)
-
- return convert_binary
-
- elif isinstance(dataType, TimestampType):
-
- def convert_timestamp(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, datetime.datetime)
- return value.astimezone(datetime.timezone.utc)
-
- return convert_timestamp
-
- elif isinstance(dataType, TimestampNTZType):
-
- def convert_timestamp_ntz(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, datetime.datetime) and value.tzinfo is None
- return value
-
- return convert_timestamp_ntz
-
- elif isinstance(dataType, DecimalType):
-
- def convert_decimal(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- assert isinstance(value, decimal.Decimal)
- return None if value.is_nan() else value
-
- return convert_decimal
-
- elif isinstance(dataType, StringType):
-
- def convert_string(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- if isinstance(value, bool):
- # To match the PySpark which convert bool to string in
- # the JVM side (python.EvaluatePython.makeFromJava)
- return str(value).lower()
- else:
- return str(value)
-
- return convert_string
-
- elif isinstance(dataType, UserDefinedType):
- udt: UserDefinedType = dataType
-
- conv = LocalDataToArrowConversion._create_converter(udt.sqlType())
-
- def convert_udt(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- else:
- return conv(udt.serialize(value))
-
- return convert_udt
-
- elif isinstance(dataType, VariantType):
-
- def convert_variant(value: Any) -> Any:
- if value is None:
- if not nullable:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return None
- elif (
- isinstance(value, dict)
- and all(key in value for key in ["value", "metadata"])
- and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
- and not variants_as_dicts
- ):
- return VariantVal(value["value"], value["metadata"])
- elif isinstance(value, VariantVal) and variants_as_dicts:
- return VariantType().toInternal(value)
- else:
- raise PySparkValueError(errorClass="MALFORMED_VARIANT")
-
- return convert_variant
-
- elif not nullable:
-
- def convert_other(value: Any) -> Any:
- if value is None:
- raise PySparkValueError(f"input for {dataType} must not be None")
- return value
-
- return convert_other
- else:
- return lambda value: value
-
- @staticmethod
- def convert(data: Sequence[Any], schema: StructType, use_large_var_types: bool) -> "pa.Table":
- assert isinstance(data, list) and len(data) > 0
-
- assert schema is not None and isinstance(schema, StructType)
-
- column_names = schema.fieldNames()
-
- column_convs = [
- LocalDataToArrowConversion._create_converter(
- field.dataType, field.nullable, variants_as_dicts=True
- )
- for field in schema.fields
- ]
-
- pylist: List[List] = [[] for _ in range(len(column_names))]
-
- for item in data:
- if isinstance(item, VariantVal):
- raise PySparkValueError("Rows cannot be of type VariantVal")
- if (
- not isinstance(item, Row)
- and not isinstance(item, tuple) # inherited namedtuple
- and hasattr(item, "__dict__")
- ):
- item = item.__dict__
- if isinstance(item, dict):
- for i, col in enumerate(column_names):
- pylist[i].append(column_convs[i](item.get(col)))
- else:
- if len(item) != len(column_names):
- raise PySparkValueError(
- errorClass="AXIS_LENGTH_MISMATCH",
- messageParameters={
- "expected_length": str(len(column_names)),
- "actual_length": str(len(item)),
- },
- )
-
- for i in range(len(column_names)):
- pylist[i].append(column_convs[i](item[i]))
-
- pa_schema = to_arrow_schema(
- StructType(
- [
- StructField(
- field.name, _deduplicate_field_names(field.dataType), field.nullable
- )
- for field in schema.fields
- ]
- ),
- prefers_large_types=use_large_var_types,
- )
-
- return pa.Table.from_arrays(pylist, schema=pa_schema)
-
-
-class ArrowTableToRowsConversion:
- """
- Conversion from Arrow Table to Rows.
- Currently, only :class:`DataFrame` in Spark Connect can use this class.
- """
-
- @staticmethod
- def _need_converter(dataType: DataType) -> bool:
- if isinstance(dataType, NullType):
- return True
- elif isinstance(dataType, StructType):
- return True
- elif isinstance(dataType, ArrayType):
- return ArrowTableToRowsConversion._need_converter(dataType.elementType)
- elif isinstance(dataType, MapType):
- # Different from PySpark, here always needs conversion,
- # since the input from Arrow is a list of tuples.
- return True
- elif isinstance(dataType, BinaryType):
- return True
- elif isinstance(dataType, (TimestampType, TimestampNTZType)):
- # Always remove the time zone info for now
- return True
- elif isinstance(dataType, UserDefinedType):
- return True
- elif isinstance(dataType, VariantType):
- return True
- else:
- return False
-
- @staticmethod
- def _create_converter(dataType: DataType) -> Callable:
- assert dataType is not None and isinstance(dataType, DataType)
-
- if not ArrowTableToRowsConversion._need_converter(dataType):
- return lambda value: value
-
- if isinstance(dataType, NullType):
- return lambda value: None
-
- elif isinstance(dataType, StructType):
- field_names = dataType.names
- dedup_field_names = _dedup_names(field_names)
-
- field_convs = [
- ArrowTableToRowsConversion._create_converter(f.dataType) for f in dataType.fields
- ]
-
- def convert_struct(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, dict)
-
- _values = [
- field_convs[i](value.get(name, None))
- for i, name in enumerate(dedup_field_names)
- ]
- return _create_row(field_names, _values)
-
- return convert_struct
-
- elif isinstance(dataType, ArrayType):
- element_conv = ArrowTableToRowsConversion._create_converter(dataType.elementType)
-
- def convert_array(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, list)
- return [element_conv(v) for v in value]
-
- return convert_array
-
- elif isinstance(dataType, MapType):
- key_conv = ArrowTableToRowsConversion._create_converter(dataType.keyType)
- value_conv = ArrowTableToRowsConversion._create_converter(dataType.valueType)
-
- def convert_map(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, list)
- assert all(isinstance(t, tuple) and len(t) == 2 for t in value)
- return dict((key_conv(t[0]), value_conv(t[1])) for t in value)
-
- return convert_map
-
- elif isinstance(dataType, BinaryType):
-
- def convert_binary(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, bytes)
- return bytearray(value)
-
- return convert_binary
-
- elif isinstance(dataType, TimestampType):
-
- def convert_timestample(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, datetime.datetime)
- return value.astimezone().replace(tzinfo=None)
-
- return convert_timestample
-
- elif isinstance(dataType, TimestampNTZType):
-
- def convert_timestample_ntz(value: Any) -> Any:
- if value is None:
- return None
- else:
- assert isinstance(value, datetime.datetime)
- return value
-
- return convert_timestample_ntz
-
- elif isinstance(dataType, UserDefinedType):
- udt: UserDefinedType = dataType
-
- conv = ArrowTableToRowsConversion._create_converter(udt.sqlType())
-
- def convert_udt(value: Any) -> Any:
- if value is None:
- return None
- else:
- return udt.deserialize(conv(value))
-
- return convert_udt
-
- elif isinstance(dataType, VariantType):
-
- def convert_variant(value: Any) -> Any:
- if value is None:
- return None
- elif (
- isinstance(value, dict)
- and all(key in value for key in ["value", "metadata"])
- and all(isinstance(value[key], bytes) for key in ["value", "metadata"])
- ):
- return VariantVal(value["value"], value["metadata"])
- else:
- raise PySparkValueError(errorClass="MALFORMED_VARIANT")
-
- return convert_variant
-
- else:
- return lambda value: value
-
- @staticmethod
- def convert(table: "pa.Table", schema: StructType) -> List[Row]:
- assert isinstance(table, pa.Table)
-
- assert schema is not None and isinstance(schema, StructType)
-
- field_converters = [
- ArrowTableToRowsConversion._create_converter(f.dataType) for f in schema.fields
- ]
-
- columnar_data = [column.to_pylist() for column in table.columns]
-
- rows: List[Row] = []
- for i in range(0, table.num_rows):
- values = [field_converters[j](columnar_data[j][i]) for j in range(table.num_columns)]
- rows.append(_create_row(fields=schema.fieldNames(), values=values))
- return rows
+__all__ = [
Review Comment:
`pyspark.sql.connect.conversion` isn't API so let's don't try to keep compatibility here.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-51206][PYTHON][CONNECT] Move Arrow conversion helpers out of Spark Connect [spark]
Posted by "wengh (via GitHub)" <gi...@apache.org>.
wengh commented on code in PR #49941:
URL: https://github.com/apache/spark/pull/49941#discussion_r1956794856
##########
python/pyspark/sql/conversion.py:
##########
@@ -0,0 +1,544 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import array
+import datetime
+import decimal
+from typing import Any, Callable, List, Sequence
+
+import pyarrow as pa
Review Comment:
added check by calling require_minimum_pyarrow_version()
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org