Nice work! There is a gap when it comes to writing single-machine, concurrent CPU-bound python code. Ray is too big, pykka is threads only, builtins are poorly abstracted. The syntax is also very nice!<p>But I'm not sure I can use this even though I have a specific use-case that feels like it would work well (high-performance pure Python downloading from cloud object storage). The examples are a bit too simple and I don't understand how I can do more complicated things.<p>I chunk up my work, run it in parallel and then I need to do a fan-in step to reduce my chunks - how do you do that in Pyper?<p>Can the processes have state? Pure functions are nice, but if I'm reaching for multiprocess, I need performance and if I need performance, I'll often want a cache of some sort (I don't want to pickle and re-instantiate a cloud client every time I download some bytes for instance).<p>How do exceptions work? Observability? Logs/prints?<p>Then there's stuff that is probably asking too much from this project, but I get it if I write my own python pipeline so it matters to me - rate limiting WIP, cancellation, progress bars.<p>But if some of these problems are/were solved and it offers an easy way to use multiprocessing in python, I would probably use it!
Great feedback, thank you. We'll certainly be working on adding more examples to illustrate more complex use cases.<p>One thing I'd mention is that we don't really imagine Pyper as a whole observability and orchestration platform. It's really a package for writing Python functions and executing them concurrently, in a flexible pattern that can be integrated with other tools.<p>For example, I'm personally a fan of Prefect as an observability platform-- you could define pipelines in Pyper then wrap it in a Prefect flow for orchestration logic.<p>Exception handling and logging can also be handled by orchestration tools (or in the business logic if appropriate, literally using try... except...)<p>For a simple progress bar, tqdm is probably the first thing to try. As it wraps anything iterable, applying it to a pipeline might look like:<p><pre><code> import time
from pyper import task
from tqdm import tqdm
@task(branch=True)
def func(limit: int):
for i in range(limit):
time.sleep(0.1)
yield i
def main():
for _ in tqdm(func(limit=20), total=20):
pass
if __name__ == "__main__":
main()</code></pre>
> <i>I don't want to pickle and re-instantiate a cloud client every time I download some bytes for instance</i><p>Have you tried multiprocessing.shared_memory to address this?
I haven't played with that much! This isn't really a problem in general for my approach to writing this sort of code - when I use multiprocessing, I use a Process class or a worker task function with a setup step followed by a while loop that pulls from a work/control queue. But in the Pyper functional programming world, it would be a concern.<p>IIRC multiprocessing.shared_memory is a much more low-level of abstraction than most python stuff, so I think I'd need to figure out how to make the client use the shared memory and I'm not sure if I could.
Do you really need to reinvent the wheel every time for parallel workloads? Just learn GNU parallel and write single-threaded code.<p>Concurrency in general isn't about parallelism. It's just about doing multiple things at the same time.
GNU Parallel is really neat, software that's so good it's boring. Closing in on being a quarter century old by now, no? I remember first reading about it in 2003 maybe?<p>I've also used 'fork in Picolisp a lot for this kind of thing, and also Elixir, which arguably has much nicer pipes.<p>But hey, it's good that Python after like thirty years or so is trying to get decent concurrency. Eventually people that use it as a first language might learn about such things too.
You really should dive more into the `multiprocess` support option and highlight how this gets around issues with the GIL. This feels like a major value add, and "does this help with CPU-bound work" being "yes" is a big deal!<p>I don't really need pipelining that much, but pipelining along with a certain level of durability and easy multiprocessing support? Now we're talking
Lowkey I hate the "\" line continuation in Python to force PEP-8 compliance in a way... Is there any Pythonistas who would write the examples in there differently to achieve a similar level of readability?<p>> pipeline = task(get_data, branch=True) \<p>> | task(step1, workers=20) \<p>> | task(step2, workers=20) \<p>> | task(step3, workers=20, multiprocess=True)
I really don’t like overloading pipes like this. I would rather chain methods like how the django orm does it.<p>you could reassign every line, but it would look nicer with chained functions.<p><pre><code> pipeline = task(get_data, branch=True)
pipeline = pipeline | task(step1, workers=20)
pipeline = pipeline | task(step2, workers=20)
pipeline = pipeline | task(step3, workers=20, multiprocess=True)
</code></pre>
edit:<p>I would be tempted to do something like this:<p><pre><code> steps = [task(step1, workers=20),
task(step2, workers=20),
task(step3, workers=20, multiprocess=True)]
pipeline = task(get_data, branch=True)
for step in steps:
pipeline = pipeline.__or__(step)</code></pre>
According to the docs, | is syntactic sugar for the .pipe method.<p><pre><code> pipeline = task(get_data, branch=True).pipe(
task(step1, workers=20)).pipe(
task(step2, workers=20)).pipe(
task(step3, workers=20, multiprocess=True))
</code></pre>
That's probably the chained method approach for those with this preference.
This style looks pretty good to me:<p><pre><code> pipeline = task(...)
pipeline |= task(...)
</code></pre>
So does this style:<p><pre><code> steps = [task(...), task(...)]
pipeline = functools.reduce(operator.or_, steps)
</code></pre>
But it appears you can just change "task" to "Task" and then:<p><pre><code> pipeline = pyper.Pipeline([Task(...), Task(...)])</code></pre>
Wrap the statement in (...) and you can drop the backslashes. See also how people split up complicated loc queries in pandas on multiple lines too.
How does this compare to <a href="https://github.com/svenkreiss/pysparkling">https://github.com/svenkreiss/pysparkling</a>?
Cool, I was just looking for something like this!<p>It's surprisingly annoying in built-in python to do something like this. The most recent thing I was trying to do was:<p>- load URLs from a file
- hand them out to one subprocess per cpu
- download them concurrently in threads or async within each subprocess
- pull the results back into a single process for formatting and storing<p>Getting this to work and handle queues, ctrl-c, exceptions etc. is just a whole mess involving python builtins created at different times with different interfaces; I hacked until I kind of got it working, but didn't love it. Bundling it all in a single tested package would be great.
This is cool - I’ve been looking for something like this. I really liked the syntax of Prefect v1 but it was overcomplicated with execution configuration in subsequent versions. I just want something to help me just run async pipelines and prevent AsyncIO weirdness - going to test this out.
From just a short skimming of the docs:<p>- my biggest issue with concurrency in python (especially with asyncio) is leaking tasks. Pyper should provide structured concurrency support a-la trio.<p>- I don't see the opposite of branch to collect the output of multiple sub pipelines into a single stage. I need this pretty much always and it is a chore to implement.<p>- Async need not force the full pipeline to be async. There should be an option to run async funcitons in background event loops. Especially as you already support threaded executions.
Your third point intrigues me a lot. I imagine for the majority of cases, it's generally more useful to work with async functions in the structure of async syntax, but I suppose it's possible to run async functions in a synchronous pipeline.<p>Even though there's currently no built-in support for this, a workaround could be to just define synchronous helper functions to handle running your async logic in an event loop.
asyncio.TaskGroup from stdlib provides structured concurrency
Nice! I'm looking forward to trying it out. This seems very similar to <a href="https://github.com/cgarciae/pypeln/">https://github.com/cgarciae/pypeln/</a>
Very good README.md, teasing and explaining very well the the provided value. Well done!
I still gotta wrap my head around concurrency, I'm not sure if using threads count for concurrency
Very simple and elegant API!
[dead]