Joblib: Failures with nested joblib parallel and dask

[mfeurer]

I'm having the issue that something cancels scikit-learn jobs if parallel is nested together with the dask backend. (Simplified) code to reproduce:

from dask.distributed import Client
from joblib import parallel_backend

from sklearn.datasets import fetch_openml
from sklearn.model_selection import GridSearchCV
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.tree import DecisionTreeClassifier
from sklearn.preprocessing import StandardScaler


model = GridSearchCV(
    estimator=Pipeline(
        steps=[
            (
                'Preprocessing', ColumnTransformer(transformers=[
                    ('NumericalPreprocessing', StandardScaler(), [0, 1, 2, 3])
                ]),
            ),
            ('Estimator', DecisionTreeClassifier())
        ],
    ),
    param_grid={'Estimator__min_samples_split': list(range(2, 101))},
)

if __name__ == '__main__':
    X, y = fetch_openml(return_X_y=True, data_id=61)
    client = Client()
    with parallel_backend('dask'):
        model.fit(X, y)

Output:

/home/feurerm/miniconda/3-4.5.4/envs/openml/bin/python3 /home/feurerm/sync_dir/projects/openml/Study-14/PythonV2/bug.py
/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/model_selection/_split.py:1978: FutureWarning: The default value of cv will change from 3 to 5 in version 0.22. Specify it explicitly to silence this warning.
  warnings.warn(CV_WARNING, FutureWarning)
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f75a6bc62f0> of <Future: status: cancelled, key: _transform_one-batch-90ed01946f8a4adf8e253d7e6b47769e>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _transform_one-batch-90ed01946f8a4adf8e253d7e6b47769e
distributed.worker - WARNING -  Compute Failed
Function:  <joblib._dask.Batch object at 0x7f75a6bc0f60>
args:      (array([[5.1, 3.5, 1.4, 0.2],
       [4.9, 3. , 1.4, 0.2],
       [4.7, 3.2, 1.3, 0.2],
       [4.6, 3.1, 1.5, 0.2],
       [5. , 3.6, 1.4, 0.2],
       [5.4, 3.9, 1.7, 0.4],
       [4.6, 3.4, 1.4, 0.3],
       [5. , 3.4, 1.5, 0.2],
       [4.4, 2.9, 1.4, 0.2],
       [4.9, 3.1, 1.5, 0.1],
       [5.4, 3.7, 1.5, 0.2],
       [4.8, 3.4, 1.6, 0.2],
       [4.8, 3. , 1.4, 0.1],
       [4.3, 3. , 1.1, 0.1],
       [5.8, 4. , 1.2, 0.2],
       [5.7, 4.4, 1.5, 0.4],
       [5.4, 3.9, 1.3, 0.4],
       [5.1, 3.5, 1.4, 0.3],
       [5.7, 3.8, 1.7, 0.3],
       [5.1, 3.8, 1.5, 0.3],
       [5.4, 3.4, 1.7, 0.2],
       [5.1, 3.7, 1.5, 0.4],
       [4.6, 3.6, 1. , 0.2],
       [5.1, 3.3, 1.7, 0.5],
       [4.8, 3.4, 1.9, 0.2],
       [5. , 3. , 1.6, 0.2],
       [5. , 3.4, 1.6, 0.4],
       [5.2, 3.5, 1.5, 0.2],
       [5.2, 3.4, 1.4, 0.2],
       [4.7, 3.2, 1.6, 0.2],
       [4.8, 3.1, 1.6, 0.2],
       [5.4, 3.4, 1.5, 0.4],
       [5.2, 4.1, 1.5, 0.1],
       [5.5, 4.2, 1.4, 0.2],
       [4.9, 
kwargs:    {}
Exception: CancelledError('_transform_one-batch-90ed01946f8a4adf8e253d7e6b47769e')

Traceback (most recent call last):
  File "/home/feurerm/sync_dir/projects/openml/Study-14/PythonV2/bug.py", line 30, in <module>
    model.fit(X, y)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/model_selection/_search.py", line 688, in fit
    self._run_search(evaluate_candidates)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/model_selection/_search.py", line 1149, in _run_search
    evaluate_candidates(ParameterGrid(self.param_grid))
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/model_selection/_search.py", line 667, in evaluate_candidates
    cv.split(X, y, groups)))
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/parallel.py", line 1016, in __call__
    self.retrieve()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/parallel.py", line 910, in retrieve
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec16f2bf8> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-6e3e3af2d5904839af62c37e3c2a1747>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
TypeError: exceptions must derive from BaseException
    self._output.extend(job.get())
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 270, in get
    return ref().result()
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec1940d90> of <Future: status: cancelled, key: _fit_and_score-batch-74bc4528b20d42cbb9cb8ddff63be4b8>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-74bc4528b20d42cbb9cb8ddff63be4b8
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 222, in result
    raise exc.with_traceback(tb)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 106, in __call__
    results.append(func(*args, **kwargs))
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/model_selection/_validation.py", line 556, in _fit_and_score
    test_scores = _score(estimator, X_test, y_test, scorer, is_multimetric)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/model_selection/_validation.py", line 599, in _score
    return _multimetric_score(estimator, X_test, y_test, scorer)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/model_selection/_validation.py", line 629, in _multimetric_score
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec1709d90> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-d81e459278c6477c912644049c4b49c4>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-d81e459278c6477c912644049c4b49c4
    score = scorer(estimator, X_test, y_test)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/metrics/scorer.py", line 240, in _passthrough_scorer
    return estimator.score(*args, **kwargs)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/utils/metaestimators.py", line 116, in <lambda>
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec17d2048> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-a5b86ba5f9034ec7b1df639f505bde19>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-a5b86ba5f9034ec7b1df639f505bde19
    out = lambda *args, **kwargs: self.fn(obj, *args, **kwargs)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/pipeline.py", line 598, in score
    Xt = transform.transform(Xt)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/compose/_column_transformer.py", line 539, in transform
    Xs = self._fit_transform(X, None, _transform_one, fitted=True)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/sklearn/compose/_column_transformer.py", line 420, in _fit_transform
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec1a12488> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-65544b059a8541b783c642e956f21d39>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-65544b059a8541b783c642e956f21d39
    self._iter(fitted=fitted, replace_strings=True), 1))
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec16f2840> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-83f642a7b46a43bdb565ce09f961d6dd>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-83f642a7b46a43bdb565ce09f961d6dd
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/parallel.py", line 1016, in __call__
    self.retrieve()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/parallel.py", line 910, in retrieve
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec1706510> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-67329437248541408258b10b4bdad1b7>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-67329437248541408258b10b4bdad1b7
    self._output.extend(job.get())
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 270, in get
    distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec1633378> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-c98f8b79e32946e780f5bbdb204a2492>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-c98f8b79e32946e780f5bbdb204a2492
