Joblib Using Dask Backend 2x Slower + Not Able to Run using Processess = True (Error in Pastebin)

[omarsumadi]

Hi Everyone,

I have been working at looking at the various potential ways to using Dask to parallelize in a cluster a function using Joblib without Dask as the explicit Backend versus using 'Dask' as the Joblib backend as setting Dask as the Backend is what's used in the Docs, but I wanted to see what I could do.

I have been experimenting with Scikit-learn as a way to bring up this discussion without bringing up packages in which there isn't a clear example in the Docs (see: https://ml.dask.org/joblib.html).

• It's important to note that Scikit Learn's implementation of RandomForestClassifier is to prefer threading, but no where does it require threading in the specific functions we use (fit function).

Here's the client setup:

import time
from sklearn.datasets import make_classification
from sklearn.ensemble import RandomForestClassifier
import joblib
from joblib import parallel_backend
import dask
from dask.distributed import Client

dask.config.set({'distributed.worker.daemon': False})
client = Client(threads_per_worker=1, n_workers=5, processes=True)
# Client(threads_per_worker=1, n_workers=5, processes=False) for testing threading.

I have some implementations to show on the various approaches I've been taking:

Approach 1: Using Submit with Either Threading or Processes on Scikit-Learn

Average Time of ~3-4 Seconds for both Process = False or True versus ~3 Seconds with Scikit Defaults without Dask (so just running the function).

def TestingMultiProcessing(X,y):
    # Runs with threads by Default
    return model.fit(X, y)

def TestingMultiProcessingForceProcesses(X,y):
   # Force the usage of Multiprocessing
    with parallel_backend('multiprocessing', n_jobs=5):
        model.fit(X, y)

X, y = make_classification(n_samples=10000, n_features=20, n_informative=15, n_redundant=5, random_state=3)

# Base Case
model = RandomForestClassifier(n_estimators=500, n_jobs=5)
start = time.time()
model.fit(X, y)
end = time.time()
print("Total Time Using Regular Interpreter")
print(end - start)

# Dask Without Joblib Backend Explicitly Set
model = RandomForestClassifier(n_estimators=500, n_jobs=5)
start = time.time()
x = client.submit(TestingMultiProcessing, X, y)
results = x.result()
end = time.time() # ~3.5 seconds on average comp

# Dask with Joblib Backend Explicitly Set
model = RandomForestClassifier(n_estimators=500)
start = time.time()
x = client.submit(TestingMultiProcessingForceProcesses, X, y) # Forcing Multiprocessing
results = x.result()
end = time.time() # ~3.5 seconds on average comp

Approach 1: Using Joblib with Dask as the Backend (~6.5-7.5 Seconds + Only Usable with Threads)

For some reason, if you set processes = True to force Dask to use processes, this will not work and give you an event loop not started error. Make sure you restart your Kernal to test.
I wanted to stick with this implementation (trying to test if processes=True will work because this seems like the correct, native approach. Does anyone know how we can run these functions with processes = True some way or why this is failing for me for any particular reason?

dask.config.set({'distributed.worker.daemon': False}) # Optional
client = Client(threads_per_worker=1, n_workers=5, processes=False) # True doesn't work (Error)
start = time.time()

with joblib.parallel_backend(backend='dask', n_jobs=5):
    model.fit(X, y)    
end = time.time()
print(end - start) # ~6.5-7.5 seconds can reach 8 seconds.

# Showing that setting Processes to True Doesn't work (Make sure you restart your Kernal)
client = Client(threads_per_worker=1, n_workers=5, processes=True) # True doesn't work (Error)
start = time.time()
with joblib.parallel_backend(backend='dask', n_jobs=5):
    model.fit(X, y)    
end = time.time()
print(end - start) # Fails with Event Loop Not Started Error

Here's the error trace for starting the client with Processes = True:
https://pastebin.com/8yP1dstK

Thanks,
Omar