Row Partition API

[richardliaw]

Is your feature request related to a problem? Please describe.
A clear and concise description of what the problem is. What kind of performance improvements would you like to see with this new API?

I'd like to be able to obtain a list of row partitions from a distributed Modin dataframe.

This would allow me to easily do distributed data-parallel training by feeding in these dataframe shards. This would be useful for both XGBoost and Horovod.

cc @tgaddair

[devin-petersohn]

Thanks @richardliaw for the request!

We should consider a developer API for cases like this that enable access to Modin internal partitions without needing to dig into them. I have been thinking about this some.

Design can look something like this:

from modin.developer.partitioning import extract_row_partitions, extract_col_partitions
import ray

row_oid_list = extract_row_partitions(df, assert_type=ray.ObjectRef)
col_oid_list =  extract_col_partitions(df, assert_type=ray.ObjectRef)

Thoughts @richardliaw?

cc @aregm

[fschlimb]

I ran into a similar issue.
In addition to being able to extract the partitions it would be good to know the locality of each partition. Knowing where a partition (currently) resides can be used to minimize data movement when passing partitions to ray actors. Minimal data movement requires pairing actors and partitions which are 'pinned' to the same node.

[richardliaw]

@devin-petersohn sounds good!

[devin-petersohn]

I have a prototype ready to try. Install via pip install git+https://github.com/devin-petersohn/modin@feat/2342

Usage is simple:

from modin.developer import unwrap_row_partitions

unwrap_row_partitions(df)

You will get a list of ray.ObjectRef each contain a row partition.

[fschlimb]

Thanks, works great for me!
What about getting sufficient information to be able to map data to local actors?

[fschlimb]

How would this work with modin on dask?

[YarShev]

Thanks, works great for me!
What about getting sufficient information to be able to map data to local actors?

Maybe it would be good to have some kind of RowPartitionsActor and ColPartitionsActor that will keep row partitions and column partitions, respectively, assigned to them. Also, these actors should keep some info on node ip address (it is supposed an actor per node).

[YarShev]

How would this work with modin on dask?

It will work the same way as for ray, but you will get a list of Dask.Future instead. If I am not correct, @devin-petersohn , please, correct me.

[devin-petersohn]

It will work the same, if you are expecting a certain engine, there is a keyword argument in the API for the engine. It will make sure that the objects are compatible with that engine or throw an error if they are not.