如何将 sklearn 管道转换为 pyspark 管道？

Question

我们有一个机器学习分类器模型，我们使用 pandas 数据框和标准 sklearn 管道（StandardScaler、RandomForestClassifier、GridSearchCV 等）进行了训练。我们正在研究 Databricks，并希望使用并行计算 spark 提供的功能将此管道扩展到大型数据集。

将我们的 sklearn 管道转换为并行计算的东西的最快方法是什么？（我们可以根据需要轻松地在 pandas 和 spark DF 之间切换。）

对于上下文，我们的选择似乎是：

使用MLLib重写流水线（耗时）
使用 sklearn-spark 桥接库

在选项2上，Spark-Sklearn好像是deprecated, but Databricks instead recommends我们用的是joblibspark。但是，这会在 Databricks 上引发异常：

from sklearn import svm, datasets
from sklearn.model_selection import GridSearchCV
from joblibspark import register_spark
from sklearn.utils import parallel_backend
register_spark() # register spark backend

iris = datasets.load_iris()
parameters = {'kernel':('linear', 'rbf'), 'C':[1, 10]}
svr = svm.SVC(gamma='auto')

clf = GridSearchCV(svr, parameters, cv=5)
with parallel_backend('spark', n_jobs=3):
    clf.fit(iris.data, iris.target)

加注

py4j.security.Py4JSecurityException: Method public int org.apache.spark.SparkContext.maxNumConcurrentTasks() is not whitelisted on class class org.apache.spark.SparkContext

Answer 1

根据Databricks说明（here and here），必要的要求是：

Python 3.6+
pyspark>=2.4
scikit-learn>=0.21
joblib>=0.14

我无法在社区 Databricks 集群中重现 你的问题运行 Python 3.7.5，Spark 3.0.0，scikit-learn 0.22 .1 和作业库 0.14.1:

import sys
import sklearn
import joblib

spark.version
# '3.0.0'

sys.version
# '3.7.5 (default, Nov  7 2019, 10:50:52) \n[GCC 8.3.0]'

sklearn.__version__
# '0.22.1'

joblib.__version__
# '0.14.1'

通过以上设置，您的代码片段运行流畅，并且确实生成了一个分类器 clf 作为：

GridSearchCV(cv=5, error_score=nan,
             estimator=SVC(C=1.0, break_ties=False, cache_size=200,
                           class_weight=None, coef0=0.0,
                           decision_function_shape='ovr', degree=3,
                           gamma='auto', kernel='rbf', max_iter=-1,
                           probability=False, random_state=None, shrinking=True,
                           tol=0.001, verbose=False),
             iid='deprecated', n_jobs=None,
             param_grid={'C': [1, 10], 'kernel': ('linear', 'rbf')},
             pre_dispatch='2*n_jobs', refit=True, return_train_score=False,
             scoring=None, verbose=0)

来自 here 的替代示例也是如此：

from sklearn.utils import parallel_backend
from sklearn.model_selection import cross_val_score
from sklearn import datasets
from sklearn import svm
from joblibspark import register_spark

register_spark() # register spark backend

iris = datasets.load_iris()
clf = svm.SVC(kernel='linear', C=1)
with parallel_backend('spark', n_jobs=3):
  scores = cross_val_score(clf, iris.data, iris.target, cv=5)

print(scores)

给予

[0.96666667 1.         0.96666667 0.96666667 1.        ]

Answer 2

感谢 desertnaut 的回复 - 这个答案对于标准的 Spark / Databricks 设置应该是正确的，所以接受它，考虑到我的问题的措辞/对其他读者的潜在用处

贡献一个单独的“答案”发现了我们案例中的问题：Databricks 支持建议我们案例中的问题是由于我们使用了一种特殊类型的集群（在 AWS 上启用了凭据直通的高并发性). grid.fit() 没有被列入此类集群的白名单，Databricks 建议他们需要与他们的工程团队一起提出来将其列入白名单。

如何将 sklearn 管道转换为 pyspark 管道？

How to convert a sklearn pipeline into a pyspark pipeline?

python

scikit-learn

pyspark

apache-spark-ml

databricks