Handling Multidimensional Pandas Dataframe

[harmojd]

MPIRE Developers,

Is MPIRE capable of working with a multidimensional Pandas dataframe for each process?

If so, can you please provide an example of how to accomplish this?

Thanks for all your help!

[sybrenjansen]

Can you be a bit more specific to what you're trying to accomplish? Does each process need its own dataframe? Do multiple processes need to access the same dataframe? Are you only reading from it, or also writing?

Please add as much info as possible so I can best answer your question.

[bleninr]

Yes.

Let df be a pandas dataframe with the following columns: id, col_1, col_2, col_3, col_4, and col_5. In this frame, each id has a certain number of rows and there are 10 ids.

Then the following

series = [x[1] for x in df.groupby("id").__iter__()]
with WorkerPool(n_jobs=ncpus) as pool:
    results = pool.map(do_fcst, series, progress_bar=True)

gives the error
Exception occurred, terminating ... : 0%| | 0/10 [00:00<?, ?it/s]
Exception: Exception occurred in Worker-0 with the following arguments:
Arg 0: 'id'
Arg 1: 'col_1'
Arg 2: 'col_2'
Arg 3: 'col_3'
Arg 4: 'col_4'
Arg 5: 'col_5'
TypeError: do_fcst() takes 1 positional argument but 6 were given.

The function do_fcst expects a pandas dataframe with 6 columns.

When we use multiprocessing package as follows:

from multiprocessing import Pool
with Pool(ncpus) as p:
    results = list(tqdm(p.imap(do_fcst, series), total=len(series)))

it works well, but we are looking for ways to reduce the time complexity.

Hope it helps!

Please let us know if you need more details.

Thank you.

[harmojd]

MPIRE Developers,

My teammate provided the additional details above.

Please let us know if you need more details.

Thanks for all your help!

[sybrenjansen]

By default MPIRE applies argument unpacking. I.e., it defaults to starmap. That is your problem here.

Luckily, you can easily avoid that. Please have a look at https://sybrenjansen.github.io/mpire/usage/map/map.html#circumvent-argument-unpacking, which should provide you with all the details you need.

It basically comes down to changing this line slightly:

series = [x[1] for x in df.groupby("id").__iter__()]

to:

series = [(x[1],) for x in df.groupby("id").__iter__()]

And then it should work

[bleninr]

Perfect! It worked! Appreciate it greatly!

Thank you.