[WIP][PYTHON] Optimize PySpark broadcast serialization with Arrow

[zhengruifeng]

What changes were proposed in this pull request?

This PR adds an optional useArrow argument to SparkContext.broadcast in PySpark Classic. Arrow IPC is selected only when all of the following are true:

• useArrow is explicitly True.
• PyArrow is installed.
• The value is a native Arrow value or implements the complete Arrow broadcast protocol.

Native PyArrow Table, RecordBatch, Array, and ChunkedArray values are supported directly.

A custom class must implement from_arrow as a class or static method and provide either to_arrow, arrow_c_stream, arrow_c_array, or arrow_array. The serialized payload contains small pickled class metadata followed by the Arrow IPC stream so the worker can invoke from_arrow.

If useArrow is None or False, PyArrow is unavailable, or the value does not implement the complete protocol, the existing pickle path is used. The public Broadcast.dump, Broadcast.load, and Broadcast.load_from_path methods remain pickle-specific.

Why are the changes needed?

Arrow-compatible Python objects already have a typed columnar representation. This opt-in protocol lets applications avoid generic object serialization and reconstruction for large Arrow-native state while retaining pickle as the compatibility path for arbitrary Python classes.

Explicit opt-in avoids changing serialization behavior for existing applications and lets custom classes control both conversion and reconstruction.

Does this PR introduce any user-facing change?

Yes. SparkContext.broadcast accepts useArrow: Optional[bool] = None. With useArrow=True, supported values use Arrow IPC and retain their native or custom reconstructed type on the driver and workers. The default behavior remains pickle.

How was this patch tested?

All Arrow broadcast tests are isolated in python/pyspark/tests/test_arrow_broadcast.py and registered in dev/sparktestsupport/modules.py under pyspark-core.

• Added codec tests for native Arrow values, to_arrow/from_arrow, the Arrow C stream protocol, incomplete-protocol fallback, and missing-PyArrow fallback.
• Added driver and worker integration tests for native and custom Arrow values.
• Added tests proving that useArrow=None and useArrow=False use pickle.
• 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)

[zhengruifeng]

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.