[Ray] task based shuffle for ray #3040
Merged
+759
−132
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
chaokunyang
changed the title
chaokunyang
force-pushed
the
ray_dag_based_simple_shuffle
branch
from
73f466d to
8427dd7
Compare
fyrestone
reviewed
May 17, 2022
chaokunyang
changed the title
chaokunyang
force-pushed
the
ray_dag_based_simple_shuffle
branch
from
37b50b3 to
fd866b2
Compare
chaokunyang
force-pushed
the
ray_dag_based_simple_shuffle
branch
from
ec143bb to
0c32530
Compare
…shuffle # Conflicts: # mars/core/operand/shuffle.py # mars/services/task/analyzer/analyzer.py # mars/services/task/execution/api.py # mars/tests/test_utils.py
2 tasks
chaokunyang
changed the title
chaokunyang
force-pushed
the
ray_dag_based_simple_shuffle
branch
from
0148219 to
7e0d239
Compare
chaokunyang
force-pushed
the
ray_dag_based_simple_shuffle
branch
from
85423b1 to
7f9e8e8
Compare
qinxuye
reviewed
Jun 21, 2022
| @@ -112,3 +128,11 @@ def iter_mapper_data( | |||
| ctx, input_id, mapper_id, pop=pop, skip_none=skip_none | |||
| ): | |||
| yield data | |||
|
|
|||
| def execute(self, ctx, op): | |||
| """The mapper stage must ensure all mapper blocks are inserted into ctx and no blocks | |||
There was a problem hiding this comment.
To clarify, no block missing for both key and index, or just for index?
There was a problem hiding this comment.
Just for fetch_by_index
fyrestone
reviewed
Jun 21, 2022
fyrestone
reviewed
Jun 21, 2022
|
Since this PR introduce |
chaokunyang
force-pushed
the
ray_dag_based_simple_shuffle
branch
from
28b4ad4 to
e5aa38b
Compare
fyrestone
reviewed
Jun 27, 2022
2 tasks
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.
What do these changes do?
This PR implements pull-based shuffle for ray task backend, which will be the basis for later push-based shuffle for ray. Main changes include:
ShuffleFetchTypewhich containsFETCH_BY_KEY/FETCH_BY_INDEX. IfFETCH_BY_INDEXis used,source_keys/source_idxes/source_mapperswill be skipped. In this way the meta and scheduling cost for supervisor will be much smaller, about 1000~10000x smaller for big shuffle. This will make mars able to schedule large-scale shuffle, even PB-scale data shuffle.FetchShuffle, and skipsource_keys/source_idxes/source_mapperswhenshuffle_fetch_typeisFETCH_BY_INDEX.FETCH_BY_INDEX.TensorReshape/TensorBincountPSRSAlign.data_keysfor subtask execution on rayShuffleManagerfor shuffle execution on rayBefore this PR, for a shuffle with 100000 partitions, every reducer subtask meta will took about 16M memory, which make mars to handle only less than 1000 partitions, i.e. less than 1 T data.
With this PR, every reducer subtask took only 2Kb memory, which make mars can handle 8000_000 partitions. If every partition has 1Gb data, this PR will enable mars handle 8PB data shuffle.
#2916
#3039
#2112
Check code requirements