Dask requires consistent Python hashing
- Dominant language
- Python
- Stars
- 1.7k
- Forks
- 778
- Avg merge
- 2h 50m
- Merged PRs (30d)
- 3
Description
As per https://github.com/dask/dask/issues/6640, Dask breaks if Python hashing is inconsistent across workers. This appears to be a bug in Distributed Dask backend as well:
**hashing.py:**
```python
from dask.distributed import Client
from dask.bag import Bag
dsk = {("x", 0): (range, 5), ("x", 1): (range, 5), ("x", 2): (range, 5)}
b = Bag(dsk, "x", 3)
def iseven(x):
return x % 2 == 0
def test_bag_groupby_pure_hash():
# https://github.com/dask/dask/issues/6640
result = b.groupby(iseven).compute()
assert result == [(False, [1, 3] * 3), (True, [0, 2, 4] * 3)]
def test_bag_groupby_normal_hash():
# https://github.com/dask/dask/issues/6640
client = Client(n_workers=3)
result = b.groupby(lambda x: "even" if iseven(x) else "odd").compute()
assert len(result) == 2
assert ("odd", [1, 3] * 3) in result
assert ("even", [0, 2, 4] * 3) in result
def main():
client = Client(n_workers=3)
test_bag_groupby_normal_hash()
test_bag_groupby_pure_hash()
print("HORRAH")
if __name__ == '__main__':
import hashing
hashing.main()
```
When run:
```shell-session
$ python hashing.py
...
File "../dask/hashing.py", line 36, in
hashing.main()
File "/home/itamarst/Devel/dask/hashing.py", line 29, in main
test_bag_groupby_normal_hash()
File "/home/itamarst/Devel/dask/hashing.py", line 22, in test_bag_groupby_normal_hash
assert len(result) == 2
```
### Solving this
The solution for Dask (https://github.com/dask/dask/pull/6660) was to set PYTHONHASHSEED for worker processes. And you can do that similar solution for Distributed in some cases, e.g. `Client()`.
However, I'm pretty sure the `distributed-worker` CLI just runs the worker inline, it's not a subprocess, so by the time you're running Python code the hash seed has already been set, and you can't change it. It could e.g. set it and the fork()+exec() Python again, I suppose.
Contributor guide
Assessment
This issue has not been assessed yet.