Scribd
Upload
23 views16 pages

Python Parallel Processing Guide

This document provides a tutorial on parallel processing in Python using the multiprocessing module. It discusses key concepts like synchronous vs asynchronous execution and demonstrates how to parallelize functions using multiprocessing.Pool objects. Specifically, it shows how to parallelize a function that counts the number of values within a given range for each row in a 2D array/list. The function is parallelized using Pool.apply(), Pool.map(), Pool.starmap(), and asynchronously using Pool.apply_async(). Overall, the document serves as a practical guide for leveraging parallel processing in Python to improve performance.

Uploaded by

Reaper Grim
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as DOCX, PDF, TXT or read online on Scribd
23 views16 pages

Python Parallel Processing Guide

This document provides a tutorial on parallel processing in Python using the multiprocessing module. It discusses key concepts like synchronous vs asynchronous execution and demonstrates how to parallelize functions using multiprocessing.Pool objects. Specifically, it shows how to parallelize a function that counts the number of values within a given range for each row in a 2D array/list. The function is parallelized using Pool.apply(), Pool.map(), Pool.starmap(), and asynchronously using Pool.apply_async(). Overall, the document serves as a practical guide for leveraging parallel processing in Python to improve performance.

Uploaded by

Reaper Grim
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as DOCX, PDF, TXT or read online on Scribd

Parallel Processing in Python – A

Practical Guide with Examples


by Selva Prabhakaran |Posted on

FacebookTwitterWhatsAppLinkedInRedditGoogle BookmarksShare
Parallel processing is a mode of operation where the task is executed
simultaneously in multiple processors in the same computer. It is meant to
reduce the overall processing time. In this tutorial, you’ll understand the
procedure to parallelize any typical logic using python’s multiprocessing
module.

Contents
1. Introduction
2. How many maximum parallel processes can you run?
3. What is Synchronous and Asynchronous execution?
4. Problem Statement: Count how many numbers exist between a given range in
each row
Solution without parallelization
5. How to parallelize any function?
6. Asynchronous Parallel Processing
7. How to Parallelize a Pandas DataFrame?
8. Exercises
9. Conclusion

1. Introduction
Parallel processing is a mode of operation where the task is executed
simultaneously in multiple processors in the same computer. It is meant to
reduce the overall processing time.
However, there is usually a bit of overhead when communicating between
processes which can actually increase the overall time taken for small tasks
instead of decreasing it.
In python, the multiprocessing module is used to run independent parallel processes
by using subprocesses (instead of threads). It allows you to leverage multiple
processors on a machine (both Windows and Unix), which means, the processes
can be run in completely separate memory locations.
By the end of this tutorial you would know:
1. How to structure the code and understand the syntax to enable parallel
processing using multiprocessing ?
2. How to implement synchronous and asynchronous parallel processing?
3. How to parallelize a Pandas DataFrame?
4. Solve 3 different usecases with the [Link]() interface.

2. How many maximum parallel processes can


you run?
The maximum number of processes you can run at a time is limited by the
number of processors in your computer. If you don’t know how many processors
are present in the machine, the cpu_count() function in multiprocessing will show it.
import multiprocessing as mp

print("Number of processors: ", mp.cpu_count())

3. What is Synchronous and Asynchronous


execution?
In parallel processing, there are two types of execution: Synchronous and
Asynchronous.
A synchronous execution is one the processes are completed in the same order in
which it was started. This is achieved by locking the main program until the
respective processes are finished.
Asynchronous, on the other hand, doesn’t involve locking. As a result, the order
of results can get mixed up but usually gets done quicker.
There are 2 main objects in multiprocessing to implement parallel execution of a
function: The Pool Class and the Process Class.
1. Pool Class
1. Synchronous execution
 [Link]() and [Link]()
 [Link]()
2. Asynchronous execution
 Pool.map_async() and Pool.starmap_async()
 Pool.apply_async() )
2. Process Class
Let’s take up a typical problem and implement parallelization using the above
techniques. In this tutorial, we stick to the Pool class, because it is most
convenient to use and serves most common practical applications.
4. Problem Statement: Count how many
numbers exist between a given range in each
row
The first problem is: Given a 2D matrix (or list of lists), count how many
numbers are present between a given range in each row. We will work on the list
prepared below.
import numpy as np

from time import time

# Prepare data

[Link](100)

arr = [Link](0, 10, size=[200000, 5])

data = [Link]()

data[:5]

Solution without parallelization


