[
https://issues.apache.org/jira/browse/SPARK-60971?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Hyukjin Kwon updated SPARK-60971:
---------------------------------
Description:
Since PySpark 4.1.0, {{GroupedData.applyInPandas}} takes a function of an
iterator of {{pandas.DataFrame}}, and {{applyInArrow}} one of an iterator of
{{pyarrow.RecordBatch}}. PySpark infers the form from the function's type
hints, with {{infer_group_pandas_eval_type_from_func}} and
{{infer_group_arrow_eval_type_from_func}} of {{pyspark.sql.pandas.typehints}},
and sends the eval type of the iterator form. The Rust-backed Python client
always sends the eval type of the other form, so such a function fails on the
Python worker:
{code:python}
def f(it: Iterator[pd.DataFrame]) -> Iterator[pd.DataFrame]:
for pdf in it:
yield pdf
df.groupBy("k").applyInPandas(f, "k long, v double").collect()
{code}
This works in PySpark 4.3.0 and raises {{PythonException}} in the client. The
same holds for {{applyInArrow}} with {{Iterator[pa.RecordBatch]}}. Infer the
eval type from the type hints as PySpark does.
PySpark also validates the function with {{_validate_vectorized_udf}} of
{{pyspark.sql.pandas.functions}} before it builds the plan. It does so in these
two functions, the cogrouped {{applyInPandas}} and {{applyInArrow}},
{{applyInPandasWithState}}, {{mapInPandas}} and {{mapInArrow}}. For example, a
function of three arguments raises {{INVALID_PANDAS_UDF}}, and a value that is
not callable raises {{TypeError: unsupported callable}}. The client sends such
a function to the server, where it fails in the Python worker. Validate it as
PySpark does.
was:
Since PySpark 4.1.0, {{GroupedData.applyInPandas}} takes a function of an
iterator of {{pandas.DataFrame}}, and {{applyInArrow}} one of an iterator of
{{pyarrow.RecordBatch}}. PySpark infers the form from the function's type
hints, with {{infer_group_pandas_eval_type_from_func}} and
{{infer_group_arrow_eval_type_from_func}} of {{pyspark.sql.pandas.typehints}},
and sends the eval type of the iterator form. The Rust-backed Python client
always sends the eval type of the other form, so such a function fails on the
Python worker:
{code:python}
def f(it: Iterator[pd.DataFrame]) -> Iterator[pd.DataFrame]:
for pdf in it:
yield pdf
df.groupBy("k").applyInPandas(f, "k long, v double").collect()
{code}
This works in PySpark 4.3.0 and raises {{PythonException}} in the client. The
same holds for {{applyInArrow}} with {{Iterator[pa.RecordBatch]}}. Infer the
eval type from the type hints as PySpark does.
> Infer the iterator forms of GroupedData.applyInPandas and applyInArrow from
> type hints in Spark Connect Rust as PySpark does
> ----------------------------------------------------------------------------------------------------------------------------
>
> Key: SPARK-60971
> URL: https://issues.apache.org/jira/browse/SPARK-60971
> Project: Spark
> Issue Type: Bug
> Components: Connect
> Affects Versions: connect-rust-4.3.0
> Reporter: Hyukjin Kwon
> Priority: Major
> Fix For: connect-rust-4.3.0
>
>
> Since PySpark 4.1.0, {{GroupedData.applyInPandas}} takes a function of an
> iterator of {{pandas.DataFrame}}, and {{applyInArrow}} one of an iterator of
> {{pyarrow.RecordBatch}}. PySpark infers the form from the function's type
> hints, with {{infer_group_pandas_eval_type_from_func}} and
> {{infer_group_arrow_eval_type_from_func}} of
> {{pyspark.sql.pandas.typehints}}, and sends the eval type of the iterator
> form. The Rust-backed Python client always sends the eval type of the other
> form, so such a function fails on the Python worker:
> {code:python}
> def f(it: Iterator[pd.DataFrame]) -> Iterator[pd.DataFrame]:
> for pdf in it:
> yield pdf
> df.groupBy("k").applyInPandas(f, "k long, v double").collect()
> {code}
> This works in PySpark 4.3.0 and raises {{PythonException}} in the client. The
> same holds for {{applyInArrow}} with {{Iterator[pa.RecordBatch]}}. Infer the
> eval type from the type hints as PySpark does.
> PySpark also validates the function with {{_validate_vectorized_udf}} of
> {{pyspark.sql.pandas.functions}} before it builds the plan. It does so in
> these two functions, the cogrouped {{applyInPandas}} and {{applyInArrow}},
> {{applyInPandasWithState}}, {{mapInPandas}} and {{mapInArrow}}. For example,
> a function of three arguments raises {{INVALID_PANDAS_UDF}}, and a value that
> is not callable raises {{TypeError: unsupported callable}}. The client sends
> such a function to the server, where it fails in the Python worker. Validate
> it as PySpark does.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
