dask-yarn does not work with joblib about dask-yarn HOT 3 CLOSED

In [1]: import joblib

In [2]: from dask.distributed import Client

In [3]: import dask_yarn

In [4]: cluster = dask_yarn.YarnCluster(environment='dask-yarn-py37.tar.gz', deploy_mode='local', worker_vcores=1, worker_memory='512 MiB')
19/01/07 15:20:38 INFO client.RMProxy: Connecting to ResourceManager at master.example.com/172.27.0.4:8032
19/01/07 15:20:38 INFO skein.Driver: Driver started, listening on 35025
19/01/07 15:20:39 INFO skein.Driver: Uploading application resources to hdfs://master.example.com:9000/user/testuser/.skein/application_1546627174314_0056
19/01/07 15:20:42 INFO hdfs.DFSClient: Created token for testuser: HDFS_DELEGATION_TOKEN owner=testuser@EXAMPLE.COM, renewer=yarn, realUser=, issueDate=1546874441993, maxDate=1547479241993, sequenceNumber=52, masterKeyId=4 on 172.27.0.4:9000
19/01/07 15:20:42 INFO security.TokenCache: Got dt for hdfs://master.example.com:9000; Kind: HDFS_DELEGATION_TOKEN, Service: 172.27.0.4:9000, Ident: (token for testuser: HDFS_DELEGATION_TOKEN owner=testuser@EXAMPLE.COM, renewer=yarn, realUser=, issueDate=1546874441993, maxDate=1547479241993, sequenceNumber=52, masterKeyId=4)
19/01/07 15:20:42 INFO skein.Driver: Submitting application...
19/01/07 15:20:42 INFO impl.YarnClientImpl: Submitted application application_1546627174314_0056

In [5]: client = Client(cluster)

In [6]: client.ncores()  # Empty because by default we start no workers
Out[6]: {}

In [7]: cluster.scale(2)  # Add some workers, note that this may take a few seconds before workers start

In [8]: client.ncores()  # After waiting for workers to start
Out[8]: {'tcp://172.27.0.3:40495': 1, 'tcp://172.27.0.3:45711': 1}

In [9]: with joblib.parallel_backend('dask'):
   ...:     results = joblib.Parallel(verbose=100)(joblib.delayed(lambda x: x**2)(x) for x in range(10))
   ...:
[Parallel(n_jobs=-1)]: Using backend DaskDistributedBackend with 2 concurrent workers.
[Parallel(n_jobs=-1)]: Done   1 tasks      | elapsed:    0.2s
[Parallel(n_jobs=-1)]: Done   2 tasks      | elapsed:    0.2s
[Parallel(n_jobs=-1)]: Done   3 tasks      | elapsed:    0.2s
[Parallel(n_jobs=-1)]: Done   4 tasks      | elapsed:    0.3s
[Parallel(n_jobs=-1)]: Done   5 tasks      | elapsed:    0.3s
[Parallel(n_jobs=-1)]: Done   6 tasks      | elapsed:    0.3s
[Parallel(n_jobs=-1)]: Done   7 tasks      | elapsed:    0.3s
[Parallel(n_jobs=-1)]: Done   8 out of  10 | elapsed:    0.3s remaining:    0.1s
[Parallel(n_jobs=-1)]: Done  10 out of  10 | elapsed:    0.3s remaining:    0.0s
[Parallel(n_jobs=-1)]: Done  10 out of  10 | elapsed:    0.4s finished

In [10]: results
Out[10]: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

In [11]: import dask

In [12]: joblib.__version__
Out[12]: '0.13.0'

In [13]: dask.__version__
Out[13]: '1.0.0'

In [14]: dask_yarn.__version__
Out[14]: '0.5.0'

In [1]: from dask.distributed import Client, LocalCluster

In [2]:

In [2]: cluster = LocalCluster(n_workers=0)

In [3]: client = Client(cluster)

In [4]: import joblib

In [5]: with joblib.parallel_backend('dask'):
   ...:     results = joblib.Parallel(verbose=100)(joblib.delayed(lambda x: x**2)(x) for x in range(10))
   ...:
[Parallel(n_jobs=-1)]: Using backend DaskDistributedBackend with 0 concurrent workers.
[Parallel(n_jobs=-1)]: Done   0 out of   0 | elapsed:    0.0s finished