[
https://issues.apache.org/jira/browse/SPARK-58050?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
L. C. Hsieh updated SPARK-58050:
--------------------------------
Description:
Follow-up of SPARK-58019/SPARK-58023/SPARK-58024, scoped down per review
discussion to the output side only.
The Arrow Python UDF worker converts every UDF result through a per-row Python
converter (defensive list copies for arrays, dict to entry-list for maps,
Row/dict to dict for structs) before {{pa.array}}. These conversions are
Spark's own, independent of PyArrow's conversion performance, and use no NumPy;
{{pa.array}} itself is cheap.
Add {{LocalDataToArrowConversion._create_results_to_arrow}}: when the
element/field/value converters are identity, results are assembled in bulk -
arrays pass returned lists to {{pa.array}} directly, map results pass dicts
directly, struct results (Rows/tuples) are transposed and assembled via
{{StructArray.from_arrays}} with a null mask. Any shape/validation mismatch
falls back to the per-row path, preserving error behavior.
Microbenchmarks (400k rows, identical outputs): array<string> 7.5x, map 4.2x,
struct 4.0x. End-to-end (6.4M rows): array<string> 2.39s -> 1.84s. Extending
the bulk path to validation-only child converters (e.g. integers) is possible
follow-up.
was:
Follow-up of SPARK-58019/SPARK-58023/SPARK-58024. Even with bulk
{{_to_pylist}}, the Arrow Python UDF worker still runs per-row Python
converters on both sides: on input, {{_create_converter}} wraps every map row
into a dict and every struct row into a Row one value at a time; on output,
{{LocalDataToArrowConversion}} converters copy every returned list, convert
every dict to an entry list and every Row to a dict before {{pa.array}}. Worker
profiling shows these per-row converters dominate the remaining gap vs pickled
UDFs on nested types (pickle currently beats arrow by 1.4-3.1x on
array/map/struct).
This change:
* adds {{ArrowTableToRowsConversion._to_rows_column}}, fusing the per-row input
converter into the bulk conversion (map rows become dicts and struct rows
become Rows built from flattened child columns; child converters are applied
per flattened column),
* bulk-assembles UDF results when element/field/value converters are identity:
arrays pass the returned lists to {{pa.array}} directly, map results pass dicts
directly (PyArrow accepts dicts for map types), struct results are transposed
and assembled via {{StructArray.from_arrays}}; anything else falls back to the
existing per-row path.
Microbenchmark (400k rows): input map->dict 3.0x, input struct->Row 1.8x,
output array 7.5x, output map 4.2x, output struct 4.0x — outputs identical.
End-to-end UDF benchmark (6.4M rows): array_string 2.39s -> 1.77s,
map<string,int> 3.65s -> 2.40s, struct 6.03s -> 4.87s on top of
SPARK-58019/58023/58024; arrow-vs-pickle outputs verified identical with
injected nulls.
Summary: Bulk-assemble Arrow Python UDF results (was: Fuse per-row
converters into bulk Arrow-to-rows conversion and bulk-assemble Arrow Python
UDF results)
> Bulk-assemble Arrow Python UDF results
> --------------------------------------
>
> Key: SPARK-58050
> URL: https://issues.apache.org/jira/browse/SPARK-58050
> Project: Spark
> Issue Type: Improvement
> Components: PySpark
> Affects Versions: 5.0.0
> Reporter: L. C. Hsieh
> Priority: Major
> Labels: pull-request-available
>
> Follow-up of SPARK-58019/SPARK-58023/SPARK-58024, scoped down per review
> discussion to the output side only.
> The Arrow Python UDF worker converts every UDF result through a per-row
> Python converter (defensive list copies for arrays, dict to entry-list for
> maps, Row/dict to dict for structs) before {{pa.array}}. These conversions
> are Spark's own, independent of PyArrow's conversion performance, and use no
> NumPy; {{pa.array}} itself is cheap.
> Add {{LocalDataToArrowConversion._create_results_to_arrow}}: when the
> element/field/value converters are identity, results are assembled in bulk -
> arrays pass returned lists to {{pa.array}} directly, map results pass dicts
> directly, struct results (Rows/tuples) are transposed and assembled via
> {{StructArray.from_arrays}} with a null mask. Any shape/validation mismatch
> falls back to the per-row path, preserving error behavior.
> Microbenchmarks (400k rows, identical outputs): array<string> 7.5x, map 4.2x,
> struct 4.0x. End-to-end (6.4M rows): array<string> 2.39s -> 1.84s. Extending
> the bulk path to validation-only child converters (e.g. integers) is possible
> follow-up.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
