L. C. Hsieh created SPARK-60415:
-----------------------------------
Summary: Do not use the columnar Arrow input path for scalar
iterator Python UDFs
Key: SPARK-60415
URL: https://issues.apache.org/jira/browse/SPARK-60415
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: L. C. Hsieh
ArrowEvalPythonExec.supportsColumnar does not check the eval type, so scalar
iterator UDFs (SQL_SCALAR_PANDAS_ITER_UDF and SQL_SCALAR_ARROW_ITER_UDF) also
take ColumnarArrowEvalPythonEvaluatorFactory.evalArrowColumnar when the child
is columnar and spark.sql.execution.arrow.pythonUDF.columnarInput.enabled is
true (the default on branch-4.2 and branch-4.3). evalArrowColumnar pairs each
result batch with one input batch and asserts that their row counts match, but
an iterator UDF only has to return the same total number of rows and may
prefetch, so its result batches need not line up with the input batches.
For example, with
spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer,
columnar input enabled and spark.sql.execution.arrow.maxRecordsPerBatch=5:
df = spark.range(10).coalesce(1).cache()
df.count()
@pandas_udf("long")
def f(it: Iterator[pd.Series]) -> Iterator[pd.Series]:
yield pd.concat(list(it)) + 1
df.select(f("id")).collect()
fails with "Batch size mismatch: pass-through has 5 rows but UDF result has 10
rows." The iterator eval types should take the row-queue path instead.
Found in the review of SPARK-60367 (https://github.com/apache/spark/pull/59616).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
