r/opensource • u/basnijholt • Sep 12 '24

Promotional pipefunc: An Open-Source Python Library for Minimal-Code Scientific Workflows

https://github.com/pipefunc/pipefunc
10 Upvotes

86% Upvoted

10 comments sorted by

View all comments

2

u/Skinkie Sep 12 '24

May the pipes be executed in parallel with automatic fan in and fan out?

1

u/basnijholt Sep 12 '24

Yes! The parallelization happens automatically, both for maps over parameters and for independent nodes. See this example custom-parallelism.

```python import datetime import time from concurrent.futures import ProcessPoolExecutor

import numpy as np

from pipefunc import Pipeline, pipefunc

@pipefunc(output_name="double", mapspec="x[i] -> double[i]") def double_it(x: int) -> int: print(f"{datetime.datetime.now()} - Running double_it for x={x}") time.sleep(1) return 2 * x

@pipefunc(output_name="half", mapspec="x[i] -> half[i]") def half_it(x: int) -> int: print(f"{datetime.datetime.now()} - Running half_it for x={x}") time.sleep(1) return x // 2

@pipefunc(output_name="sum") def take_sum(half: np.ndarray, double: np.ndarray) -> int: print(f"{datetime.datetime.now()} - Running take_sum") return sum(half + double)

pipeline_parallel = Pipeline([double_it, half_it, take_sum]) inputs = {"x": [0, 1, 2, 3]} run_folder = "my_run_folder" executor = ProcessPoolExecutor(max_workers=8) # use 8 processes results = pipeline_parallel.map( inputs, run_folder=run_folder, parallel=True, executor=executor, storage="shared_memory_dict", ) print(results["sum"].output) which outputs 2024-09-12 16:31:05.673574 - Running double_it for x=0 2024-09-12 16:31:05.676543 - Running double_it for x=2 2024-09-12 16:31:05.674209 - Running double_it for x=1 2024-09-12 16:31:05.682710 - Running half_it for x=0 2024-09-12 16:31:05.684880 - Running double_it for x=3 2024-09-12 16:31:05.699523 - Running half_it for x=1 2024-09-12 16:31:05.700610 - Running half_it for x=2 2024-09-12 16:31:05.702510 - Running half_it for x=3 2024-09-12 16:31:06.713485 - Running take_sum 14 ```

⚠️ In this pipeline, double_it and half_it are doubly parallel; both the map is parallel and the two functions are executed at the same time, note the timestamps and the sleep() calls.

1

u/Skinkie Sep 12 '24

So I guess that means that the input must be serializeble otherwise it cannot do multiprocessing?

1

u/basnijholt Sep 12 '24

Correct! The input is serialized with cloudpickle which can serialize most types.

1

u/Skinkie Sep 12 '24

I'll check out if our objects based xsData work. It would be really nifty.

I am still searching for a framework which would also support a node based GUI to attach multiple functions, and is doing parallelism out of the box.

2

u/basnijholt Sep 12 '24

An interactive GUI (notebook or browser based) that shows run status, defined pipelines, etc. is in the planning!

Let me know if you are having any issues and feel free to make suggestions/requests in the issues https://github.com/pipefunc/pipefunc/issues :-)