REFACTOR-#3853: interacting with Dask interface through 'DaskWrapper' class #3854
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
anmyachev
force-pushed
the
issue3853
branch
4 times, most recently
from
57203c5
to
31c6a0b
Compare
|
@anmyachev, that is what we were discussing earlier when restructuring, great changes! What about Ray? |
I plan to do the same for Ray, but on the separate PR. Is it ok for you? :) |
|
|
Codecov Report
@@ Coverage Diff @@
## master #3854 +/- ##
===========================================
+ Coverage 62.00% 89.90% +27.89%
===========================================
Files 214 214
Lines 17370 17359 -11
===========================================
+ Hits 10771 15607 +4836
+ Misses 6599 1752 -4847
📣 Codecov can now indicate which changes are the most critical in Pull Requests. Learn more |
|
Yes, that should be done in a separate PR. |
anmyachev
commented
Dec 15, 2021
|
|
||
| Returns | ||
| ------- | ||
| ray.ObjectRef or list | ||
| Ray identifier of the result being put to Plasma store. | ||
| """ | ||
| return deploy_ray_func.options(num_returns=num_returns).remote(func, kwargs) | ||
| return deploy_ray_func.options(num_returns=num_returns).remote( | ||
| func, *args, kwargs |
There was a problem hiding this comment.
It looks strange, for this to be so **kwargs, deploy_ray_func signature should be changed.
There was a problem hiding this comment.
Historically we have not been able to pass **kwargs and *args to a Ray task for implementation reasons. To get around this we pass kwargs as a dictionary.
There was a problem hiding this comment.
Do I understand correctly that such remote call did not work before?
>>> def t2(*args, **kwargs):
... print(args)
... print(kwargs)
...
>>> t2_rem = ray.remote(t2)
>>> t2_rem.remote("test_args", 2, 6, test_keyword="some value")
ObjectRef(4ee449587774c1f0ffffffffffffffffffffffff0100000001000000)
>>> (pid=28664) ('test_args', 2, 6)
(pid=28664) {'test_keyword': 'some value'}
| @@ -62,3 +67,22 @@ def materialize(cls, future): | |||
| """ | |||
| client = default_client() | |||
| return client.gather(future) | |||
|
|
|||
| @classmethod | |||
| def scatter(cls, data, **kwargs): | |||
There was a problem hiding this comment.
I prefer to call this put. It will be a little weird to say DaskTask.put or DaskTask.scatter because tasks aren't an object store.
There was a problem hiding this comment.
What about using DaskWrapper class name instead of DaskTask?
There was a problem hiding this comment.
Will change to put.
|
|
||
| Returns | ||
| ------- | ||
| ray.ObjectRef or list | ||
| Ray identifier of the result being put to Plasma store. | ||
| """ | ||
| return deploy_ray_func.options(num_returns=num_returns).remote(func, kwargs) | ||
| return deploy_ray_func.options(num_returns=num_returns).remote( | ||
| func, *args, kwargs |
There was a problem hiding this comment.
Historically we have not been able to pass **kwargs and *args to a Ray task for implementation reasons. To get around this we pass kwargs as a dictionary.
modin/core/execution/dask/implementations/pandas_on_dask/partitioning/partition.py
Outdated
Show resolved
Hide resolved
dchigarev
reviewed
Dec 20, 2021
modin/core/execution/dask/implementations/pandas_on_dask/partitioning/partition.py
Outdated
Show resolved
Hide resolved
dchigarev
reviewed
Dec 20, 2021
Will this make sense to create a common interface for every |
This is the main goal of the refactoring, but it will only be done in the third step :) (After updating |
There was a problem hiding this comment.
@anmyachev is there another PR that does something similar? or is this a different PR?
There is a separate pull request PR where I'm trying to bring the interface of ray and dask to the same look. The main remaining problem in that pull request is to synchronize the interface for working with actors. This pull request (3854) can be considered separately. |
YarShev
reviewed
Feb 2, 2022
| @@ -64,9 +62,9 @@ def get(self): | |||
| """ | |||
| self.drain_call_queue() | |||
| # blocking operation | |||
| if isinstance(self.future, pandas.DataFrame): | |||
| if not isinstance(self.future, Future): | |||
There was a problem hiding this comment.
Why do we need to change this?
There was a problem hiding this comment.
Thus, the code looks clearer, we materialize when the object can be materialized.
There was a problem hiding this comment.
A partition should hold a pandas DataFrame. We may come to undefined behavior or errors in this case.
There was a problem hiding this comment.
Some code is written as if there could also be an object of pandas.series type.
There was a problem hiding this comment.
Each partition holds a pandas DataFrame, but not a pandas Series. That is what we require and say in docs. So I am in favor of checking for a pandas DataFrame only, to be strongly restricted.
There was a problem hiding this comment.
- We should add a check for a future passed in the constructor to detect if it is really Dask.Future (as for Ray).
- I still tend to add a check in get if the materialized data is a pandas DataFrame.
- We should remove isinstance for a pandas Series in to_pandas.
There was a problem hiding this comment.
- Check is added.
- This is not possible, because after calling
applyfunction, the partition can contain an arbitrary object underFutureobject (More details in REFACTOR-#4206: add assert check into__init__method ofPandasOnDaskDataframePartitionclass #4207). - It is impossible according to 2 point.
There was a problem hiding this comment.
- I see, to_numpy is presumably the only case when we have a NumPy array in a partition. Okay, let's leave it as is but we should think of this.
- We can remove isinstance for a pandas Series in to_pandas since we don't add that check in get.
There was a problem hiding this comment.
- This should not be done, because after calling some function on the
DataFrameobject, an object ofSeriestype may be obtained, which will be stored in the partition (in that case,getfunction will returnSeries).
modin/core/execution/dask/implementations/pandas_on_dask/partitioning/partition.py
Outdated
Show resolved
Hide resolved
modin/core/execution/dask/implementations/pandas_on_dask/partitioning/partition.py
Outdated
Show resolved
Hide resolved
anmyachev
force-pushed
the
issue3853
branch
2 times, most recently
from
f2aec07
to
d004417
Compare
|
@YarShev ready for review |
YarShev
reviewed
Feb 9, 2022
|
Please fix conflicts. |
Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>
Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>
Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>
Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>
Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>
YarShev
reviewed
Mar 14, 2022
docs/flow/modin/core/execution/dask/implementations/pandas_on_dask/index.rst
Outdated
Show resolved
Hide resolved
modin/core/execution/dask/implementations/pandas_on_dask/dataframe/dataframe.py
Outdated
Show resolved
Hide resolved
modin/core/execution/dask/implementations/pandas_on_dask/io/io.py
Outdated
Show resolved
Hide resolved
modin/core/execution/dask/implementations/pandas_on_dask/dataframe/dataframe.py
Outdated
Show resolved
Hide resolved
Co-authored-by: Yaroslav Igoshev <Poolliver868@mail.ru>
Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>
|
@YarShev ready for review. I answered all the comments except about the filename. |
Co-authored-by: Yaroslav Igoshev <Poolliver868@mail.ru>
Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>
|
@YarShev ready for review |
YarShev
reviewed
Mar 17, 2022
docs/flow/modin/core/execution/dask/implementations/pandas_on_dask/index.rst
Outdated
Show resolved
Hide resolved
|
Also, do not forget to add a release note. |
Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>
|
@YarShev the error with |
YarShev
approved these changes
Mar 17, 2022
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Signed-off-by: Anatoly Myachev anatoly.myachev@intel.com
What do these changes do?
deployfunctions.client.submitfunction,DaskTask.deployfunction is called.client.materializeinstead ofclient.gatherclient.scatter(maybe rename toput) instead ofclient.scatterflake8 modin/ asv_bench/benchmarks scripts/doc_checker.pyblack --check modin/ asv_bench/benchmarks scripts/doc_checker.pygit commit -sdistributed.client.default_clientfunction into the single place #3853addedand passingdocs/developer/architecture.rstis up-to-date