[
https://issues.apache.org/jira/browse/SPARK-59718?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dongjoon Hyun updated SPARK-59718:
----------------------------------
Parent: SPARK-58540
Issue Type: Sub-task (was: Improvement)
> Add opt-in in-process Python UDF execution via Arrow C Data Interface
> ---------------------------------------------------------------------
>
> Key: SPARK-59718
> URL: https://issues.apache.org/jira/browse/SPARK-59718
> Project: Spark
> Issue Type: Sub-task
> Components: PySpark
> Affects Versions: 4.4.0
> Reporter: L. C. Hsieh
> Assignee: L. C. Hsieh
> Priority: Major
> Labels: pull-request-available
>
> PySpark Arrow UDFs execute in separate Python worker processes and exchange
> Arrow data with executor JVMs. For workloads with large inputs or outputs and
> relatively inexpensive UDF computation, cross-process data transfer can
> contribute substantially to execution time.
> This proposal adds an opt-in execution path that embeds CPython in the
> executor JVM using JEP and exchanges Arrow arrays through the Arrow C Data
> Interface. This allows sharing compatible Arrow buffers across the JVM/Python
> boundary without copying them, including buffers allocated by other libraries
> and wrapped by PyArrow. Row-to-Arrow conversion and any subsequent conversion
> back to Spark rows still incur costs.
> **Proposed functionality**
> - An `inprocess_udf` Python API accepting one PyArrow array per input column
> and returning a PyArrow array.
> - Catalyst planning and physical execution support.
> - An executor plugin managing interpreter initialization and shutdown.
> - Validation of result lengths and types, Arrow resource cleanup, and task
> cancellation checks.
> - Integration with the existing PySpark SQL test suite.
> - Comparative benchmarks against worker-based Arrow UDFs.
> Existing Python UDF execution remains unchanged. Users explicitly select the
> new API. JEP is a provided dependency, is not bundled with the Spark
> distribution, and must be installed and configured separately.
> **Deployment model and trade-offs**
> The initial implementation supports one concurrent task per executor. Users
> can retain application-level task parallelism by deploying multiple
> executors, typically with `spark.executor.cores` equal to `spark.task.cpus`.
> Interpreter pooling is outside the initial scope.
> This configuration introduces trade-offs associated with additional JVM
> processes, including startup and memory overhead and changes to cache and
> shuffle behavior. The preliminary equal-resource experiments below
> demonstrate that this deployment model can retain a throughput advantage for
> the tested workloads.
> Embedding Python also reduces process isolation: a native Python crash can
> terminate the executor JVM. Task cancellation checks are supported, but
> arbitrary running Python or native code cannot be safely interrupted;
> cancellation may need to wait for the current invocation to return.
> The feature is intended for users who can control executor configuration and
> accept these isolation trade-offs in exchange for lower JVM/Python data
> exchange overhead. It is not intended to replace worker-based Arrow UDF
> execution universally.
> The initial API targets classic PySpark. Spark Connect support is outside the
> initial scope.
> **Preliminary performance results**
> Benchmarks compare in-process execution with worker-based Arrow UDFs using
> the same Arrow inputs, outputs, and UDF operations, avoiding pandas
> conversion differences.
> *Single-task comparison (`local[1]`)*
> For an identity UDF over one column of 1,000-character strings, the PySpark
> ASV benchmark measured approximately **3.0× speedup** across datasets of
> 500,000, 1 million, and 2 million rows.
> Both implementations used the same single-task configuration, cached inputs,
> and an output-consuming sink.
> *Equal-resource, multi-task comparison*
> A separate experiment used four total CPU slots, 4 GiB aggregate executor
> heap, and a 7 GiB total container memory limit:
> | Configuration | UDF execution | Executor layout |
> |---|---|---|
> | A | Worker-based Arrow UDF | One four-core executor |
> | B | Worker-based Arrow UDF | Four single-core executors |
> | C | In-process UDF | Four single-core executors |
> All configurations supported four concurrent tasks. Median query times were:
> | Workload | A | B | C | C speedup over A | C speedup over B |
> |---|---:|---:|---:|---:|---:|
> | Identity over 800,000 strings of 1,000 characters | 0.336 s | 0.486 s |
> 0.261 s | 1.29× | 1.86× |
> | Sum of ten integer columns over 10 million rows | 1.382 s | 1.152 s | 0.570
> s | 2.42× | 2.02× |
> Results are medians of ten timed queries per configuration after warmup,
> using separate executor JVMs and 16 input partitions. All input partitions
> were cached in memory. Output checksums were validated, and all 960 measured
> tasks completed successfully without spilling.
> The single-task and multi-task experiments differ in environment,
> partitioning, and actual batch sizes; their speedup ratios should not be
> interpreted as a controlled scaling comparison. The multi-task experiment ran
> on one machine in resource-limited containers, and the subsecond string
> queries showed noticeable variability. Broader workload and
> distributed-cluster evaluation remains necessary.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