Let’s see how long it takes to compute it without parallelization. For this, we
iterate the function howmany_within_range() (written below) to check how many
numbers lie within range and returns the count.
# Solution Without Paralleization

def howmany_within_range(row, minimum, maximum):

"""Returns how many numbers lie within `maximum` and `minimum` in a given
`row`"""

count = 0

for n in row:
if minimum <= n <= maximum:

count = count + 1

return count

results = []

for row in data:

[Link](howmany_within_range(row, minimum=4, maximum=8))

print(results[:10])

#> [3, 1, 4, 4, 4, 2, 1, 1, 3, 3]

5. How to parallelize any function?


The general way to parallelize any operation is to take a particular function that
should be run multiple times and make it run parallelly in different processors.
To do this, you initialize a Pool with n number of processors and pass the
function you want to parallelize to one of Pool s parallization methods.
[Link]() provides the apply() , map() and starmap() methods to make any
function run in parallel.
Nice! So what’s the difference between apply() and map() ?
Both apply and map take the function to be parallelized as the main argument. But
the difference is, apply() takes an args argument that accepts the parameters passed
to the ‘function-to-be-parallelized’ as an argument, whereas, map can take only
one iterable as an argument.
So, map() is really more suitable for simpler iterable operations but does the job
faster.
We will get to starmap() once we see how to
parallelize howmany_within_range() function with apply() and map() .

5.1. Parallelizing using [Link]()


Let’s parallelize the howmany_within_range() function using [Link]() .
# Parallelizing using [Link]()
import multiprocessing as mp

# Step 1: Init [Link]()

pool = [Link](mp.cpu_count())

# Step 2: `[Link]` the `howmany_within_range()`

results = [[Link](howmany_within_range, args=(row, 4, 8)) for row in data]

# Step 3: Don't forget to close

[Link]()

print(results[:10])

#> [3, 1, 4, 4, 4, 2, 1, 1, 3, 3]

5.2. Parallelizing using [Link]()


[Link]() accepts only one iterable as argument. So as a workaround, I modify
the howmany_within_range function by setting a default to
the minimum and maximum parameters to create a
new howmany_within_range_rowonly() function so it accetps only an iterable list of
rows as input. I know this is not a nice usecase of map() , but it clearly shows how
it differs from apply() .
# Parallelizing using [Link]()

import multiprocessing as mp

# Redefine, with only 1 mandatory argument.


def howmany_within_range_rowonly(row, minimum=4, maximum=8):

count = 0

for n in row:

if minimum <= n <= maximum:

count = count + 1

return count

pool = [Link](mp.cpu_count())

results = [Link](howmany_within_range_rowonly, [row for row in data])

[Link]()

print(results[:10])

#> [3, 1, 4, 4, 4, 2, 1, 1, 3, 3]

5.3. Parallelizing using [Link]()


In previous example, we have to redefine howmany_within_range function to make
couple of parameters to take default values. Using starmap() , you can avoid doing
this. How you ask?
Like [Link]() , [Link]() also accepts only one iterable as argument, but
in starmap() , each element in that iterable is also a iterable. You can to provide the
arguments to the ‘function-to-be-parallelized’ in the same order in this inner
iterable element, will in turn be unpacked during execution.
So effectively, [Link]() is like a version of [Link]() that accepts arguments.
# Parallelizing with [Link]()

import multiprocessing as mp
pool = [Link](mp.cpu_count())

results = [Link](howmany_within_range, [(row, 4, 8) for row in data])

[Link]()

print(results[:10])

#> [3, 1, 4, 4, 4, 2, 1, 1, 3, 3]

6. Asynchronous Parallel Processing


The asynchronous equivalents apply_async() , map_async() and starmap_async() lets you
do execute the processes in parallel asynchronously, that is the next process can
start as soon as previous one gets over without regard for the starting order. As a
result, there is no guarantee that the result will be in the same order as the input.

6.1 Parallelizing with Pool.apply_async()


apply_async()is very similar to apply() except that you need to provide a callback
function that tells how the computed results should be stored.
However, a caveat with apply_async() is, the order of numbers in the result gets
jumbled up indicating the processes did not complete in the order it was started.
A workaround for this is, we redefine a new howmany_within_range2() to accept and
return the iteration number (i) as well and then sort the final results.
# Parallel processing with Pool.apply_async()

import multiprocessing as mp

pool = [Link](mp.cpu_count())
results = []

# Step 1: Redefine, to accept `i`, the iteration number

def howmany_within_range2(i, row, minimum, maximum):

