[Dask-on-Ray] Propagate Dask-on-Ray scheduler config to (rest of) cluster

[clarkzinzow]

When setting the Dask-on-Ray scheduler as the Dask scheduler via a global config within the driver

dask.config.set(scheduler=ray_dask_get)
# The following will use the Dask-on-Ray scheduler
df.compute()

this config won't be propagated to any workers. Therefore, if you do some Dask computation within a Ray task or actor, it won't use the Dask-on-Ray scheduler:

dask.config.set(scheduler=ray_dask_get)

@ray.remote
def foo(df):
    # The following will NOT use the Dask-on-Ray scheduler.
    df.compute()

This can be extremely counter-intuitive, since to users, setting a global config implies that the config is set everywhere, including on other workers in their cluster.

We should find a way to either automatically propagate the Dask-on-Ray scheduler config to downstream tasks and actors, or at least provide an API that users can manually invoke that will do said propagation.

• [Gold] Automatic config propagation.
• [Silver] API for config propagation that the user can use.
• [Bronze] Documentation on the need to set the global config within tasks and actors.

[clarkzinzow]

Somewhat related to #17927 and #16743, where we wish to propagate the Dask-on-Ray scheduler config to the Ray Client server.

[clarkzinzow]

Note that Dask supports environment variables and serialized configs, but neither will suffice since we need to give a callable as the scheduler.

[amogkam]

One option could be to have Dask on Ray be its own importable library that's just a wrapper around dask, but sets all the necessary configurations during import.