Add from_partitioned to create Dataset from any data source implements partitioned

[kira-lin]

Why are these changes needed?

We intend to propose a protocol to make large, distributed, partitioned data exchange between frameworks(like ray, modin, dask) easier. Several PRs are in progress, please check here.

It's also possible to do for ray dataset, but to start with, MLDataset is simpler.

Related issue number

Checks

• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[ericl]

MLDataset is deprecated, so I don't think we should be accepting patches to it.

[kira-lin]

Yes, I can do this for ray dataset. I just want to first discuss about this protocol. @fschlimb

[ericl]

Sounds good. Btw, the integration point for Datasets would be to define a custom datasource (e.g., PartitionedDatasource or similar), via the datasource API: https://github.com/ray-project/ray/blob/master/python/ray/data/datasource/datasource.py

[kira-lin]

I did not use Datasource API for a few reasons. 1. data can be in ray object store before from_partitioned 2. we want to utilize data's locality given by location in the partitions.

[kira-lin]

I did not use Datasource API for a few reasons:

1. data can be in ray object store before from_partitioned
2. we want to utilize data's locality given by location in the partitions.

[fschlimb]

I did not use Datasource API for a few reasons:

1. data can be in ray object store before from_partitioned
2. we want to utilize data's locality given by location in the partitions.

Why is that not possible with Datasource?

[fschlimb]

Broader discussion started here: data-apis/consortium-feedback#7

[ericl]

Why is that not possible with Datasource?

I believe both of these are possible with Datasource. For locality, Ray will internally manage locality-aware execution, the use of node: labels is not recommended since it interferes with auto-scaling and fault tolerance.

[kira-lin]

The data could be in dask, that make things hard. Anyway, I'll update it once the protocol settles

[ericl]

If the data is in Dask on Ray, then locality scheduling will apply to those objects. We don't support Dask unless it's run via Dask on Ray.

Alternatively, we could modify the data source API to allow custom read tasks to be generated (e.g., a PartitionedDataSource could generate read tasks that run on specific nodes according to locality).

Can you let me know if one of the above alternatives works?

[ericl]

This should be refactored into a PartitionedDataSource.

[fschlimb]

Why is that not possible with Datasource?

I believe both of these are possible with Datasource. For locality, Ray will internally manage locality-aware execution, the use of node: labels is not recommended since it interferes with auto-scaling and fault tolerance.

Yes, this is generally understood, and as long as we consider "native" ray objects only, this will work just fine without anything special. If we consider data which comes from somewhere else (say, like "dask" or "YetOnotherFancyFamework") the protocol allows us to manually guarantee locality - probably using label:node would be most appropriate - if only for tasks which put the data into ray space. The protocol tries to allow this kind of interoperability without requiring consumers to necessarily support all frameworks.

Notice: ray limits the possibilities to support proper zero-copy when consuming non-ray objects. But at least we can avoid data transfer between nodes.

[bveeramani]

‼️ ACTION REQUIRED ‼️

We've switched our code formatter from YAPF to Black (see #21311).

To prevent issues with merging your code, here's what you'll need to do:

1. Install Black

pip install -I black==21.12b0

2. Format changed files with Black

curl -o format-changed.sh https://gist.githubusercontent.com/bveeramani/42ef0e9e387b755a8a735b084af976f2/raw/7631276790765d555c423b8db2b679fd957b984a/format-changed.sh
chmod +x ./format-changed.sh
./format-changed.sh
rm format-changed.sh

3. Commit your changes.

git add --all
git commit -m "Format Python code with Black"

4. Merge master into your branch.

git pull upstream master

5. Resolve merge conflicts (if necessary).

After running these steps, you'll have the updated format.sh.

[stale]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 14 days if no further activity occurs. Thank you for your contributions.

• If you'd like to keep this open, just leave any comment, and the stale label will be removed.

[stale]

Hi again! The issue will be closed because there has been no more activity in the 14 days since the last message.

Please feel free to reopen or open a new issue if you'd still like it to be addressed.

Again, you can always ask for help on our discussion forum or Ray's public slack channel.

Thanks again for opening the issue!