在Dask并行运行两个机器学习培训

Question

我已经和Docker上的工人一起实现了Dask。我用Docker compose文件启动10名工人，如下所示：

 docker-compose up -d --scale worker=10

要运行两个模型的机器学习培训，我执行以下操作：

y1 = data1[label1]
X1 = data1[features1] 

y2 = data2[label2]
X2 = data2[features2] 

with joblib.parallel_backend('dask'):
        try:
            model1.fit(X1, y1)
            model2.fit(X2, y2)
        except Exception as e:
            logging.error('There's an error ' + str(e))

现在，我想同时进行两次训练。我可以使用工人1到5进行培训1，工人6到10进行培训2。但是如何告诉Dask distributed为一个任务使用一些工人，为另一个任务使用其他工人

Answer 1

这个问题的层次相对较高，但我将提供一些可能有用的建议

首先，您编写的代码在本地运行大部分内容。要并行执行ML培训，您需要：

1. 在群集上工作（本地或远程）
2. 将数据存储在Dask阵列或数据帧中
3. 使用dask.delayed任务

或

1. 使用client.submit()API

1。创建（本地）群集

从您的代码中，不清楚您是否实例化了客户机，因此可能只需再次检查您是否遵循the dask-ml docs instructions此处：

from dask.distributed import Client
import joblib

client = Client(processes=False)        # create local cluster
# import coiled                         # or connect to remote cluster
# client = Client(coiled.Cluster())     

with joblib.parallel_backend('dask'):
    # your scikit-learn code

但是，请注意，scitkit学习的Dask joblib后端对于扩展CPU限制的工作负载非常有用。要扩展到RAM绑定的工作负载（大于内存数据集），您将需要考虑使用一个^ {CD3> }并行估计器，如下面建议的。p>

2。在Dask阵列中存储数据

下面的最小代码示例将两个虚拟数据集设置为Dask数组，并实例化一个K-Means聚类算法

import dask_ml.datasets
import dask_ml.cluster
import matplotlib.pyplot as plt

# create dummy datasets
X, y = dask_ml.datasets.make_blobs(n_samples=10000000,
                                   chunks=1000000,
                                   random_state=0,
                                   centers=3)

X2, y2 = dask_ml.datasets.make_blobs(n_samples=10000000,
                                   chunks=1000000,
                                   random_state=3,
                                   centers=3)

# persist predictor sets to cluster memory
X = X.persist()
X2 = X2.persist()

# instantiate KM model
km = dask_ml.cluster.KMeans(n_clusters=3, init_max_iter=2, oversampling_factor=10)

3。与Dask并行的培训。延迟

下面的代码使用dask.delayedAPI并行运行培训。接下来是the best practices outlined in the Dask docs

from dask import delayed
import dask

X = delayed(X)
X2 = delayed(X2)

@delayed
def train(model, X):
    return model.fit(X)

# define task graphs (lazy evaluation, no computation triggered)
km1 = train(km, X)
km2 = train(km, X2)

# trigger computation and yield fitted models in parallel
km1, km2 = dask.compute(km1, km2)

4。与期货和client.submit

或者，您可以使用client.submit()API进行并行训练。这会立即返回指向正在进行的计算的未来，并最终返回存储的结果。在the docs here中阅读更多内容

根据你提出的问题，我认为你的首要任务是让训练并行进行。这不需要手动将任务分配给特定的工人；Dask负责为您安排工作时间并优化员工分布。如果您真的对手动将特定任务分配给特定的工作人员感兴趣，我建议您看看this SO answer