Fix partitioning in explicit-comms shuffle #1356
Conversation
rjzamora
added
bug
| # Make sure partitions are properly ordered | ||
| futures = [_futures.pop(i) for i in range(npartitions)] |
There was a problem hiding this comment.
@ayushdg - FYI: I think this means the ordering of partitions could have been "wrong" even before the dtype-casting change in dask :/
There was a problem hiding this comment.
Ah I see. Based on the limited testing I've done the biggest change in results on my tests were from #1323 , even when paired with an older version of Dask prior to the dtype change.
| ddf = dd.from_pandas(df.copy(), npartitions=input_nparts).persist( | ||
| workers=all_workers | ||
| ) | ||
| ddf1 = dd.from_pandas(df.copy(), npartitions=input_nparts) |
There was a problem hiding this comment.
Strange Finding: Even without the changes in this PR, I get a "cancelled" error when I persist ddf and then perform both an explicit comms shuffle and then a task-based shuffle. I don't understand the cause of this yet.
There was a problem hiding this comment.
Are we ok still moving ahead with the changes even though there's a cancelation error?
There was a problem hiding this comment.
Yes. The changes in this PR are independent of the cancellation error (it happens in branch-24.08 without these changes).
Another important detail: I only get the error when query-planning is disabled. Therefore, I assume the problem has to do with a key-name collision that doesn't happen with dask-expr (which is much more disciplined about key names than the legacy API is).
| ddf, | ||
| ddf1, |
There was a problem hiding this comment.
Seems a bit confusing to me that we were previously modifying the initial collection.
There was a problem hiding this comment.
Makes sense to avoid that. I think @madsbk may have an idea if this was an oversight or was intentional when he gets back.
| ddf = dd.from_pandas(df.copy(), npartitions=input_nparts).persist( | ||
| workers=all_workers | ||
| ) | ||
| ddf1 = dd.from_pandas(df.copy(), npartitions=input_nparts) |
There was a problem hiding this comment.
Are we ok still moving ahead with the changes even though there's a cancelation error?
| ddf, | ||
| ddf1, |
There was a problem hiding this comment.
Makes sense to avoid that. I think @madsbk may have an idea if this was an oversight or was intentional when he gets back.
| # TODO: Use `partition_by_hash` if/when dtype-casting is added | ||
| # (See: https://github.com/rapidsai/dask-cuda/pull/1356) |
There was a problem hiding this comment.
The link refers to this PR, what is there exactly to see here for additional information on this TODO?
There was a problem hiding this comment.
Ah good catch! I was planning to file an issue to add a cast_dtype argument to partition_by_hash (similar to the argument added to partitioning_index in dask/dask#10705). I'll update the link before this gets merged.
rjzamora
added
5 - Ready to Merge
|
/merge |
Closes #1355
Current version of the explicit-comms shuffle does not produce partitioning that is consistent with
dask.dataframe.