[jira] [Created] (SPARK-53426) Support named argument for Arrow Python UDTFs

["Allison Wang (Jira)" <ji...@apache.org>]

Allison Wang created SPARK-53426:
------------------------------------

             Summary: Support named argument for Arrow Python UDTFs
                 Key: SPARK-53426
                 URL: https://issues.apache.org/jira/browse/SPARK-53426
             Project: Spark
          Issue Type: Sub-task
          Components: PySpark
    Affects Versions: 4.1.0
            Reporter: Allison Wang


Named argument does not work for Python UDTFs table arguments:

 
{code:java}
    def test_arrow_udtf_with_named_arguments(self):
        @arrow_udtf(returnType="result_id bigint, multiplier_used int")
        class NamedArgsUDTF:
            def eval(
                self,
                table_data: "pa.RecordBatch",
                multiplier: "pa.Array"
            ) -> Iterator["pa.Table"]:
                assert isinstance(
                    table_data, pa.RecordBatch
                ), f"Expected pa.RecordBatch for table_data, got {type(table_data)}"
                assert isinstance(
                    multiplier, pa.Array
                ), f"Expected pa.Array for multiplier, got {type(multiplier)}"                multiplier_val = multiplier[0].as_py()                # Convert record batch to table
                table = pa.table(table_data)
                id_column = table.column("id")                # Multiply each id by the multiplier
                multiplied_ids = pa.compute.multiply(id_column, pa.scalar(multiplier_val))                result_table = pa.table({
                    "result_id": multiplied_ids,
                    "multiplier_used": pa.array([multiplier_val] * table.num_rows, type=pa.int32())
                })
                yield result_table        # Test with DataFrame API using named arguments
        input_df = self.spark.range(3)  # [0, 1, 2]
>       result_df = NamedArgsUDTF(table_data=input_df.asTable(), multiplier=lit(5))python/pyspark/sql/tests/arrow/test_arrow_udtf.py:812:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
python/pyspark/sql/udtf.py:450: in __call__
    j_named_arg = sc._jvm.PythonSQLUtils.namedArgumentExpression(key, j_arg)
../../.virtualenvs/spark/lib/python3.11/site-packages/py4j/java_gateway.py:1322: in __call__
    return_value = get_return_value(
python/pyspark/errors/exceptions/captured.py:288: in deco
    return f(*a, **kw)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _answer = 'xspy4j.Py4JException: Method namedArgumentExpression([class java.lang.String, class org.apache.spark.sql.TableArg]) d....ClientServerConnection.run(ClientServerConnection.java:108)\\n\tat java.base/java.lang.Thread.run(Thread.java:840)\\n'
gateway_client = <py4j.clientserver.JavaClient object at 0x109e20090>
target_id = 'z:org.apache.spark.sql.api.python.PythonSQLUtils', name = 'namedArgumentExpression'    def get_return_value(answer, gateway_client, target_id=None, name=None):
        """Converts an answer received from the Java gateway into a Python object.        For example, string representation of integers are converted to Python
        integer, string representation of objects are converted to JavaObject
        instances, etc.        :param answer: the string returned by the Java gateway
        :param gateway_client: the gateway client used to communicate with the Java
            Gateway. Only necessary if the answer is a reference (e.g., object,
            list, map)
        :param target_id: the name of the object from which the answer comes from
            (e.g., *object1* in `object1.hello()`). Optional.
        :param name: the name of the member from which the answer comes from
            (e.g., *hello* in `object1.hello()`). Optional.
        """
        if is_error(answer)[0]:
            if len(answer) > 1:
                type = answer[1]
                value = OUTPUT_CONVERTER[type](answer[2:], gateway_client)
                if answer[1] == REFERENCE_TYPE:
                    raise Py4JJavaError(
                        "An error occurred while calling {0}{1}{2}.\n".
                        format(target_id, ".", name), value)
                else:
>                   raise Py4JError(
                        "An error occurred while calling {0}{1}{2}. Trace:\n{3}\n".
                        format(target_id, ".", name, value))
E                   py4j.protocol.Py4JError: An error occurred while calling z:org.apache.spark.sql.api.python.PythonSQLUtils.namedArgumentExpression. Trace:
E                   py4j.Py4JException: Method namedArgumentExpression([class java.lang.String, class org.apache.spark.sql.TableArg]) does not exist
E                       at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:321)
E                       at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:342)
E                       at py4j.Gateway.invoke(Gateway.java:276)
E                       at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
E                       at py4j.commands.CallCommand.execute(CallCommand.java:79)
E                       at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:184)
E                       at py4j.ClientServerConnection.run(ClientServerConnection.java:108)
E                       at java.base/java.lang.Thread.run(Thread.java:840) {code}
 

 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org