"""Returns how many numbers lie within `maximum` and `minimum` in a given
`row`"""

count = 0

for n in row:

if minimum <= n <= maximum:

count = count + 1

return (i, count)

# Step 2: Define callback function to collect the output in `results`

def collect_result(result):

global results

[Link](result)

# Step 3: Use loop to parallelize

for i, row in enumerate(data):

pool.apply_async(howmany_within_range2, args=(i, row, 4, 8),


callback=collect_result)
# Step 4: Close Pool and let all the processes complete

[Link]()

[Link]() # postpones the execution of next line of code until all


processes in the queue are done.

# Step 5: Sort results [OPTIONAL]

[Link](key=lambda x: x[0])

results_final = [r for i, r in results]

print(results_final[:10])

#> [3, 1, 4, 4, 4, 2, 1, 1, 3, 3]

It is possible to use apply_async() without providing a callback function. Only that, if


you don’t provide a callback, then you get a list of [Link] objects which
contains the computed output values from each process. From this, you need to
use the [Link]() method to retrieve the desired final result.
# Parallel processing with Pool.apply_async() without callback function

import multiprocessing as mp

pool = [Link](mp.cpu_count())

results = []

# call apply_async() without callback

result_objects = [pool.apply_async(howmany_within_range2, args=(i, row, 4, 8))


for i, row in enumerate(data)]
# result_objects is a list of [Link] objects

results = [[Link]()[1] for r in result_objects]

[Link]()

[Link]()

print(results[:10])

#> [3, 1, 4, 4, 4, 2, 1, 1, 3, 3]

6.2 Parallelizing with Pool.starmap_async()


You saw how apply_async() works. Can you imagine and write up an equivalent version
for starmap_async and map_async? The implementation is below anyways.
# Parallelizing with Pool.starmap_async()

import multiprocessing as mp

pool = [Link](mp.cpu_count())

results = []

results = pool.starmap_async(howmany_within_range2, [(i, row, 4, 8) for i, row


in enumerate(data)]).get()

# With map, use `howmany_within_range_rowonly` instead

# results = pool.map_async(howmany_within_range_rowonly, [row for row in


data]).get()
[Link]()

print(results[:10])

#> [3, 1, 4, 4, 4, 2, 1, 1, 3, 3]

7. How to Parallelize a Pandas DataFrame?


So far you’ve seen how to parallelize a function by making it work on lists.
But when working in data analysis or machine learning projects, you might want
to parallelize Pandas Dataframes, which are the most commonly used objects
(besides numpy arrays) to store tabular data.
When it comes to parallelizing a DataFrame , you can make the function-to-be-
parallelized to take as an input parameter:
 one row of the dataframe
 one column of the dataframe
 the entire dataframe itself
The first 2 can be done using multiprocessing module itself. But for the last one,
that is parallelizing on an entire dataframe, we will use the pathos package that
uses dill for serialization internally.
First, lets create a sample dataframe and see how to do row-wise and column-
wise paralleization. Something like using [Link]() on a user defined function but
in parallel.
import numpy as np

import pandas as pd

import multiprocessing as mp

df = [Link]([Link](3, 10, size=[5, 2]))

print([Link]())

#> 0 1

#> 0 8 5

#> 1 5 3

#> 2 3 4
#> 3 4 4

#> 4 7 9

We have a dataframe. Let’s apply the hypotenuse function on each row, but
running 4 processes at a time.
To do this, we exploit the [Link](name=False) . By setting name=False , you are
passing each row of the dataframe as a simple tuple to the hypotenuse function.
# Row wise Operation

def hypotenuse(row):

return round(row[1]**2 + row[2]**2, 2)**0.5

with [Link](4) as pool:

result = [Link](hypotenuse, [Link](name=False), chunksize=10)

output = [round(x, 2) for x in result]

print(output)

#> [9.43, 5.83, 5.0, 5.66, 11.4]

That was an example of row-wise parallelization. Let’s also do a column-wise


parallelization. For this, I use [Link]() to pass an entire column as a series to
the sum_of_squares function.
# Column wise Operation

def sum_of_squares(column):

return sum([i**2 for i in column[1]])

with [Link](2) as pool:

result = [Link](sum_of_squares, [Link](), chunksize=10)

output = [x for x in result]


print(output)

#> [163, 147]

Now comes the third part – Parallelizing a function that accepts a Pandas
Dataframe, NumPy Array, etc. Pathos follows the multiprocessing style of: Pool >
Map > Close > Join > Clear. Check out the pathos docs for more info.
import numpy as np

