Trying (and failing) at Finding a Way to run "Unconvential" Functions In Dask

[omarsumadi]

Hi Everyone,

Want to say thank you of course to everyone creating and maintain Dask.

Problem:

I'm trying to run some functions outside of the official Dask realm of support, so something that isn't Pandas, Joblib-based, or ML Integrations / Scikit. More specifically, I am trying to run functions that use Multiprocessing, but to hook them into the Dask System so that they can run on my Dask Cluster.

Working Code To Test the Problem / Provide Basis:

I'll provide an example using Scikit-learn of what I'm trying to do because it might be more familiar - This works perfect.

import dask
import time
import joblib
from dask_kubernetes import KubeCluster, make_pod_spec
from dask.distributed import Client
from sklearn.datasets import make_classification
from sklearn.ensemble import RandomForestClassifier

client = Client(threads_per_worker=1, n_workers=10, processes=True) # Force Processes and Single Threaded

def TestingMultiProcessing(X,y):
    return model.fit(X, y)

# Create a RFC with 5 Processes
X, y = make_classification(n_samples=10000, n_features=20, n_informative=15, n_redundant=5, random_state=3)
model = RandomForestClassifier(n_estimators=500, n_jobs=5)
with joblib.parallel_backend('dask'):
    model.fit(X, y)

Problem Explained

But what if something isn't integrated perfectly with Dask - what do we do - what if its just using regular Multiprocessing or some other wonky library?

I came across another using who had a similar problem in dask/distributed#2718. In this issue, Tensorflow was using multiprocessing and they were getting around that turning worker daemons off and using submits. In addition, https://github.com/saturncloud/dask-pytorch-ddp allows using Multiprocessing with Dask by turning daemons off. So I was thinking that maybe turning daemons to False in the config would allow Dask to takeover and run its workers inplace of whatever was being used, but I was wrong.

dask.config.set({'distributed.worker.daemon': False}) # Enable Daemons
x = client.submit(TestingMultiProcessing, X, y)
results = x.result()

I tried this, but get:

In addition, the function takes a very, very long time to complete versus using Joblib as the backend, and was only using 1 worker rather than 5.

Discussion:

Is there any way to allow for functions that aren't supported to run using ALL Dask Workers and not just one, although their implementation might use something like Multiprocessing instead (something not supported by Dask?).

Here's another example:
https://pastebin.com/grKBuLrW

Looking for Feedback - please check out the auxiliary example.