使用dask作为任务调度来并行运行机器学习模型

Question

因此，基本上我想要并行运行ML Pipelines。我一直在使用scikit-learn，因此决定使用DaskGridSearchCV。

我有一个gridSearchCV = DaskGridSearchCV(pipeline, grid, scoring=evaluator)对象列表，并按顺序运行它们：

for gridSearchCV in list:
    gridSearchCV.fit(train_data, train_target)
    predicted = gridSearchCV.predict(test_data)

如果我有N个不同的GridSearch对象，我想尽可能多地利用所有可用资源。如果有资源可以同时并行运行2、3、4，...或N，我想这样做。

因此，我开始根据dask的文档尝试一些操作。首先，我尝试了一下dask.threaded，dask.multiprocessing但结果却变慢了，而且我不断：

/Library/Python/2.7/site-packages/sklearn/externals/joblib/parallel.py:540: UserWarning: Multiprocessing backed parallel loops cannot be nested below threads, setting n_jobs=1

这是代码片段：

def run_pipeline(self, gs, data):

    train_data, test_data, train_target, expected = train_test_split(data, target, test_size=0.25, random_state=33)

    model = gs.fit(train_data, train_target)
    predicted = gs.predict(test_data)


values = [delayed(run_pipeline)(gs, df) for gs in gs_list]
compute(*values, get=dask.threaded.get)

也许我走错路了，您对我有什么建议吗？

Answer 1

是的，但我有一个 GridSearch 对象列表，例如一个使用 DecisionTree，另一个使用 RandomForest。只要有资源，我就想并行运行它们。

如果这是您的目标，我会将它们全部合并到同一个网格中。Scikit-Learn Pipelines 支持跨步骤的网格搜索，这将允许您仅在单个对象中进行搜索GridSearchCV（有关 scikit-learn 文档中的示例，请参阅此处）。如果您只有一个估计器（而不是管道），则可以使用Pipeline单步作为代理。例如：

from sklearn.tree import DecisionTreeClassifier
from sklearn.ensemble import RandomForestClassifier
from sklearn.pipeline import Pipeline
import dask_searchcv as dcv

pipeline = Pipeline([('est', DecisionTreeClassifier())])

grid = [
    {'est': [DecisionTreeClassifier()],
     'max_features': ['sqrt', 'log2'],
     # more parameters for DecisionTreeClassifier
    },
    {'est': [RandomForestClassifier()],
     'max_features': ['sqrt', 'log2'],
     # more parameters for RandomForesetClassifier
    },
    # more estimator/parameter subsets
]

gs = dcv.GridSearchCV(pipeline, grid)
gs.fit(train_data, train_target)
gs.predict(test_data)

请注意，对于这种特定情况（所有估计器共享相同的参数，您可以合并网格：

grid = {'est': [DecisionTreeClassifier(), RandomForestClassifier()],
        'max_features': ['sqrt', 'log2'],
        # more parameters for all estimators}

至于为什么你的延迟示例不起作用 -是为了包装不dask.delayed调用 dask 代码的函数。由于您在延迟函数（也使用 dask 进行计算）内调用对象（使用 dask 进行计算），因此您嵌套了对 dask 调度程序的调用，这最多可能导致性能不佳和奇怪的错误最坏的情况是。fitdask_searchcv.GridSearchCV