import pandas as pd

import multiprocessing as mp

from [Link] import ProcessingPool as Pool

df = [Link]([Link](3, 10, size=[500, 2]))

def func(df):

return [Link]

cores=mp.cpu_count()

df_split = np.array_split(df, cores, axis=0)

# create the multiprocessing pool

pool = Pool(cores)

# process the DataFrame by mapping function to each df across the pool


df_out = [Link]([Link](func, df_split))

# close down the pool and join

[Link]()

[Link]()

[Link]()

Thanks to notsoprocoder for this contribution based on pathos. If you are


familiar with pandas dataframes but want to get hands-on and master it, check
out these pandas exercises .

8. Exercises
Problem 1: Use [Link]() to get the row wise common items in list_a and list_b .
list_a = [[1, 2, 3], [5, 6, 7, 8], [10, 11, 12], [20, 21]]

list_b = [[2, 3, 4, 5], [6, 9, 10], [11, 12, 13, 14], [21, 24, 25]]

Show Solution
Problem 2: Use [Link]() to run the following python scripts in parallel.
Script names: ‘[Link]’, ‘[Link]’, ‘[Link]’
Show Solution
Problem 3: Normalize each row of 2d array (list) to vary between 0 and 1.
list_a = [[2, 3, 4, 5], [6, 9, 10, 12], [11, 12, 13, 14], [21, 24, 25, 26]]

Parallel Processing in Python


Last Updated: 27-12-2019

Parallel processing can increase the number of tasks done by your program which reduces the
overall processing time. These help to handle large scale problems.
In this section we will cover the following topics:
 Introduction to parallel processing
 Multi Processing Python library for parallel processing
 IPython parallel framework

Introduction to parallel processing


For parallelism, it is important to divide the problem into sub-units that do not depend on other
sub-units (or less dependent). A problem where the sub-units are totally independent of other
sub-units is called embarrassingly parallel.
For example, An element-wise operation on an array. In this case, the operation needs to aware
of the particular element it is handling at the moment.
In another scenario, a problem which is divided into sub-units have to share some data to
perform operations. These results in the performance issue because of the communication cost.

There are two main ways to handle parallel programs:


 Shared Memory
In shared memory, the sub-units can communicate with each other through the same
memory space. The advantage is that you don’t need to handle the communication
explicitly because this approach is sufficient to read or write from the shared memory.
But the problem arises when multiple process access and change the same memory
location at the same time. This conflict can be avoided using synchronization techniques.
 Distributed memory
In distributed memory, each process is totally separated and has its own memory space.
In this scenario, communication is handled explicitly between the processes. Since the
communication happens through a network interface, it is costlier compared to shared
memory.
Threads are one of the ways to achieve parallelism with shared memory. These are the
independent sub-tasks that originate from a process and share memory. Due to Global
Interpreter Lock (GIL) , threads can’t be used to increase performance in Python. GIL is a
mechanism in which Python interpreter design allow only one Python instruction to run at a
time. GIL limitation can be completely avoided by using processes instead of thread. Using
processes have few disadvantages such as less efficient inter-process communication than
shared memory, but it is more flexible and explicit.

Multiprocessing for parallel processing

Using the standard multiprocessing module, we can efficiently parallelize simple tasks by
creating child processes. This module provides an easy-to-use interface and contains a set of
utilities to handle task submission and synchronization.

Process and Pool Class

Process
By subclassing [Link], you can create a process that runs independently. By
extending the __init__ method you can initialize resource and by
implementing [Link]() method you can write the code for the subprocess. In the below
code, we see how to create a process which prints the assigned id:
To spawn the process, we need to initialize our Process object and
invoke [Link]() method. Here [Link]() will create a new process and will invoke
the [Link]() method.

Common questions

Powered by AI

Asynchronous processing is often preferred when tasks have varying completion times or when a quick response from partially completed tasks is desirable, as it allows for greater responsiveness and better utilization of resources . However, the trade-offs include a lack of control over the order of task completion and potential complexity in handling end results, due to out-of-order execution . Additionally, asynchronous processing generally requires managing callback functions or result objects, which can increase the complexity of code management compared to the more straightforward synchronous model where results are returned in a predictable sequence .

