Multiprocessing in Python. Work smarter not harder. Dive into the… | by JoseTomasLavinLeon | Dec, 2023

Picture this: you’ve purchased to teach an ML model and should fetch positive variables for a sample of 500k observations. You’ve purchased a Python course of that fetches these variables for one comment, nonetheless it’s helpful useful resource consuming. Why? On account of it entails fairly a couple of queries to various databases, and you then undoubtedly’ve purchased to course of the knowledge after extraction. Let’s say your full course of for one comment takes 1 second. Now, if we run a for loop to get the knowledge for all 500k, we’re 500,000 seconds — that’s 8333.33 minutes = 138.88 hours = 5.7 days. And that’s assuming the tactic is working 24/7. Realistically, we’d be a steady week at least 😵😵. How can we avoid burning a whole week on this? Enter Multiprocessing 🚀🚀

This Python library permits us to type out various duties concurrently, benefiting from our machine’s processing power. Instead of working one exercise at a time sequentially, multiprocessing lets us divide the workload into unbiased processes that run in parallel, slashing the complete execution time significantly.

“The[Pool] object presents a helpful strategy of parallelizing the execution of a function all through various enter values, distributing the enter info all through processes (info parallelism)”

Let’s see the best way it really works with an occasion. Picture this: we’ve purchased an inventory of 1000 devices that we have now to course of, and there’s a function to course of each merchandise/comment (processing_item). We’re in a position to simulate a function that takes about 1 second, a random value obtained from a uniform distribution.

import random
import time
#Devices we have now to course of
info = report(range(1, 1001))
#Processing function
def processing_item(merchandise):
# Simulate merchandise processing
processing_time = random.uniform(0.5, 1)
final result = merchandise + processing_time
time.sleep(processing_time)
return final result

If we course of each merchandise in a for loop and retailer it in an inventory, how prolonged will it take?

list_results = []
start_time = time.time()
for merchandise in info:
list_results.append(processing_item(merchandise))
end_time = time.time()
total_time = end_time - start_time
print(f'Processing time: {total_time} seconds')
>>> Processing ti 515.6806468963623 seconds

Now, let’s use multiprocessing. For that, let’s take the Pool() methodology. With it, you’ll have the ability to apply capabilities to datasets in parallel, which is especially useful for compute-intensive duties. The Pool() methodology robotically handles the responsibility process to obtainable processes, managing concurrency and communication between them.

We’ll create a function that iterates over a little bit of issues (which can seemingly be in our pool) and applies our processing_itemfunction to each of them.

