This is an automated email from the ASF dual-hosted git repository. gurwls223 pushed a commit to branch branch-4.0 in repository https://gitbox.apache.org/repos/asf/spark.git
commit 4be264114db4fe9bfe8fcf404692fe1bf0e45569 Author: Hyukjin Kwon <[email protected]> AuthorDate: Wed Mar 5 11:12:52 2025 +0900 Revert "[SPARK-48516][PYTHON][CONNECT] Turn on Arrow optimization for Python UDFs by default" This reverts commit 2de9be2d561b5193136e9e7dbdccacf647cead5c. --- python/docs/source/user_guide/sql/arrow_pandas.rst | 4 ++-- python/docs/source/user_guide/sql/type_conversions.rst | 2 +- python/pyspark/ml/base.py | 3 +-- python/pyspark/ml/tuning.py | 6 +----- python/pyspark/sql/connect/udf.py | 10 ---------- python/pyspark/sql/functions/builtin.py | 3 +-- python/pyspark/sql/udf.py | 14 +------------- python/pyspark/sql/utils.py | 9 --------- .../main/scala/org/apache/spark/sql/internal/SQLConf.scala | 2 +- 9 files changed, 8 insertions(+), 45 deletions(-) diff --git a/python/docs/source/user_guide/sql/arrow_pandas.rst b/python/docs/source/user_guide/sql/arrow_pandas.rst index b9e389f8fe7d..fde40140110f 100644 --- a/python/docs/source/user_guide/sql/arrow_pandas.rst +++ b/python/docs/source/user_guide/sql/arrow_pandas.rst @@ -356,8 +356,8 @@ Arrow Python UDFs are user defined functions that are executed row-by-row, utili transfer and serialization. To define an Arrow Python UDF, you can use the :meth:`udf` decorator or wrap the function with the :meth:`udf` method, ensuring the ``useArrow`` parameter is set to True. Additionally, you can enable Arrow optimization for Python UDFs throughout the entire SparkSession by setting the Spark configuration -``spark.sql.execution.pythonUDF.arrow.enabled`` to true, which is the default. It's important to note that the Spark -configuration takes effect only when ``useArrow`` is either not set or set to None. +``spark.sql.execution.pythonUDF.arrow.enabled`` to true. It's important to note that the Spark configuration takes +effect only when ``useArrow`` is either not set or set to None. The type hints for Arrow Python UDFs should be specified in the same way as for default, pickled Python UDFs. diff --git a/python/docs/source/user_guide/sql/type_conversions.rst b/python/docs/source/user_guide/sql/type_conversions.rst index 80f8aa83db7e..2f13701995ef 100644 --- a/python/docs/source/user_guide/sql/type_conversions.rst +++ b/python/docs/source/user_guide/sql/type_conversions.rst @@ -57,7 +57,7 @@ are listed below: - Default * - spark.sql.execution.pythonUDF.arrow.enabled - Enable PyArrow in PySpark. See more `here <arrow_pandas.rst>`_. - - True + - False * - spark.sql.pyspark.inferNestedDictAsStruct.enabled - When enabled, nested dictionaries are inferred as StructType. Otherwise, they are inferred as MapType. - False diff --git a/python/pyspark/ml/base.py b/python/pyspark/ml/base.py index 0bdfa27fc702..224ef34fd5ed 100644 --- a/python/pyspark/ml/base.py +++ b/python/pyspark/ml/base.py @@ -328,8 +328,7 @@ class UnaryTransformer(HasInputCol, HasOutputCol, Transformer): def _transform(self, dataset: DataFrame) -> DataFrame: self.transformSchema(dataset.schema) - # TODO(SPARK-48515): Use Arrow Python UDF - transformUDF = udf(self.createTransformFunc(), self.outputDataType(), useArrow=False) + transformUDF = udf(self.createTransformFunc(), self.outputDataType()) transformedDataset = dataset.withColumn( self.getOutputCol(), transformUDF(dataset[self.getInputCol()]) ) diff --git a/python/pyspark/ml/tuning.py b/python/pyspark/ml/tuning.py index c85f8438079f..d3506bf1c6b0 100644 --- a/python/pyspark/ml/tuning.py +++ b/python/pyspark/ml/tuning.py @@ -906,12 +906,8 @@ class CrossValidator( from pyspark.sql.connect.udf import UserDefinedFunction else: from pyspark.sql.functions import UserDefinedFunction # type: ignore[assignment] - from pyspark.util import PythonEvalType - # TODO(SPARK-48515): Use Arrow Python UDF - checker_udf = UserDefinedFunction( - checker, BooleanType(), evalType=PythonEvalType.SQL_BATCHED_UDF - ) + checker_udf = UserDefinedFunction(checker, BooleanType()) for i in range(nFolds): training = dataset.filter(checker_udf(dataset[foldCol]) & (col(foldCol) != lit(i))) validation = dataset.filter( diff --git a/python/pyspark/sql/connect/udf.py b/python/pyspark/sql/connect/udf.py index 6045e441222d..ab3a2da48ba5 100644 --- a/python/pyspark/sql/connect/udf.py +++ b/python/pyspark/sql/connect/udf.py @@ -41,7 +41,6 @@ from pyspark.sql.udf import ( UDFRegistration as PySparkUDFRegistration, UserDefinedFunction as PySparkUserDefinedFunction, ) -from pyspark.sql.utils import has_arrow from pyspark.errors import PySparkTypeError, PySparkRuntimeError if TYPE_CHECKING: @@ -59,7 +58,6 @@ def _create_py_udf( returnType: "DataTypeOrString", useArrow: Optional[bool] = None, ) -> "UserDefinedFunctionLike": - is_arrow_enabled = False if useArrow is None: is_arrow_enabled = False try: @@ -80,14 +78,6 @@ def _create_py_udf( eval_type: int = PythonEvalType.SQL_BATCHED_UDF - if is_arrow_enabled and not has_arrow: - is_arrow_enabled = False - warnings.warn( - "Arrow optimization failed to enable because PyArrow is not installed. " - "Falling back to a non-Arrow-optimized UDF.", - RuntimeWarning, - ) - if is_arrow_enabled: try: is_func_with_args = len(getfullargspec(f).args) > 0 diff --git a/python/pyspark/sql/functions/builtin.py b/python/pyspark/sql/functions/builtin.py index a19c3261844d..a47db6e8c0cd 100644 --- a/python/pyspark/sql/functions/builtin.py +++ b/python/pyspark/sql/functions/builtin.py @@ -26308,8 +26308,7 @@ def udf( Defaults to :class:`StringType`. useArrow : bool, optional whether to use Arrow to optimize the (de)serialization. When it is None, the - Spark config "spark.sql.execution.pythonUDF.arrow.enabled" takes effect, - which is "true" by default. + Spark config "spark.sql.execution.pythonUDF.arrow.enabled" takes effect. Examples -------- diff --git a/python/pyspark/sql/udf.py b/python/pyspark/sql/udf.py index 79fbf46f005d..ad0cf8774328 100644 --- a/python/pyspark/sql/udf.py +++ b/python/pyspark/sql/udf.py @@ -116,7 +116,7 @@ def _create_py_udf( # Note: The values of 'SQL Type' are DDL formatted strings, which can be used as `returnType`s. # Note: The values inside the table are generated by `repr`. X' means it throws an exception # during the conversion. - is_arrow_enabled = False + if useArrow is None: from pyspark.sql import SparkSession @@ -129,18 +129,6 @@ def _create_py_udf( else: is_arrow_enabled = useArrow - if is_arrow_enabled: - try: - require_minimum_pandas_version() - require_minimum_pyarrow_version() - except ImportError: - is_arrow_enabled = False - warnings.warn( - "Arrow optimization failed to enable because PyArrow/pandas is not installed. " - "Falling back to a non-Arrow-optimized UDF.", - RuntimeWarning, - ) - eval_type: int = PythonEvalType.SQL_BATCHED_UDF if is_arrow_enabled: diff --git a/python/pyspark/sql/utils.py b/python/pyspark/sql/utils.py index 63beda40dc52..b0782d04cba3 100644 --- a/python/pyspark/sql/utils.py +++ b/python/pyspark/sql/utils.py @@ -63,15 +63,6 @@ if TYPE_CHECKING: from pyspark.pandas._typing import IndexOpsLike, SeriesOrIndex -has_arrow: bool = False -try: - import pyarrow # noqa: F401 - - has_arrow = True -except ImportError: - pass - - FuncT = TypeVar("FuncT", bound=Callable[..., Any]) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index dcf512dfd664..85b967753882 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -3505,7 +3505,7 @@ object SQLConf { "can only be enabled when the given function takes at least one argument.") .version("3.4.0") .booleanConf - .createWithDefault(true) + .createWithDefault(false) val PYTHON_UDF_ARROW_CONCURRENCY_LEVEL = buildConf("spark.sql.execution.pythonUDF.arrow.concurrency.level") --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
