PyPeln: Concurrent Data Pipelines

0.4.9 · active · verified Sun Apr 12

PyPeln (pronounced "pypeline") is a Python library designed for building concurrent data pipelines with ease. It offers a simple, functional API that supports multiprocessing (processes), multithreading (threads), and asynchronous programming (asyncio tasks) with the same interface. This allows developers to create multi-stage pipelines and maintain fine-grained control over computational resources. The library is currently at version 0.4.9 and provides solutions for medium-scale data tasks where frameworks like Spark or Dask might be considered overkill.

Warnings

breaking The `maxsize` argument was removed from all `from_iterable` functions in version 0.4.0.
Fix: Review `from_iterable` calls and remove the `maxsize` argument if present. The functionality for `maxsize` was later introduced for `to_stage` and `to_iterable`, and `ordered` functions in 0.4.6.
gotcha Python 3.6 versions between 0.4.0 and 0.4.1 experienced import errors due to reliance on `typing.Protocol` (introduced in Python 3.8). Although the library generally targets Python 3.6+, the `typing.Protocol` dependency broke compatibility in these specific versions.
Fix: Upgrade to PyPeln version 0.4.2 or newer, which includes conditional dependencies and imports to correctly support Python >= 3.6.2. Ensure `typing_extensions` is installed.
gotcha The `task` module, which provides `asyncio` based pipelines, is only available and fully supported for Python versions 3.7 and above, despite the overall library supporting Python >= 3.6.2.
Fix: When using `pl.task` for asynchronous operations, ensure your Python environment is 3.7 or newer. For older Python 3.6 environments, use `pl.process` or `pl.thread` instead.
gotcha The `maxsize` parameter in `pl.process.map` and `pl.thread.map` was not correctly respected, potentially leading to unbounded queues and out-of-memory issues in earlier versions.
Fix: Upgrade to PyPeln version 0.4.7 or newer to ensure `maxsize` properly functions as a backpressure mechanism, preventing upstream stages from overwhelming downstream ones.
gotcha Process workers using multiprocessing start method 'spawn' could raise an `AttributeError` on certain systems.
Fix: Upgrade to PyPeln version 0.4.9. This issue was specifically addressed in this release.

Install

pip install pypeln Install stable version

Imports

pypeln
```
import pypeln as pl
```
Standard convention for importing the library.
pypeln.process
```
from pypeln import process as pl_process
```
While 'pr' was seen in older discussions, 'pl_process' or directly using 'pl.process' is clearer and more idiomatic.
pypeln.thread
```
from pypeln import thread as pl_thread
```
Similar to 'process', prefer a more descriptive alias or direct access via 'pl.thread'.
pypeln.task
```
from pypeln import task as pl_task
```
Similar to 'process', prefer a more descriptive alias or direct access via 'pl.task'. The alias 'io' was considered but 'task' is the module's name.

Quickstart

This quickstart demonstrates a basic multiprocessing pipeline. It defines a source iterable (`data`), then chains `map` and `filter` operations using `pl.process`. The pipeline squares numbers, filters for even ones, and prints each processed item. The `workers` and `maxsize` parameters control concurrency and backpressure for each stage. The `list()` call at the end triggers the execution and collects results.

import pypeln as pl
import time
import random

def slow_square(x):
    time.sleep(random.uniform(0.1, 0.5)) # Simulate work
    return x * x

def is_even(x):
    return x % 2 == 0

def print_item(x):
    print(f"Processing: {x}")
    return x

data = range(10) # [0, 1, ..., 9]

# Build a multiprocessing pipeline
# 1. Map: square each number (4 workers)
# 2. Filter: keep only even numbers (2 workers)
# 3. Map: print the item (1 worker, ordered output)
results = (pl.process.map(slow_square, data, workers=4, maxsize=4) 
           .filter(is_even, workers=2, maxsize=2) 
           .map(print_item, workers=1, maxsize=1, ordered=True))

# Consume the results (this starts the pipeline execution)
final_list = list(results)
print(f"Final result: {final_list}")

view raw JSON →