import time
import logging
logging.basicConfig(diploma=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
def process_data_chunk(chunk):
start_time = time.time()
outcomes = []
# Procesar cada elemento en el chunk
for merchandise in chunk:
outcomes.append(processing_item(merchandise))
end_time = time.time()
elapsed_time = end_time - start_time
logging.info(f"Chunk processed in {elapsed_time} seconds")
return outcomes

Let’s uncover completely completely different capabilities we’re ready to make use of with this library, their execs, cons, and the situations they take for this exercise:

A. map:

• Ordered: map maintains the order of results in the similar order as a result of the enter parts. If order is crucial and likewise you need results in the similar order they’ve been despatched, map is the appropriate various.
• Blocking: map is obstructing, which suggests it waits for all outcomes to complete sooner than persevering with with code execution.

chunk_size = 10
data_chunks = [data[i:i + chunk_size] for i in range(0, len(info), chunk_size)]
# Map processing
total_start_time_map = time.time()
with multiprocessing.Pool() as pool:
map_results = pool.map(process_data_chunk, data_chunks)
total_elapsed_time_map = time.time() - total_start_time_map
logging.info(f"Complete time using map: {total_elapsed_time_map} seconds")
>>> Complete time using map: 80.26290893554688 seconds

B. imap:

• Iterative: imap (iterative map) presents outcomes iteratively as processes full. That’s helpful for memory effectivity, as outcomes aren’t all saved in memory sooner than being delivered.
• Non-blocking: In distinction to map, imap doesn’t anticipate all outcomes to complete sooner than persevering with. This allows starting to work with outcomes as rapidly as they’re obtainable, doubtlessly bettering time effectivity.

chunk_size = 10
data_chunks = [data[i:i + chunk_size] for i in range(0, len(info), chunk_size)]
# Imap course of
total_start_time_imap = time.time()
with multiprocessing.Pool() as pool:
imap_results = pool.imap(process_data_chunk, data_chunks)
imap_results_list = report(imap_results)
total_elapsed_time_imap = time.time() - total_start_time_imap
logging.info(f"Complete time using imap: {total_elapsed_time_imap} seconds")
>>> Complete time using imap: 64.96913719177246 seconds

C. imap_unordered:

• No Guarantee of Order: Similar to imap, imap_unordered presents outcomes iteratively and is non-blocking. However, in distinction to imap, it doesn’t guarantee that outcomes protect the distinctive order of enter parts. If final result order is just not important, and effectivity is a priority, imap_unordered is more likely to be a additional atmosphere pleasant alternative.
• Most Parallelization: By not worrying about final result order, imap_unordered can further parallelize processes, as a result of it doesn’t wish to attend for one course of to complete sooner than starting the next.

chunk_size = 10
data_chunks = [data[i:i + chunk_size] for i in range(0, len(info), chunk_size)]
# Imap unordered processing
total_start_time_imap = time.time()
with multiprocessing.Pool() as pool:
imap_results = pool.imap_unordered(process_data_chunk, data_chunks)
imap_unordered_results_list = report(imap_results)
# Calcular el tiempo full para imap
total_elapsed_time_imap = time.time() - total_start_time_imap
logging.info(f"Complete time using imap_unordered: {total_elapsed_time_imap} seconds")
>>> Complete time using imap_unordered: 66.17712831497192 seconds

If we alter the chunk_size to 5? 100? 500?

#CHUNK SIZE = 5
2023-11-24 22:06:50,715 - INFO - Complete time using for loop: 507.21000385284424 seconds
2023-11-24 22:08:00,830 - INFO - Complete time using map: 70.11264514923096 seconds
2023-11-24 22:09:05,105 - INFO - Complete time using imap: 64.27440094947815 seconds
2023-11-24 22:10:09,506 - INFO - Complete time using imap_unordered: 64.39984011650085 seconds
#CHUNK SIZE = 100
2023-11-24 20:43:32,620 - INFO - Complete time using for loop: 496.25063395500183 seconds
2023-11-24 20:45:08,686 - INFO - Complete time using map: 96.06432795524597 seconds
2023-11-24 20:46:48,598 - INFO - Complete time using imap: 99.91018414497375 seconds
2023-11-24 20:48:26,368 - INFO - Complete time using imap_unordered: 97.7699191570282 seconds
#CHUNK SIZE = 500
2023-11-24 21:44:45,745 - INFO - Complete time using for loop: 490.71753096580505 seconds
2023-11-24 21:48:55,715 - INFO - Complete time using map: 249.96965885162354 seconds
2023-11-24 21:53:10,435 - INFO - Complete time using imap: 254.71954774856567 seconds
2023-11-24 21:57:27,536 - INFO - Complete time using imap_unordered: 257.09930396080017 seconds

Having fun with spherical with the chunk_size reveals that we’re in a position to receive significantly completely completely different effectivity, making it an important challenge to consider. Plainly the additional granular the chunk_size, the a lot much less time it requires.

Given what now we have now seen, when the order whereby you might receive the knowledge isn’t associated, there are two important advantages for using **imap**:

1. Faster processing, additional throughput.
2. As a result of it’s non-blocking, it may possibly prevent processed knowledge as you go together with imap, whereas using map requires prepared in your full course of to finish sooner than saving. A extremely prolonged course of may crash, and with map, you’ll have the ability to’t save your progress midway.

Multiprocessing is a sturdy software program which will undoubtedly stop various days of processing. In my explicit case at work, we went from a 12-day processing timeframe for our choices to 2 days using imap_unordered—a whopping 6x low cost! Lastly, keep in mind the truth that this parallelization can put a giant load on the databases housing your info. I prefer to suggest monitoring the processing, doing it steadily, and shutting down connections after a sure amount to avoid overwhelming the databases. Hope the article proves helpful!