You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "timsaucer (via GitHub)" <gi...@apache.org> on 2026/05/15 18:30:16 UTC
[PR] docs: user guide + runnable examples for distributing expressions (4/4) [datafusion-python]
timsaucer opened a new pull request, #1547:
URL: https://github.com/apache/datafusion-python/pull/1547
# Which issue does this PR close?
No associated issue. **PR 4 of 4** stacked on [#1546](https://github.com/apache/datafusion-python/pull/1546). The diff against \`main\` is cumulative until the prior PRs merge — review the commits on \`pr4-docs-examples\` directly for the PR4 delta.
# Rationale for this change
PRs 1-3 close the round-trip for Python UDFs and add the toggle that controls it. None of that is discoverable without user-facing documentation. This PR ships the user guide page that explains the multiprocessing / Ray / datafusion-distributed patterns, the runnable examples, and the centralized Security section that nails down what the toggle does and does not protect against.
# What changes are included in this PR?
**User guide.** \`docs/source/user-guide/io/distributing_work.rst\` is the new canonical page. It walks through:
- The shipped-expression model (what travels inline vs by name).
- Worker setup (\`datafusion.ipc.set_worker_ctx\`).
- Sender-side configuration (\`datafusion.ipc.set_sender_ctx\` and \`SessionContext.with_python_udf_inlining\`).
- A Security section that is the single source of truth for the cloudpickle / \`pickle.loads\` threat model.
- Pointers to the runnable examples and \`datafusion-distributed\`.
\`docs/source/user-guide/io/index.rst\` gets the toctree entry.
**Runnable examples.**
- \`examples/multiprocessing_pickle_expr.py\` — \`Pool.map\` of a closure-capturing UDF across processes, with the worker initializer wiring the worker context. The closure carries non-trivial state to demonstrate that captured state survives the round-trip.
- \`examples/ray_pickle_expr.py\` — Ray actor analogue.
- \`examples/datafusion-ffi-example/python/tests/_test_pickle_strict_ffi.py\` — strict-mode refusal exercised end-to-end against an FFI capsule scalar UDF. Kept under the FFI example crate because it needs that crate's compiled artifacts. The leading \`_\` keeps pytest from auto-collecting it as a unit test; the FFI example's own test harness runs it explicitly.
- \`examples/README.md\` picks up index entries for the new files.
**Docstring centralization.** Three docstrings previously carried near-duplicate copies of the pickle / cloudpickle security warning. Reduced each to a one-line summary plus a pointer to the Security section so there's a single canonical home for the threat model:
- \`PythonLogicalCodec::with_python_udf_inlining\` rustdoc.
- \`SessionContext.with_python_udf_inlining\` docstring.
- \`datafusion.ipc\` module docstring.
The crate-level \`codec.rs\` module rustdoc also updates "pure-Python scalar UDFs" to "scalar / aggregate / window UDFs" now that PR 2 has shipped agg + window inline.
# Are there any user-facing changes?
Docs and examples only. No code behavior changes, no new public APIs.
\`api change\` not added.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] docs: user guide + runnable examples for distributing expressions [datafusion-python]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer merged PR #1547:
URL: https://github.com/apache/datafusion-python/pull/1547
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] docs: user guide + runnable examples for distributing expressions [datafusion-python]
Posted by "ntjohnson1 (via GitHub)" <gi...@apache.org>.
ntjohnson1 commented on code in PR #1547:
URL: https://github.com/apache/datafusion-python/pull/1547#discussion_r3318827450
##########
docs/source/user-guide/io/distributing_work.rst:
##########
@@ -0,0 +1,391 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+.. or more contributor license agreements. See the NOTICE file
+.. distributed with this work for additional information
+.. regarding copyright ownership. The ASF licenses this file
+.. to you under the Apache License, Version 2.0 (the
+.. "License"); you may not use this file except in compliance
+.. with the License. You may obtain a copy of the License at
+
+.. http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+.. software distributed under the License is distributed on an
+.. "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+.. KIND, either express or implied. See the License for the
+.. specific language governing permissions and limitations
+.. under the License.
+
+Distributing DataFusion work
+============================
+
+Splitting a DataFusion workload across multiple processes — for
+throughput, isolation, or to use a worker pool — comes in a few
+different shapes depending on what is being split.
+
+* **Expression-level distribution** ✅ *supported today*. The driver
+ builds a DataFusion :py:class:`~datafusion.Expr`, sends it to
+ worker processes, and each worker evaluates the expression against
+ its own slice of data. Suits embarrassingly-parallel workloads
+ where the driver decides up front how to partition.
+* **Query-level distribution via datafusion-distributed** 🚧 *work in
+ progress upstream*. A single logical / physical plan is split into
+ stages and run across worker nodes. The driver writes one SQL or
+ DataFrame query; the runtime decides partitioning.
+* **Query-level distribution via Apache Ballista** 🚧 *work in
+ progress upstream*. Similar query-level model, with a more
+ cluster-management-oriented runtime.
+
+Only the first option is ready for use from datafusion-python today.
+The other two are documented below so the surrounding story is in
+one place; integration details will land here as those projects
+become usable from datafusion-python.
+
+Expression-level distribution
+-----------------------------
+
+DataFusion expressions support distribution directly: pass one to a
+worker process and Python's standard
+`pickle <https://docs.python.org/3/library/pickle.html>`_ machinery
+serializes it transparently — the same machinery
+:py:meth:`multiprocessing.pool.Pool.map`, Ray's ``@ray.remote``, and
+similar libraries already use to ship function arguments. Python UDFs
+— scalar, aggregate, and window — travel inside the serialized
+expression; the receiver does not need to pre-register them.
+
+Basic worker-pool example
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Define a worker function that takes the expression plus a batch and
+returns the evaluated result:
+
+.. code-block:: python
+
+ import pyarrow as pa
+ from datafusion import SessionContext
+
+
+ def evaluate(expr, batch):
+ # `expr` arrived here via the pool's automatic pickling —
+ # no manual serialization needed in user code.
+ ctx = SessionContext()
+ df = ctx.from_pydict({"a": batch})
+ return df.with_column("result", expr).select("result").to_pydict()["result"]
+
+Then build the expression in the driver and fan it out:
+
+.. code-block:: python
+
+ import multiprocessing as mp
+ from datafusion import col, udf
+
+ double = udf(
+ lambda arr: pa.array([(v.as_py() or 0) * 2 for v in arr]),
+ [pa.int64()], pa.int64(), volatility="immutable", name="double",
+ )
+ expr = double(col("a"))
+
+ mp_ctx = mp.get_context("forkserver")
+ with mp_ctx.Pool(processes=4) as pool:
+ results = pool.starmap(
+ evaluate,
+ [(expr, [1, 2, 3]), (expr, [10, 20, 30])],
+ )
+ print(results) # [[2, 4, 6], [20, 40, 60]]
Review Comment:
NIT: I think technically if you use the `>>>` doctest format it can run and verify this result even in rst
##########
docs/source/user-guide/io/distributing_work.rst:
##########
@@ -0,0 +1,391 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+.. or more contributor license agreements. See the NOTICE file
+.. distributed with this work for additional information
+.. regarding copyright ownership. The ASF licenses this file
+.. to you under the Apache License, Version 2.0 (the
+.. "License"); you may not use this file except in compliance
+.. with the License. You may obtain a copy of the License at
+
+.. http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+.. software distributed under the License is distributed on an
+.. "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+.. KIND, either express or implied. See the License for the
+.. specific language governing permissions and limitations
+.. under the License.
+
+Distributing DataFusion work
+============================
+
+Splitting a DataFusion workload across multiple processes — for
+throughput, isolation, or to use a worker pool — comes in a few
+different shapes depending on what is being split.
+
+* **Expression-level distribution** ✅ *supported today*. The driver
+ builds a DataFusion :py:class:`~datafusion.Expr`, sends it to
+ worker processes, and each worker evaluates the expression against
+ its own slice of data. Suits embarrassingly-parallel workloads
+ where the driver decides up front how to partition.
+* **Query-level distribution via datafusion-distributed** 🚧 *work in
+ progress upstream*. A single logical / physical plan is split into
+ stages and run across worker nodes. The driver writes one SQL or
+ DataFrame query; the runtime decides partitioning.
+* **Query-level distribution via Apache Ballista** 🚧 *work in
+ progress upstream*. Similar query-level model, with a more
+ cluster-management-oriented runtime.
+
+Only the first option is ready for use from datafusion-python today.
+The other two are documented below so the surrounding story is in
+one place; integration details will land here as those projects
+become usable from datafusion-python.
+
+Expression-level distribution
+-----------------------------
+
+DataFusion expressions support distribution directly: pass one to a
+worker process and Python's standard
+`pickle <https://docs.python.org/3/library/pickle.html>`_ machinery
+serializes it transparently — the same machinery
+:py:meth:`multiprocessing.pool.Pool.map`, Ray's ``@ray.remote``, and
+similar libraries already use to ship function arguments. Python UDFs
+— scalar, aggregate, and window — travel inside the serialized
+expression; the receiver does not need to pre-register them.
+
+Basic worker-pool example
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Define a worker function that takes the expression plus a batch and
+returns the evaluated result:
+
+.. code-block:: python
+
+ import pyarrow as pa
+ from datafusion import SessionContext
+
+
+ def evaluate(expr, batch):
+ # `expr` arrived here via the pool's automatic pickling —
+ # no manual serialization needed in user code.
+ ctx = SessionContext()
+ df = ctx.from_pydict({"a": batch})
+ return df.with_column("result", expr).select("result").to_pydict()["result"]
+
+Then build the expression in the driver and fan it out:
+
+.. code-block:: python
+
+ import multiprocessing as mp
+ from datafusion import col, udf
+
+ double = udf(
+ lambda arr: pa.array([(v.as_py() or 0) * 2 for v in arr]),
+ [pa.int64()], pa.int64(), volatility="immutable", name="double",
+ )
+ expr = double(col("a"))
+
+ mp_ctx = mp.get_context("forkserver")
+ with mp_ctx.Pool(processes=4) as pool:
+ results = pool.starmap(
+ evaluate,
+ [(expr, [1, 2, 3]), (expr, [10, 20, 30])],
+ )
+ print(results) # [[2, 4, 6], [20, 40, 60]]
+
+When saved to a ``.py`` file and executed with the ``spawn`` or
Review Comment:
NIT I don't know if you use a standard note call out elsewhere in the docs. This is also mostly instructions related to multiprocessing in python rather than datafusion specific so maybe an external link to those docs for those unfamiliar.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org