ez-parallel 0.1.4

Easy Parallel Multiprocessing

ez-parallel

Easy Parallel Multiprocessing

Installation

pip install -U ez-parallel

or install with Poetry

poetry add ez-parallel

Usage

• Process a list of items by using parallel workers
  • Define what a worker does
  • Define how to iterate through the data
  • Just run
• Display a global progress bar
• Does the same for multithread

Multithread vs Multiprocessing

In multiprocessing, new processes will be launched, they won't share memory. The user should implement a way to store the results of a worker and gather these results when multiprocess() returns.

With multithreading, new threads will be launched, they all share the memory of the parent process. This also restricts the runtime to a single CPU-core, as threads from a process do not get allocated to different cores. There will be no performance improvement when the distributed work is CPU-bound.

How to choose? (guidelines)

• CPU-heavy (data transformation, data preprocessing): multiprocessing.
• IO-heavy (DB requests, File I/O): multithreading.

Examples

How to process a list?

import time

from ez_parallel import list_iterator, queue_worker, multiprocess

@queue_worker
def work_one_thing(x: int) -> int:
  # do something
  a = x + 2
  time.sleep(0.1)
  
  # Worked on ONE thing = return 1
  return 1

# Data
things_to_process = list(range(1000000))

# Create the iterator over the things to process
iter_fn, nb_things = list_iterator(things_to_process)

# Process all the things in parallel with 20 processes
multiprocess(
  worker_fn=work_one_thing,
  input_iterator_fn=iter_fn,
  total=nb_things,
  nb_workers=20,
  description=str('Process the things')
)

How to Process a list by batch?

import time
from typing import List

from ez_parallel import batch_iterator_from_sliceable, queue_worker, multiprocess


@queue_worker
def work_one_thing(x: List[int]) -> int:
  # do something
  a = [y + 2 for y in x]
  time.sleep(0.1)
  
  # Worked on ONE thing = return 1
  return len(x)

# Data
things_to_process = list(range(1000000))

# Create the iterator over the things to process
# This will yield batches of 128 things
iter_fn = batch_iterator_from_sliceable(items=things_to_process, batch_size=128)
nb_things = len(things_to_process)

# Process all the things in parallel with 20 processes
multiprocess(
  worker_fn=work_one_thing,
  input_iterator_fn=iter_fn,
  total=nb_things,
  nb_workers=20,
  description='Process the things'
)

How to Process a list by batch in multithread?

import time
from typing import List

from ez_parallel import batch_iterator_from_sliceable, queue_worker, multithread


@queue_worker
def work_one_thing(x: List[int]) -> int:
  # do something
  a = [y + 2 for y in x]
  time.sleep(0.1)
  
  # Worked on ONE thing = return 1
  return len(x)

# Data
things_to_process = list(range(1000000))

# Create the iterator over the things to process
# This will yield batches of 128 things
iter_fn = batch_iterator_from_sliceable(items=things_to_process, batch_size=128)
nb_things = len(things_to_process)

# Process all the things in parallel with 20 processes
multithread(
  worker_fn=work_one_thing,
  input_iterator_fn=iter_fn,
  total=nb_things,
  nb_workers=20,
  description='Process the things'
)

How to collect results in multiprocessing?

(Suggestion using temporary files) In this scenario, results are recorded as JSONL files, the final result is the concatenation of all files.

import glob
import json
import os
import random
import shutil
import string
import tempfile
from typing import List

from ez_parallel import batch_iterator_from_sliceable, queue_worker, multithread


def random_file_name() -> str:
  return ''.join(random.choices(string.ascii_letters, k=32))  


# Tmp folder
tmp_folder = os.path.join(tempfile.gettempdir(), random_file_name())
os.makedirs(tmp_folder)

@queue_worker
def work_one_thing(x: List[int]) -> int:
  # do something
  tmp_file = os.path.join(tmp_folder, random_file_name())
  with open(tmp_file, 'w') as out:
    for number in x:
      out.write(json.dumps({"number": number, "square": number ** 2}) + '\n')
  
  # Worked on ONE thing = return 1
  return len(x)

# Data
things_to_process = list(range(1000000))

# Create the iterator over the things to process
# This will yield batches of 128 things
iter_fn = batch_iterator_from_sliceable(items=things_to_process, batch_size=128)
nb_things = len(things_to_process)

# Process all the things in parallel with 20 processes
multithread(
  worker_fn=work_one_thing,
  input_iterator_fn=iter_fn,
  total=nb_things,
  nb_workers=20,
  description='Process the things'
)


# concatenate all the temporary files
with open('final_results.jsonl', 'w') as out:
  for f in glob.glob(os.path.join(tmp_folder, '*')):
    with open(f) as src:
      out.write(src.read())  

# Delete temporary files / folder
shutil.rmtree(tmp_folder)      

How to collection results in multithreading

A lot easier and straightforward, because all the threads share the same memory.

from typing import List

from ez_parallel import batch_iterator, queue_worker, multithread

# List are threadsafe in Python
results = []

@queue_worker
def work_one_thing(x: List[int]) -> int:
  # do something
  results.extend({"number": y, "square": y ** 2} for y in x)
  
  # Worked on ONE thing = return 1
  return len(x)

# Data
things_to_process = list(range(1000000))

# Create the iterator over the things to process
# This will yield batches of 128 things
iter_fn, nb_things = batch_iterator(items=things_to_process, batch_size=128)

# Process all the things in parallel with 20 processes
multithread(
  worker_fn=work_one_thing,
  input_iterator_fn=iter_fn,
  total=nb_things,
  nb_workers=20,
  description='Process the things'
)

print(len(results))

🛡 License

This project is licensed under the terms of the MIT license. See LICENSE for more details.

📃 Citation

@misc{ez-parallel,
  author = {Julien Rossi},
  title = {Easy Parallel Multiprocessing},
  year = {2021},
  publisher = {GitHub},
  journal = {GitHub repository},
  howpublished = {\url{https://github.com/j-rossi-nl/ez-parallel}}
}

Credits

This project was generated with python-package-template.