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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