return ref().result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _transform_one-batch-90ed01946f8a4adf8e253d7e6b47769e
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec1898840> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-b1e2e00cca37484489aa60c6605a757b>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-b1e2e00cca37484489aa60c6605a757b
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec18987b8> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-2a90e3e4814e4806b8b6929103c7fc08>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-2a90e3e4814e4806b8b6929103c7fc08
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec184c158> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-32640bac340943f5a94d1d0b4c68ed68>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-32640bac340943f5a94d1d0b4c68ed68
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec17f3ae8> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-9dab314edb5f4e6fa8e455a7ef21c347>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-9dab314edb5f4e6fa8e455a7ef21c347
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec17d2a60> of <Future: status: cancelled, type: list, key: _fit_and_score-batch-23de936dd0c340beb2be4a1ab3681ec3>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-23de936dd0c340beb2be4a1ab3681ec3
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec17f3268> of <Future: status: cancelled, key: _fit_and_score-batch-33c922e812ec4b7d8931ec7cc8c195e4>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-33c922e812ec4b7d8931ec7cc8c195e4
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec185f488> of <Future: status: cancelled, key: _fit_and_score-batch-20ba616a25ae498c8bfae09df8cdee90>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-20ba616a25ae498c8bfae09df8cdee90
distributed.client - ERROR - Error in callback <function DaskDistributedBackend.apply_async.<locals>.callback_wrapper at 0x7f0ec1709488> of <Future: status: cancelled, key: _fit_and_score-batch-7e7e0d44916640fc95ed2d49c623f95d>:
Traceback (most recent call last):
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 287, in execute_callback
    fn(fut)
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
    result = future.result()
  File "/home/feurerm/miniconda/3-4.5.4/envs/openml/lib/python3.7/site-packages/distributed/client.py", line 224, in result
    raise result
concurrent.futures._base.CancelledError: _fit_and_score-batch-7e7e0d44916640fc95ed2d49c623f95d

Process finished with exit code 1

Some more notes:

• Unfortunately, I cannot reproduce the issue without scikit-learn
• Obviously, the column transformer would include multiple parallel pipelines. I removed them for simplicity
• I believe this issue is different to #957 as I use the default cluster (and not a SLURMCluster), and I use nested calls to joblib
• The code works with the loky, multiprocessing and threading backend.
• It's possible to fix this by monkey-patching the column transformer as shown below

# Sequential column transformer which replaces the call to Parallel by a list comprehension
class SequentialColumnTransformer(ColumnTransformer):

    def _fit_transform(self, X, y, func, fitted=False):

        transformers = list(self._iter(fitted=fitted, replace_strings=True))
        return [
            func(
                transformer=clone(trans) if not fitted else trans,
                X=_get_column(X, column),
                y=y,
                weight=weight,
                message_clsname='ColumnTransformer',
                message=self._log_message(name, idx, len(transformers)))
            for idx, (name, trans, column, weight)
            in enumerate(self._iter(fitted=fitted, replace_strings=True), 1)
        ]


