parallelpipe

parallelpipe is a pipeline parallelization library for Python.

A pipeline is composed by one or more stages. Each stage take the output of the previous stage as an input and performs some operations on it like map, filter, reduce, etc. This is an extension of the normal producer/consumer pattern where we can have multiple stages. Every stage receives the input data in a queue and push the results to another queue that is connected with the next stage.

In this example we define a stage function that takes as an input an iterator returning urls and return the corresponding content after downloading it:

fromparallelpipeimportstageimportrequests@stage(workers=4)deffetch_urls(urls):
forurlinurls:
result=requests.get(url)
yieldresult.content

To use this stage just run

urls= ['http://test.com', ...]
pipe=urls|fetch_urlsforcontentinpipe.results():
print(len(content))

We built a basic pipe with only one stage. This stage have 4 workers that will start processing in parallel the input urls. The main process will receive the downloaded content as soon as one of them is available and print the corresponding length. Notice that a pipeline input can be any iterable; this will be automatically wrapped into a stage.

Let's say we are interested into the title string inside the HTML content. We can add another stage to do that:

importreRE_TITLE=re.compile("<title>(.*?)</title>", re.M)
@stage(workers=2)defget_titles(contents):
forcontentincontents:
match=RE_TITLE.search(content)
ifmatchisnotNone:
yieldmatch.group(1)
pipe=urls|fetch_urls|get_titlesfortitleinpipe.results():
print(title)

Again this second stage will start processing the content as soon as it is available and yield his output. Notice that also this task is parallelized, since we set workers to 2. As you can see this stage is not an exact map, since the number of titles returned could be less then the number of document (we check for the presence of the title tag).

Let's add now one more stage to return the most common title:

fromcollectionsimportCounter@stage()defmost_common(titles):
commons=Counter(titles).most_common(1)
yieldcommons[0]
pipe=urls|fetch_urls|get_titles|most_commonprint(pipe.execute())

To calculate the most common title we need to aggregate all results, so we can use only one worker. We also use pipe.execute() instead of pipe.results() because we know only one result will be returned.

Parametrical stage

@stage(workers=4)defadd_n(input, n):
fornumberininput:
yieldnumber+npipe=range(100) |add_n(7)
forresultinpipe.results():
print(result)

In this example our stage function not only require the input iterator but also one or more extra parameter to perform his computation. In the moment we build our pipeline we can configure this extra paramerters simply calling the stage with them as input. Remember, all parameters can be passed except for the first one that is the mandatory input iterator.

Mapping stage

If your stage do pure mapping, i.e. it return exactly one result for every input element you can simplify your code using the map_stage decorator:

fromparallelpipeimportmap_stage@map_stage(workers=4)defadd_n(number, n):
returnnumber+n

Queue Size

When you build a stage you can define how big his output queue is. Setting an output queue limit can be useful if the current stage can produce much faster then how the following stage can consume. In this case, once the queue size is reached the stage stop processing his input and wait for the consumer to free a slot.

# only 30 elements can queue in output before blocking this stage@stage(workers=4, qsize=30)defadd_n(input, n):
fornumberininput:
yieldnumber+n

By default qsize=0 that means the queue have no limit.

Setup a stage

Setup the stage queue and workers can also be done after defining the stage calling the setup() method.

add_n.setup(workers=2, qsize=0)

Use the Stage class directly

So far we built stages using decorators on functions, but we can also use the Stage class directly:

fromparallelpipeimportStagedefadd_n(input, n):
fornumberininput:
yieldnumber+npipe=Stage(range, 10) |Stage(add_n, 5)

As you can see in the previous example the Stage class take as input an iterator function and any extra parameters needed by it. The first stage is a producer so will not be called with any input iterator. When we use the stage class explictly we can use setup() to configure how many workers we need and the queue size:

pipe=Stage(range, 10).setup(qsize=5) |Stage(add_n, 5).setup(workers=2)

The setup() method return the stage itself, so we can set it up during the pipeline definition.

Exception handling

During the execution of your stage function an exception can occur. When a stage detects an exception it will automatically consume and ignore all the input from the previous stage and then a TaskException will be throw on the main process.

@stage(workers=2)defadd_one(numbers):
fornumberinnumbers:
yieldnumber+1

>>>pipe= [2, 3, "ops", 7] |add_one>>>print(sum(pipe.results()))
Processadd_one-0:
Traceback (mostrecentcalllast):
File"/Users/gt/miniconda2/lib/python2.7/multiprocessing/process.py", line258, in_bootstrapself.run()
File"/Users/gt/Desktop/code/parallelpipe/parallelpipe.py", line67, inrunforiteminres:
File"example.py", line7, inadd_oneyieldnumber+1TypeError: cannotconcatenate'str'and'int'objectsTraceback (mostrecentcalllast):
File"example.py", line10, in<module>print(sum(pipe.results()))
File"/Users/gt/Desktop/code/parallelpipe/parallelpipe.py", line249, inresultsraiseTaskException(msg)
parallelpipe.TaskException: Thetask"add_one-0"raisedTypeError("cannot concatenate 'str' and 'int' objects",)

If you want to avoid that a single bad input blocks your pipeline you can of course catch any exception inside the stage function so that the pipeline can continue and produce the rest of the results.