You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "zhengruifeng (via GitHub)" <gi...@apache.org> on 2026/08/05 04:17:12 UTC
[PR] [WIP][PYTHON] Optimize PySpark broadcast serialization with Arrow [spark]
zhengruifeng opened a new pull request, #57781:
URL: https://github.com/apache/spark/pull/57781
### What changes were proposed in this pull request?
This PR adds a versioned internal serialization envelope for PySpark Classic broadcast variables. Native PyArrow Table, RecordBatch, Array, and ChunkedArray values use Arrow IPC streams. All other Python values continue to use pickle.
The public Broadcast.dump, Broadcast.load, and Broadcast.load_from_path methods remain pickle-specific for compatibility. The Arrow import is also avoided for ordinary broadcasts unless PyArrow is already loaded.
Tests cover Arrow type and value preservation, the pickle fallback, driver and worker reads, and encrypted broadcast transport.
### Why are the changes needed?
PyArrow values already have a columnar representation. Serializing them through Arrow IPC avoids the Python object reconstruction cost of generic pickle and substantially reduces deserialization CPU for large Arrow-native broadcast values while keeping the existing fallback for general Python classes.
### Does this PR introduce _any_ user-facing change?
Yes. In PySpark Classic, broadcasting a native PyArrow Table, RecordBatch, Array, or ChunkedArray now uses Arrow IPC internally. The value exposed on the driver and workers retains the same PyArrow type and data. Other broadcast values continue to use pickle.
### How was this patch tested?
- Added codec round-trip and pickle-fallback unit tests.
- Added driver and worker integration tests, including an encrypted transport variant.
- build/sbt -Phive package passed.
- The focused codec and unencrypted integration tests passed with Python 3.13 in the spark-dev-313 conda environment.
- Ruff lint, Ruff format checking, git diff checking, line-length scanning, and non-ASCII scanning passed.
The encrypted variant could not complete locally because the environment crashes in the OpenSSL FIPS self-test. The existing pickle-only BroadcastTest.test_broadcast_with_encryption reproduces the identical crash, so this is not caused by the Arrow serialization path.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][PYTHON] Optimize PySpark broadcast serialization with Arrow [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on PR #57781:
URL: https://github.com/apache/spark/pull/57781#issuecomment-5190148250
Benchmark results
Environment: conda spark-dev-313, Python 3.13.12, PyArrow 24.0.0, ASV with 3 repeats. End-to-end cases use local[4], 4 input rows, and 4 partitions so broadcast work dominates.
End-to-end broadcast of a native PyArrow table
| Path | Broadcast rows | Pickle | Arrow |
| --- | ---: | ---: | ---: |
| RDD | 100k | 41.7 +/- 2 ms | 40.5 +/- 2 ms |
| RDD | 1M | 121 +/- 2 ms | 117 +/- 2 ms |
| Arrow UDF | 100k | 69.3 +/- 4 ms | 78.2 +/- 4 ms |
| Arrow UDF | 1M | 150 +/- 8 ms | 142 +/- 5 ms |
The larger payload is about 44 MB. Arrow is approximately 3-5% faster there, although the difference is modest.
Isolated serde for a native PyArrow table, 1M rows
| Operation | Pickle | Arrow |
| --- | ---: | ---: |
| Driver serialization | 26.2 +/- 2 ms | 23.2 +/- 0.6 ms |
| Worker-path deserialization | 3.85 +/- 0.08 ms | 23.2 +/- 1 ms |
| Serialized size | 44 MB | 44 MB |
Native PyArrow values already have efficient pickle support, so this is not the main target case.
Isolated serde for the custom protocol example backed by list[float]
| Values | Operation | Pickle | Arrow |
| ---: | --- | ---: | ---: |
| 100k | Driver serialization | 1.44 +/- 0.04 ms | 1.65 +/- 0.1 ms |
| 100k | Worker-path deserialization | 2.21 +/- 0.09 ms | 0.136 +/- 0.04 ms |
| 100k | Serialized size | 900 KB | 800 KB |
| 1M | Driver serialization | 14.1 +/- 0.2 ms | 13.8 +/- 0.3 ms |
| 1M | Worker-path deserialization | 46.4 +/- 10 ms | 0.762 +/- 0.07 ms |
| 1M | Serialized size | 9.00 MB | 8.00 MB |
For 1M floats, combined serde falls from about 60.5 ms to 14.6 ms, approximately 4.2x faster. Retained worker payload memory falls from 32,289,080 bytes for the Python list and float objects to 8,000,000 bytes for the Arrow float64 buffer, approximately 75% less.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org