# Use a different backend in the inner call to parallel
class LokyColumnTransformer(ColumnTransformer):
    def _fit_transform(self, X, y, func, fitted=False):
        with parallel_backend('loky'):
            return super()._fit_transform(X, y, func, fitted)

CC @amueller

[amueller]

Any chance @ogrisel or @lesteve can look into this? There's related issues in the sklearn tracker (https://github.com/scikit-learn/scikit-learn/issues/12315 https://github.com/scikit-learn/scikit-learn/issues/15383) and I thought dask interop was a priority ;)

[ogrisel]

This problem can be randomly by test_nested_parallelism_with_dask, e.g.:

https://ci.appveyor.com/project/joblib-ci/joblib/builds/29489976/job/rwy1jh1htu3cmev2

================================== FAILURES ===================================
______________________ test_nested_parallelism_with_dask ______________________
    @with_numpy
    @skipif(distributed is None, reason='This test requires dask')
    def test_nested_parallelism_with_dask():
        client = distributed.Client(n_workers=2, threads_per_worker=2)  # noqa

        # 10 MB of data as argument to trigger implicit scattering
        data = np.ones(int(1e7), dtype=np.uint8)
        for i in range(2):
            with parallel_backend('dask'):
>               backend_types_and_levels = _recursive_backend_info(data=data)
backend_types_and_levels = [('DaskDistributedBackend', 0), ('DaskDistributedBackend', 0), ('DaskDistributedBackend', 0), ('DaskDistributedBackend', 0)]
client     = <Client: 'tcp://127.0.0.1:1953' processes=2 threads=4, memory=5.24 GB>
data       = array([1, 1, 1, ..., 1, 1, 1], dtype=uint8)
i          = 1
joblib\test\test_parallel.py:1510: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
joblib\test\test_parallel.py:1474: in _recursive_backend_info
    for i in range(1))
joblib\parallel.py:1017: in __call__
    self.retrieve()
joblib\parallel.py:911: in retrieve
    self._output.extend(job.get())
joblib\_dask.py:270: in get
    return ref().result()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Future: cancelled, key: _recursive_backend_info-batch-bf029f0ecb4641beb240c5b10df1087d>
timeout = None
    def result(self, timeout=None):
        """ Wait until computation completes, gather result to local process.

        If *timeout* seconds are elapsed before returning, a
        ``dask.distributed.TimeoutError`` is raised.
        """
        if self.client.asynchronous:
            return self.client.sync(self._result, callback_timeout=timeout)

        # shorten error traceback
        result = self.client.sync(self._result, callback_timeout=timeout, raiseit=False)
        if self.status == "error":
            typ, exc, tb = result
            raise exc.with_traceback(tb)
        elif self.status == "cancelled":
>           raise result
E           concurrent.futures._base.CancelledError: _recursive_backend_info-batch-bf029f0ecb4641beb240c5b10df1087d
result     = CancelledError('_recursive_backend_info-batch-bf029f0ecb4641beb240c5b10df1087d')
self       = <Future: cancelled, key: _recursive_backend_info-batch-bf029f0ecb4641beb240c5b10df1087d>
timeout    = None

Observed on this PR: #975.

[pierreglaser]

Thanks I can reproduce. It looks like it's due to a bad interaction between scikit-learn's Pipeline/ColumnTransformer and dask. When I replace the pipeline by the estimator only, the script runs successfully. I'll investigate.

[ogrisel]

This is fixed by the combination of #1055 and #106.

This code still triggers the following low level warnings in the asyncio module of the standard library:

task: <Task pending name='Task-6045' coro=<DaskDistributedBackend._collect() running at /home/ogrisel/code/joblib/joblib/_dask.py:197> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7fae2aaf0b50>()]> cb=[IOLoop.add_future.<locals>.<lambda>() at /home/ogrisel/miniconda3/envs/pylatest/lib/python3.8/site-packages/tornado/ioloop.py:690]>

but it does not break anymore and the results look good.

[pierreglaser]

This code still triggers the following low level warnings in the asyncio module of the standard library:

I'm planning on fixing this in the near future.

[TomAugspurger]

That's great to hear, thanks for working on this Pierre!

On Wed, Jul 1, 2020 at 5:14 AM Pierre Glaser [e-mail]
wrote:

This code still triggers the following low level warnings in the asyncio
module of the standard library:

I'm planning on fixing this in the near future.

—
You are receiving this because you are subscribed to this thread.
Reply to this email directly, view it on GitHub
https://github.com/joblib/joblib/issues/959#issuecomment-652329836, or
unsubscribe
https://github.com/notifications/unsubscribe-auth/AAKAOITSBMJY3DKSFXGY3YDRZMEBVANCNFSM4JKJNT4A
.