BluePyParallel: Bluebrain Python Embarrassingly Parallel library

Provides an embarrassingly parallel tool with sql backend.

Introduction

Provides an embarrassingly parallel tool with sql backend, inspired by BluePyMM of @wvangeit.

Installation

This package should be installed using pip:

bash pip install bluepyparallel

Usage

General computation

```python

factoryname = "multiprocessing" # Can also be None, dask or ipyparallel batchsize = 10 # This value is used to split the data into batches before processing them chunk_size = 1000 # This value is used to gather the elements to process before sending them to the workers

Setup the parallel factory

parallelfactory = initparallelfactory( factoryname, batchsize=batchsize, chunksize=chunksize, processes=4, # This parameter is specific to the multiprocessing factory )

Get the mapper from the factory

mapper = parallelfactory.getmapper()

Use the mapper to map the given function to each element of mapped_data and gather the results

result = sorted(mapper(function, mappeddata, *functionargs, **function_kwargs)) ```

Working with Pandas

This library provides a specific function working with large :class:pandas.DataFrame: :func:bluepyparallel.evaluator.evaluate. This function converts the DataFrame into a list of dict (one for each row), then maps a given function to element and finally gathers the results.

Example:

```python input_df = pd.DataFrame(index=[1, 2], columns=['data'], data=[100, 200])

def evaluationfunction(row): result1, result2 = computesomething(row['data']) return {'newcolumn1': result1, 'newcolumns2': result2}

Use the mapper to map the given function to each element of the DataFrame

resultdf = evaluate( inputdf, # This is the DataFrame to process evaluationfunction, # This is the function that should be applied to each row of the DataFrame parallelfactory="multiprocessing", # This could also be a Factory previously defined newcolumns=[['newcolumn1', 0], ['newcolumns2', None]], # this defines default values for columns ) assert resultdf.columns == ['data', 'newcolumns1', 'newcolumns2'] `` It is in a way a generalisation of the pandas.apply` method.

Working with an SQL backend

As it aims at working with time consuming functions, it also provides a checkpoint and resume mechanism using a SQL backend. The SQL backend uses the SQLAlchemy library, so it can work with a large variety of database types (like SQLite, PostgreSQL, MySQL, ...). To activate this feature, just pass a URL that can be processed by SQLAlchemy to the db_url parameter of :func:bluepyparallel.evaluator.evaluate.

.. note:: A specific driver might have to be installed to access the database (like psycopg2 <https://www.psycopg.org/docs/>_ for PostgreSQL for example).

Example:

```python

Use the mapper to map the given function to each element of the DataFrame

resultdf = evaluate( inputdf, # This is the DataFrame to process evaluationfunction, # This is the function that should be applied to each row of the DataFrame parallelfactory="multiprocessing", # This could also be a Factory previously defined db_url="sqlite:///db.sql", # This could also just be "db.sql" and would be automatically turned to SQLite URL ) ```

Now, if the computation crashed for any reason, the partial result is stored in the db.sql file. If the crash was due to an external cause (therefore executing the code again should work), it is possible to resume the computation from the last computed element. Thus, only the missing elements are computed, which can save a lot of time.

Running with distributed Dask MPI on HPC systems

This is an example of a sbatch script that can be adapted to execute the script using multiple nodes and workers with distributed dask and MPI. In this example, the code called by the run.py should be parallelized using BluePyParallel.

Dask variables are not strictly required, but highly recommended, and they can be fine tuned.

```bash

!/bin/bash -l

Dask configuration

export DASKDISTRIBUTEDLOGGINGDISTRIBUTED="info" export DASKDISTRIBUTEDWORKERUSEFILELOCKING=False export DASKDISTRIBUTEDWORKERMEMORYTARGET=False # don't spill to disk export DASKDISTRIBUTEDWORKERMEMORYSPILL=False # don't spill to disk export DASK_DISTRIBUTEDWORKERMEMORYPAUSE=0.80 # pause execution at 80% memory use export DASKDISTRIBUTEDWORKERMEMORYTERMINATE=0.95 # restart the worker at 95% use export DASKDISTRIBUTEDWORKERMULTIPROCESSINGMETHOD=spawn export DASKDISTRIBUTEDWORKERDAEMON=True

Reduce dask profile memory usage/leak (see https://github.com/dask/distributed/issues/4091)

export DASKDISTRIBUTEDWORKERPROFILEINTERVAL=10000ms # Time between statistical profiling queries export DASKDISTRIBUTEDWORKERPROFILE__CYCLE=1000000ms # Time between starting new profile

Split tasks to avoid some dask errors (e.g. Event loop was unresponsive in Worker)

export PARALLELBATCHSIZE=1000

srun -v run.py ```

To ensure only the evaluate function is run with parallel dask, one has to initialise the parallel factory before anything else is done in the code. For example, run.py could look like:

python if __name__ == "__main__": parallel_factory = init_parallel_factory('dask_dataframe') df = pd.read_csv("inuput_data.csv") df = some_preprocessing(df) df = evaluate(df, function_to_evaluate, parallel_factory=parallel_factory) df.to_csv("output_data.csv")

This is because everything before init_parallel_factory will be run in parallel, as mpi is not initialized yet.

.. note:: We recommend to use dask_dataframe instead of dask, as it is in practice more stable for large computations.

Funding & Acknowledgment

The development of this software was supported by funding to the Blue Brain Project, a research center of the École polytechnique fédérale de Lausanne (EPFL), from the Swiss government’s ETH Board of the Swiss Federal Institutes of Technology.

For license and authors, see LICENSE.txt and AUTHORS.md respectively.

Copyright © 2023-2024 Blue Brain Project/EPFL