Python's Global Interpreter Lock (GIL) restricts threads in Python from executing multiple threads in parallel within the same process because it allows only one thread to execute at a time. To circumvent the GIL's limitations, the `multiprocessing` module uses separate memory spaces by running multiple processes instead of threads. This approach allows Python to achieve parallelism by creating independent subprocesses that do not share memory, entirely avoiding the constraints imposed by the GIL . While this introduces more overhead than threads due to separate memory spaces and communication needs, it enables genuine parallel execution .

Python's multiprocessing module can efficiently handle tasks like matrix operations, where independent computations can be run simultaneously to reduce processing time. For instance, counting numbers within a specific range across rows of a 2D matrix can be parallelized using multiprocessing techniques such as Pool.apply() for independent row operations . Similarly, DataFrames can be processed efficiently by applying functions to rows or columns in parallel, using techniques like `Pool.imap()` for processing subsets of data concurrently . Such parallelism significantly reduces computation time in data preprocessing or feature extraction tasks in large datasets, enhancing the performance of machine learning workflows .

Pool.apply() allows individual, synchronous process execution by accepting both a function and its arguments separately. It is useful when precise control over each function call is necessary . Pool.map() simplifies parallel processing by accepting only one iterable, consisting of inputs to the function being parallelized, and is suitable for tasks that can be expressed as a transformation over a sequence . Unlike apply() and map(), Pool.starmap() is appropriate for functions that require multiple arguments, as it can take an iterable of argument tuples, unpacking each tuple into multiple arguments during function execution . The choice between these methods depends on the nature of the operation being parallelized: use apply() for independent, non-iterable tasks, map() for operations over a single iterable, and starmap() when multiple arguments per operation are required .

`multiprocessing.Pool.map_async()` is designed for executing a function over an iterable of inputs asynchronously, where results might not be returned in the order they were started. This method allows capturing the result as a whole with the `get()` method when all inputs are processed, making it suitable for batch processing tasks where results order is not crucial . Conversely, `Pool.apply_async()` operates independently, allowing more granular control over individual function calls with optional callback functions. This method suits scenarios where immediate responses for each operation are needed, even before processing all items . Choosing between them depends on the requirement for sequence control and granularity of result handling .

'Embarrassingly parallel' problems are those that can be easily split into independent tasks, which neither rely on results from other tasks nor require inter-process communication. Examples in Python applications include element-wise operations on arrays or processing files independently in a dataset . The primary benefits include simplified implementation, as no synchronization or complex data exchange is necessary, leading to more efficient use of computational resources. In real-world scenarios, these tasks are appealing because they require minimal overhead and scale efficiently with increased computational resources, making them ideal for large-scale data processing and analysis tasks in fields such as scientific computing and big data environments .

The pathos package offers additional flexibility and functionality when parallelizing operations across entire Pandas Dataframes due to its use of the dill serialization library, which supports serialization of more complex Python objects than pickle, such as lambda functions, closures, and more complex class instances. This capability allows pathos to circumvent some limitations of Python's standard `multiprocessing` module, which uses pickle and can struggle with serializing complex data structures or functions. Moreover, pathos provides a more seamless interface for distributing DataFrame tasks across multiple processors effectively, allowing for easier scalability and handling of large datasets .

Synchronous execution in Python's multiprocessing ensures that processes complete in the order they were started, effectively locking the main program until the processes finish. This execution method utilizes Pool.map() and Pool.apply(), providing predictable sequencing of results. Asynchronous execution, on the other hand, allows processes to run independently of their start order, potentially completing quicker. Functions like Pool.map_async() and Pool.apply_async() are used for this purpose. The primary advantage is often improved overall performance due to shorter completion times, though the results may not be returned in a predictable order . Asynchronous execution can be beneficial in scenarios where task completion time varies, whereas synchronous execution might be favored for its ordered and predictable results .

Shared memory allows multiple processes to communicate by reading from and writing to the same memory space, which can simplify communication since it does not require explicit data transfer. However, this can lead to conflicts if multiple processes attempt to access and modify the same memory area simultaneously, requiring synchronization techniques such as locks to manage access . Distributed memory, by contrast, gives each process its own memory space, which avoids contention issues but necessitates explicit communication between processes, typically through network interfaces, making it slower and more complex to manage . The choice between the two approaches often hinges on the specific application needs, the size of data to be handled, and the complexity of communication required .

The number of processors affects the maximum parallel processes you can run, as it is generally limited by the number of available processors. In Python, the `multiprocessing.cpu_count()` function can be used to determine the number of processors available on a machine. This function returns the count of CPU cores, which can then inform how many processes might efficiently be run in parallel at a time . Running more processes than available CPU cores might lead to context switching and reduced performance .

You might also like