Speed up dask.bag.Bag.random_sample

[crusaderky]

Closes #10351

%timeit random_state_data_python(1, 0)
%timeit random_state_data_python(10, 0)
%timeit random_state_data_python(100, 0)
%timeit random_state_data_python(1000, 0)

Before:

265 µs ± 1.72 µs per loop (mean ± std. dev. of 7 runs, 1,000 loops each)
2.64 ms ± 16.8 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)
26.5 ms ± 146 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)
272 ms ± 1.06 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

After:

131 µs ± 1.47 µs per loop (mean ± std. dev. of 7 runs, 10,000 loops each)
234 µs ± 1.21 µs per loop (mean ± std. dev. of 7 runs, 1,000 loops each)
1.26 ms ± 9.68 µs per loop (mean ± std. dev. of 7 runs, 1,000 loops each)
20.4 ms ± 137 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

[crusaderky]

Annoyingly this doctest will fail if a user hasn't installed numpy. I don't think it's a big problem.

[erayaslan]

numpy (and dask) RandomState is frozen and wont be getting any updates. Cant we use the new Generator interface here?

And if all we want is a bunch of reproducible parallel streams, perhaps we can move _spawn_bitgens from da.random to dask.utils and use that. Bonus would be getting rid of all those 624s (Mersenne Twister detail) in the code.

Do let me know if I am not making any sense.

[erayaslan]

On second look, perhaps just using default_rng instead of RandomState in line 2557 is enough. 624 is too deep in the stack to handle.

Feel free to ignore. Sorry

[crusaderky]

Sorry for the wait. I replaced RandomState with default_rng. Do you think this is sufficient to merge this PR?

[erayaslan]

yes, good for merge afaic. thank you